Class: Thread::Queue

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

Overview

The Thread::Queue class implements multi-producer, multi-consumer queues. It is especially useful in threaded programming when information must be exchanged safely between multiple threads. The Thread::Queue class implements all the required locking semantics.

The class implements FIFO (first in, first out) type of queue. In a FIFO queue, the first tasks added are the first retrieved.

Example:

queue = Thread::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

consumer.join

Instance Method Summary collapse

Constructor Details

#initialize(*args) ⇒ Object

call-seq:

Thread::Queue.new -> empty_queue
Thread::Queue.new(enumerable) -> queue

Creates a new queue instance, optionally using the contents of an enumerable for its initial state.

Example:

	q = Thread::Queue.new
  #=> #<Thread::Queue:0x00007ff7501110d0>
  q.empty?
  #=> true

	q = Thread::Queue.new([1, 2, 3])
	#=> #<Thread::Queue:0x00007ff7500ec500>
  q.empty?
  #=> false
  q.pop
  #=> 1



927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
 
# File 'thread_sync.c', line 927

static VALUE
rb_queue_initialize(int argc, VALUE *argv, VALUE self)
{
    VALUE initial;
    struct rb_queue *q = queue_ptr(self);
    if ((argc = rb_scan_args(argc, argv, "01", &initial)) == 1) {
        initial = rb_to_array(initial);
    }
    RB_OBJ_WRITE(self, queue_list(q), ary_buf_new());
    ccan_list_head_init(queue_waitq(q));
    if (argc == 1) {
        rb_ary_concat(q->que, initial);
    }
    return self;
}

Instance Method Details

#clear ⇒ Object

Removes all objects from the queue.



1143
1144
1145
1146
1147
1148
1149
1150
 
# File 'thread_sync.c', line 1143

static VALUE
rb_queue_clear(VALUE self)
{
    struct rb_queue *q = queue_ptr(self);

    rb_ary_clear(check_array(self, q->que));
    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 raise a ClosedQueueError.

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

  • when empty? is true, deq(false) will not suspend the thread and will return nil. deq(true) will raise a ThreadError.

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

Example:

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



987
988
989
990
991
992
993
994
995
996
997
998
999
 
# File 'thread_sync.c', line 987

static VALUE
rb_queue_close(VALUE self)
{
    struct rb_queue *q = queue_ptr(self);

    if (!queue_closed_p(self)) {
        FL_SET(self, QUEUE_CLOSED);

        wakeup_all(queue_waitq(q));
    }

    return self;
}

#closed? ⇒ Boolean

Returns true if the queue is closed.

Returns:

  • (Boolean) 


1008
1009
1010
1011
1012
 
# File 'thread_sync.c', line 1008

static VALUE
rb_queue_closed_p(VALUE self)
{
    return RBOOL(queue_closed_p(self));
}

#empty? ⇒ Boolean

Returns true if the queue is empty.

Returns:

  • (Boolean) 


1131
1132
1133
1134
1135
 
# File 'thread_sync.c', line 1131

static VALUE
rb_queue_empty_p(VALUE self)
{
    return RBOOL(queue_length(self, queue_ptr(self)) == 0);
}

#freeze ⇒ Object

The queue can’t be frozen, so this method raises an exception:

Thread::Queue.new.freeze # Raises TypeError (cannot freeze #<Thread::Queue:0x...>)



1176
1177
1178
1179
1180
1181
 
# File 'thread_sync.c', line 1176

static VALUE
rb_queue_freeze(VALUE self)
{
    rb_raise(rb_eTypeError, "cannot freeze " "%+"PRIsVALUE, self);
    UNREACHABLE_RETURN(self);
}

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

Returns the length of the queue.



1161
1162
1163
1164
1165
 
# File 'thread_sync.c', line 1161

static VALUE
rb_queue_length(VALUE self)
{
    return LONG2NUM(queue_length(self, queue_ptr(self)));
}

#marshal_dump ⇒ Object

:nodoc:



1587
1588
1589
1590
1591
1592
 
# File 'thread_sync.c', line 1587

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

#num_waiting ⇒ Object

Returns the number of threads waiting on the queue.



1189
1190
1191
1192
1193
1194
1195
 
# File 'thread_sync.c', line 1189

static VALUE
rb_queue_num_waiting(VALUE self)
{
    struct rb_queue *q = queue_ptr(self);

    return INT2NUM(q->num_waiting);
}

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

Pushes the given object to the queue.



1024
1025
1026
1027
1028
 
# File 'thread_sync.c', line 1024

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