Convert Queue and SizedQueue to rb builtin

A large part of `thread_sync.c` was migrated already, might as well
go all the way. It also allow to remove a bunch of Rdoc commands.

1 files changed, 343 insertions, 1 deletions

diff --git a/thread_sync.rb b/thread_sync.rb
index a722b7ec1a..398c0d02b7 100644
--- a/thread_sync.rb
+++ b/thread_sync.rb
@@ -1,7 +1,62 @@
 # frozen_string_literal: true
 
 class Thread
+  #  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
   class Queue
+    # Document-method: Queue::new
+    #
+    # 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
+    def initialize(enumerable = nil)
+      Primitive.queue_initialize(enumerable)
+    end
+
     # call-seq:
     #   pop(non_block=false, timeout: nil)
     #
@@ -21,9 +76,129 @@ class Thread
     end
     alias_method :deq, :pop
     alias_method :shift, :pop
+
+    undef_method :initialize_copy
+
+    # call-seq:
+    #   push(object)
+    #   enq(object)
+    #   <<(object)
+    #
+    # Pushes the given +object+ to the queue.
+    def push(object)
+      Primitive.cexpr!('queue_do_push(self, queue_ptr(self), object)')
+    end
+    alias_method :enq, :push
+    alias_method :<<, :push
+
+    # call-seq:
+    #   close
+    #
+    # 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
+    def close
+      Primitive.cstmt! %{
+        if (!queue_closed_p(self)) {
+            FL_SET_RAW(self, QUEUE_CLOSED);
+
+            wakeup_all(&queue_ptr(self)->waitq);
+        }
+
+        return self;
+      }
+    end
+
+    # call-seq: closed?
+    #
+    # Returns +true+ if the queue is closed.
+    def closed?
+      Primitive.cexpr!('RBOOL(FL_TEST_RAW(self, QUEUE_CLOSED))')
+    end
+
+    # call-seq:
+    #   length
+    #   size
+    #
+    # Returns the length of the queue.
+    def length
+      Primitive.cexpr!('LONG2NUM(queue_ptr(self)->len)')
+    end
+    alias_method :size, :length
+
+    # call-seq: empty?
+    #
+    # Returns +true+ if the queue is empty.
+    def empty?
+      Primitive.cexpr!('RBOOL(queue_ptr(self)->len == 0)')
+    end
+
+    # Removes all objects from the queue.
+    def clear
+      Primitive.cstmt! %{
+        queue_clear(queue_ptr(self));
+        return self;
+      }
+    end
+
+    # call-seq:
+    #   num_waiting
+    #
+    # Returns the number of threads waiting on the queue.
+    def num_waiting
+      Primitive.cexpr!('INT2NUM(queue_ptr(self)->num_waiting)')
+    end
+
+    def marshal_dump # :nodoc:
+      raise TypeError, "can't dump #{self.class}"
+    end
+
+    # call-seq:
+    #   freeze
+    #
+    # The queue can't be frozen, so this method raises an exception:
+    #   Thread::Queue.new.freeze # Raises TypeError (cannot freeze #<Thread::Queue:0x...>)
+    def freeze
+      raise TypeError, "cannot freeze #{self}"
+    end
   end
 
-  class SizedQueue
+  # This class represents queues of specified size capacity.  The push operation
+  # may be blocked if the capacity is full.
+  #
+  # See Thread::Queue for an example of how a Thread::SizedQueue works.
+  class SizedQueue < Queue
+    # Document-method: SizedQueue::new
+    # call-seq: new(max)
+    #
+    # Creates a fixed-length queue with a maximum size of +max+.
+    def initialize(vmax)
+      Primitive.szqueue_initialize(vmax)
+    end
+
     # call-seq:
     #   pop(non_block=false, timeout: nil)
     #
@@ -66,8 +241,93 @@ class Thread
     end
     alias_method :enq, :push
     alias_method :<<, :push
+
+    # call-seq:
+    #   close
+    #
+    # Similar to Thread::Queue#close.
+    #
+    # The difference is behavior with waiting enqueuing threads.
+    #
+    # If there are waiting enqueuing threads, they are interrupted by
+    # raising ClosedQueueError('queue closed').
+    def close
+      Primitive.cstmt! %{
+        if (!queue_closed_p(self)) {
+            struct rb_szqueue *sq = szqueue_ptr(self);
+
+            FL_SET(self, QUEUE_CLOSED);
+            wakeup_all(szqueue_waitq(sq));
+            wakeup_all(szqueue_pushq(sq));
+        }
+        return self;
+      }
+    end
+
+    # Removes all objects from the queue.
+    def clear
+      Primitive.cstmt! %{
+        struct rb_szqueue *sq = szqueue_ptr(self);
+        queue_clear(&sq->q);
+        wakeup_all(szqueue_pushq(sq));
+        return self;
+      }
+    end
+
+    # Returns the number of threads waiting on the queue.
+    def num_waiting
+      Primitive.cstmt! %{
+        struct rb_szqueue *sq = szqueue_ptr(self);
+        return INT2NUM(sq->q.num_waiting + sq->num_waiting_push);
+      }
+    end
+
+    # Returns the maximum size of the queue.
+    def max
+      Primitive.cexpr!('LONG2NUM(szqueue_ptr(self)->max)')
+    end
+
+    # call-seq: max=(number)
+    #
+    # Sets the maximum size of the queue to the given +number+.
+    def max=(vmax)
+      Primitive.cstmt! %{
+        long max = NUM2LONG(vmax);
+        if (max <= 0) {
+            rb_raise(rb_eArgError, "queue size must be positive");
+        }
+
+        long diff = 0;
+        struct rb_szqueue *sq = szqueue_ptr(self);
+
+        if (max > sq->max) {
+            diff = max - sq->max;
+        }
+        sq->max = max;
+        sync_wakeup(szqueue_pushq(sq), diff);
+        return vmax;
+      }
+    end
   end
 
+  #  Thread::Mutex implements a simple semaphore that can be used to
+  #  coordinate access to shared data from multiple concurrent threads.
+  #
+  #  Example:
+  #
+  #    semaphore = Thread::Mutex.new
+  #
+  #    a = Thread.new {
+  #      semaphore.synchronize {
+  #        # access shared resource
+  #      }
+  #    }
+  #
+  #    b = Thread.new {
+  #      semaphore.synchronize {
+  #        # access shared resource
+  #      }
+  #    }
   class Mutex
     # call-seq:
     #    Thread::Mutex.new   -> mutex
@@ -149,6 +409,83 @@ class Thread
     end
   end
 
+  #  ConditionVariable objects augment class Mutex. Using condition variables,
+  #  it is possible to suspend while in the middle of a critical section until a
+  #  condition is met, such as a resource becomes available.
+  #
+  #  Due to non-deterministic scheduling and spurious wake-ups, users of
+  #  condition variables should always use a separate boolean predicate (such as
+  #  reading from a boolean variable) to check if the condition is actually met
+  #  before starting to wait, and should wait in a loop, re-checking the
+  #  condition every time the ConditionVariable is waken up.  The idiomatic way
+  #  of using condition variables is calling the +wait+ method in an +until+
+  #  loop with the predicate as the loop condition.
+  #
+  #    condvar.wait(mutex) until condition_is_met
+  #
+  #  In the example below, we use the boolean variable +resource_available+
+  #  (which is protected by +mutex+) to indicate the availability of the
+  #  resource, and use +condvar+ to wait for that variable to become true.  Note
+  #  that:
+  #
+  #  1.  Thread +b+ may be scheduled before thread +a1+ and +a2+, and may run so
+  #      fast that it have already made the resource available before either
+  #      +a1+ or +a2+ starts. Therefore, +a1+ and +a2+ should check if
+  #      +resource_available+ is already true before starting to wait.
+  #  2.  The +wait+ method may spuriously wake up without signalling. Therefore,
+  #      thread +a1+ and +a2+ should recheck +resource_available+ after the
+  #      +wait+ method returns, and go back to wait if the condition is not
+  #      actually met.
+  #  3.  It is possible that thread +a2+ starts right after thread +a1+ is waken
+  #      up by +b+.  Thread +a2+ may have acquired the +mutex+ and consumed the
+  #      resource before thread +a1+ acquires the +mutex+.  This necessitates
+  #      rechecking after +wait+, too.
+  #
+  #  Example:
+  #
+  #    mutex = Thread::Mutex.new
+  #
+  #    resource_available = false
+  #    condvar = Thread::ConditionVariable.new
+  #
+  #    a1 = Thread.new {
+  #      # Thread 'a1' waits for the resource to become available and consumes
+  #      # the resource.
+  #      mutex.synchronize {
+  #        condvar.wait(mutex) until resource_available
+  #        # After the loop, 'resource_available' is guaranteed to be true.
+  #
+  #        resource_available = false
+  #        puts "a1 consumed the resource"
+  #      }
+  #    }
+  #
+  #    a2 = Thread.new {
+  #      # Thread 'a2' behaves like 'a1'.
+  #      mutex.synchronize {
+  #        condvar.wait(mutex) until resource_available
+  #        resource_available = false
+  #        puts "a2 consumed the resource"
+  #      }
+  #    }
+  #
+  #    b = Thread.new {
+  #      # Thread 'b' periodically makes the resource available.
+  #      loop {
+  #        mutex.synchronize {
+  #          resource_available = true
+  #
+  #          # Notify one waiting thread if any.  It is possible that neither
+  #          # 'a1' nor 'a2 is waiting on 'condvar' at this moment.  That's OK.
+  #          condvar.signal
+  #        }
+  #        sleep 1
+  #      }
+  #    }
+  #
+  #    # Eventually both 'a1' and 'a2' will have their resources, albeit in an
+  #    # unspecified order.
+  #    [a1, a2].each {|th| th.join}
   class ConditionVariable
     # Document-method: ConditionVariable::new
     #
@@ -193,3 +530,8 @@ class Thread
     end
   end
 end
+
+Mutex = Thread::Mutex
+ConditionVariable = Thread::ConditionVariable
+Queue = Thread::Queue
+SizedQueue = Thread::SizedQueue