Class: Queue

Inherits:
Object show all
Defined in:
thread_sync.c,
thread_sync.c

Overview

This class provides a way to synchronize communication between threads.

Example:

require ‘thread’

	queue = Queue.new

producer = Thread.new do

5.times do |i|
   sleep rand(i) # simulate expense
   queue << i
   puts "#{i} produced"
end

end

consumer = Thread.new do

5.times do |i|
   value = queue.pop
   sleep rand(i/2) # simulate expense
   puts "consumed #{value}"
end

end

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Object

Creates a new queue instance.



666
667
668
669
670
671
672
 
# File 'thread_sync.c', line 666

static VALUE
rb_queue_initialize(VALUE self)
{
    RSTRUCT_SET(self, QUEUE_QUE, ary_buf_new());
    RSTRUCT_SET(self, QUEUE_WAITERS, ary_buf_new());
    return self;
}

Instance Method Details

#clear ⇒ Object

Removes all objects from the queue.



847
848
849
850
851
852
 
# File 'thread_sync.c', line 847

static VALUE
rb_queue_clear(VALUE self)
{
    rb_ary_clear(GET_QUEUE_QUE(self));
    return self;
}

#close ⇒ Object

Closes the queue. A closed queue cannot be re-opened.

After the call to close completes, the following are true:

  • closed? will return true

  • close will be ignored.

  • calling enq/push/<< will return nil.

  • when empty? is false, calling deq/pop/shift will return an object from the queue as usual.

ClosedQueueError is inherited from StopIteration, so that you can break loop block.

Example:

  	q = Queue.new
    Thread.new{
      while e = q.deq # wait for nil to break loop
        # ...
      end
    }
    q.close



716
717
718
719
720
 
# File 'thread_sync.c', line 716

static VALUE
rb_queue_close(VALUE self)
{
    return queue_do_close(self, FALSE);
}

#closed? ⇒ Boolean

Returns true if the queue is closed.

Returns:

  • (Boolean) 


729
730
731
732
733
 
# File 'thread_sync.c', line 729

static VALUE
rb_queue_closed_p(VALUE self)
{
    return queue_closed_p(self) ? Qtrue : Qfalse;
}

#empty? ⇒ Boolean

Returns true if the queue is empty.

Returns:

  • (Boolean) 


835
836
837
838
839
 
# File 'thread_sync.c', line 835

static VALUE
rb_queue_empty_p(VALUE self)
{
    return queue_length(self) == 0 ? Qtrue : Qfalse;
}

#length ⇒ Object #size ⇒ Object Also known as: size

Returns the length of the queue.



863
864
865
866
867
868
 
# File 'thread_sync.c', line 863

static VALUE
rb_queue_length(VALUE self)
{
    unsigned long len = queue_length(self);
    return ULONG2NUM(len);
}

#marshal_dump ⇒ Object

:nodoc:



1213
1214
1215
1216
1217
1218
 
# File 'thread_sync.c', line 1213

static VALUE
undumpable(VALUE obj)
{
    rb_raise(rb_eTypeError, "can't dump %"PRIsVALUE, rb_obj_class(obj));
    UNREACHABLE;
}

#num_waiting ⇒ Object

Returns the number of threads waiting on the queue.



876
877
878
879
880
881
 
# File 'thread_sync.c', line 876

static VALUE
rb_queue_num_waiting(VALUE self)
{
    unsigned long len = queue_num_waiting(self);
    return ULONG2NUM(len);
}

#pop(non_block = false) ⇒ Object #deq(non_block = false) ⇒ Object #shift(non_block = false) ⇒ Object Also known as: deq, shift

Retrieves data from the queue.

If the queue is empty, the calling thread is suspended until data is pushed onto the queue. If non_block is true, the thread isn’t suspended, and an exception is raised.



821
822
823
824
825
826
 
# File 'thread_sync.c', line 821

static VALUE
rb_queue_pop(int argc, VALUE *argv, VALUE self)
{
    int should_block = queue_pop_should_block(argc, argv);
    return queue_do_pop(self, should_block);
}

#push(object) ⇒ Object #enq(object) ⇒ Object #<<(object) ⇒ Object Also known as: enq, <<

Pushes the given object to the queue.



745
746
747
748
749
 
# File 'thread_sync.c', line 745

static VALUE
rb_queue_push(VALUE self, VALUE obj)
{
    return queue_do_push(self, obj);
}