ruby/thread_sync.rb

# 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)
    #
    # 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
    # +ThreadError+ is raised.
    #
    # If +timeout+ seconds have passed and no data is available +nil+ is
    # returned. If +timeout+ is +0+ it returns immediately.
    def pop(non_block = false, timeout: nil)
      if non_block && timeout
        raise ArgumentError, "can't set a timeout if non_block is enabled"
      end
      Primitive.rb_queue_pop(non_block, timeout)
    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!('LEN2NUM(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

  # 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)
    #
    # 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 +ThreadError+ is raised.
    #
    # If +timeout+ seconds have passed and no data is available +nil+ is
    # returned. If +timeout+ is +0+ it returns immediately.
    def pop(non_block = false, timeout: nil)
      if non_block && timeout
        raise ArgumentError, "can't set a timeout if non_block is enabled"
      end
      Primitive.rb_szqueue_pop(non_block, timeout)
    end
    alias_method :deq, :pop
    alias_method :shift, :pop

    # call-seq:
    #   push(object, non_block=false, timeout: nil)
    #   enq(object, non_block=false, timeout: nil)
    #   <<(object)
    #
    # Pushes +object+ to the queue.
    #
    # If there is no space left in the queue, waits until space becomes
    # available, unless +non_block+ is true.  If +non_block+ is true, the
    # thread isn't suspended, and +ThreadError+ is raised.
    #
    # If +timeout+ seconds have passed and no space is available +nil+ is
    # returned. If +timeout+ is +0+ it returns immediately.
    # Otherwise it returns +self+.
    def push(object, non_block = false, timeout: nil)
      if non_block && timeout
        raise ArgumentError, "can't set a timeout if non_block is enabled"
      end
      Primitive.rb_szqueue_push(object, non_block, timeout)
    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!('LEN2NUM(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! %{
        rb_len_t max = NUM2LEN(vmax);
        if (max <= 0) {
            rb_raise(rb_eArgError, "queue size must be positive");
        }

        rb_len_t 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
    #
    # Creates a new Mutex
    def initialize
    end

    # call-seq:
    #    mutex.locked?  -> true or false
    #
    # Returns +true+ if this lock is currently held by some thread.
    def locked?
      Primitive.cexpr! %q{ RBOOL(mutex_locked_p(mutex_ptr(self))) }
    end

    # call-seq:
    #    mutex.owned?  -> true or false
    #
    # Returns +true+ if this lock is currently held by current thread.
    def owned?
      Primitive.rb_mut_owned_p
    end

    # call-seq:
    #    mutex.lock  -> self
    #
    # Attempts to grab the lock and waits if it isn't available.
    # Raises +ThreadError+ if +mutex+ was locked by the current thread.
    def lock
      Primitive.rb_mut_lock
    end

    # call-seq:
    #    mutex.try_lock  -> true or false
    #
    # Attempts to obtain the lock and returns immediately. Returns +true+ if the
    # lock was granted.
    def try_lock
      Primitive.rb_mut_trylock
    end

    # call-seq:
    #    mutex.unlock  -> self
    #
    # Releases the lock.
    # Raises +ThreadError+ if +mutex+ wasn't locked by the current thread.
    def unlock
      Primitive.rb_mut_unlock
    end

    # call-seq:
    #    mutex.synchronize { ... }    -> result of the block
    #
    # Obtains a lock, runs the block, and releases the lock when the block
    # completes.  See the example under Thread::Mutex.
    def synchronize
      raise ThreadError, "must be called with a block" unless defined?(yield)

      Primitive.rb_mut_synchronize
    end

    # call-seq:
    #    mutex.sleep(timeout = nil)    -> number or nil
    #
    # Releases the lock and sleeps +timeout+ seconds if it is given and
    # non-nil or forever.  Raises +ThreadError+ if +mutex+ wasn't locked by
    # the current thread.
    #
    # When the thread is next woken up, it will attempt to reacquire
    # the lock.
    #
    # Note that this method can wakeup without explicit Thread#wakeup call.
    # For example, receiving signal and so on.
    #
    # Returns the slept time in seconds if woken up, or +nil+ if timed out.
    def sleep(timeout = nil)
      Primitive.rb_mut_sleep(timeout)
    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
    #
    # Creates a new condition variable instance.
    def initialize
    end

    undef_method :initialize_copy

    # :nodoc:
    def marshal_dump
      raise TypeError, "can't dump #{self.class}"
    end

    # Document-method: Thread::ConditionVariable#signal
    #
    # Wakes up the first thread in line waiting for this lock.
    def signal
      Primitive.rb_condvar_signal
    end

    # Document-method: Thread::ConditionVariable#broadcast
    #
    # Wakes up all threads waiting for this lock.
    def broadcast
      Primitive.rb_condvar_broadcast
    end

    # Document-method: Thread::ConditionVariable#wait
    # call-seq: wait(mutex, timeout=nil)
    #
    # Releases the lock held in +mutex+ and waits; reacquires the lock on wakeup.
    #
    # If +timeout+ is given, this method returns after +timeout+ seconds passed,
    # even if no other thread doesn't signal.
    #
    # This method may wake up spuriously due to underlying implementation details.
    #
    # Returns the slept result on +mutex+.
    def wait(mutex, timeout=nil)
      Primitive.rb_condvar_wait(mutex, timeout)
    end
  end

  # Use the Monitor class when you want to have a lock object for blocks with
  # mutual exclusion.
  #
  #   lock = Monitor.new
  #   lock.synchronize do
  #     # exclusive access
  #   end
  #
  # Contrary to Mutex, Monitor is reentrant:
  #
  #   lock = Monitor.new
  #   lock.synchronize do
  #     lock.synchronize do
  #       # exclusive access
  #     end
  #   end
  class Monitor
    # call-seq:
    #   synchronize { } -> result of the block
    #
    # Enters exclusive section and executes the block.  Leaves the exclusive
    # section automatically when the block exits.  See example under
    # +MonitorMixin+.
    def synchronize(&)
      Primitive.rb_monitor_synchronize
    end

    # call-seq:
    #   try_enter -> true or false
    #
    # Attempts to enter exclusive section.  Returns +false+ if lock fails.
    def try_enter
      Primitive.rb_monitor_try_enter
    end

    # call-seq:
    #   enter -> nil
    #
    # Enters exclusive section.
    def enter
      Primitive.rb_monitor_enter
    end

    # call-seq:
    #   exit -> nil
    #
    # Leaves exclusive section.
    def exit
      Primitive.rb_monitor_exit
    end

    # internal methods for MonitorMixin
    def mon_check_owner # :nodoc:
      Primitive.rb_monitor_check_owner
    end

    def mon_locked? # :nodoc:
      Primitive.rb_monitor_locked_p
    end

    def mon_owned? # :nodoc:
      Primitive.rb_monitor_owned_p
    end

    # internal methods for MonitorMixin::ConditionVariable
    def wait_for_cond(cond, timeout) # :nodoc:
      Primitive.rb_monitor_wait_for_cond(cond, timeout)
    end

    # Creates a new Monitor::ConditionVariable associated with the
    # Monitor object.
    #
    def new_cond
      ConditionVariable.new(self)
    end

    # Condition variables, allow to suspend the current thread while in
    # the middle of a critical section until a condition is met, such as
    # a resource being available.
    #
    #  Example:
    #
    #    monitor = Thread::Monitor.new
    #
    #    resource_available = false
    #    condvar = monitor.new_cond
    #
    #    a1 = Thread.new {
    #      # Thread 'a1' waits for the resource to become available and consumes
    #      # the resource.
    #      monitor.synchronize {
    #        condvar.wait_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'.
    #      monitor.synchronize {
    #        condvar.wait_until { resource_available }
    #        resource_available = false
    #        puts "a2 consumed the resource"
    #      }
    #    }
    #
    #    b = Thread.new {
    #      # Thread 'b' periodically makes the resource available.
    #      loop {
    #        monitor.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
      def initialize(monitor) # :nodoc:
        @monitor = monitor
        @cond = Thread::ConditionVariable.new
      end

      # Releases the lock held in the associated monitor and waits; reacquires the lock on wakeup.
      #
      # If +timeout+ is given, this method returns after +timeout+ seconds passed,
      # even if no other thread doesn't signal.
      #
      def wait(timeout = nil)
        @monitor.mon_check_owner
        @monitor.wait_for_cond(@cond, timeout)
      end

      #
      # Calls wait repeatedly while the given block yields a truthy value.
      #
      def wait_while
        while yield
          wait
        end
      end

      #
      # Calls wait repeatedly until the given block yields a truthy value.
      #
      def wait_until
        until yield
          wait
        end
      end

      #
      # Wakes up the first thread in line waiting for this lock.
      #
      def signal
        @monitor.mon_check_owner
        @cond.signal
      end

      #
      # Wakes up all threads waiting for this lock.
      #
      def broadcast
        @monitor.mon_check_owner
        @cond.broadcast
      end
    end
  end
end

Mutex = Thread::Mutex # :nodoc:
Monitor = Thread::Monitor # :nodoc:
ConditionVariable = Thread::ConditionVariable # :nodoc:
Queue = Thread::Queue # :nodoc:
SizedQueue = Thread::SizedQueue # :nodoc: