Gecko [ lorencats.com ]

Name	Size	Permission
actor	[ DIR ]	drwxr-xr-x
atomic	[ DIR ]	drwxr-xr-x
atomic_reference	[ DIR ]	drwxr-xr-x
channel	[ DIR ]	drwxr-xr-x
collection	[ DIR ]	drwxr-xr-x
concern	[ DIR ]	drwxr-xr-x
edge	[ DIR ]	drwxr-xr-x
executor	[ DIR ]	drwxr-xr-x
synchronization	[ DIR ]	drwxr-xr-x
thread_safe	[ DIR ]	drwxr-xr-x
utility	[ DIR ]	drwxr-xr-x
actor.rb	3.53 KB	-rw-r--r--
agent.rb	21.58 KB	-rw-r--r--
array.rb	1.05 KB	-rw-r--r--
async.rb	17.7 KB	-rw-r--r--
atom.rb	8.85 KB	-rw-r--r--
atomics.rb	1.94 KB	-rw-r--r--
channel.rb	6.49 KB	-rw-r--r--
configuration.rb	6.57 KB	-rw-r--r--
constants.rb	204 B	-rw-r--r--
dataflow.rb	2.15 KB	-rw-r--r--
delay.rb	6.73 KB	-rw-r--r--
edge.rb	1.26 KB	-rw-r--r--
errors.rb	2.16 KB	-rw-r--r--
exchanger.rb	12.81 KB	-rw-r--r--
executors.rb	1.02 KB	-rw-r--r--
future.rb	4.12 KB	-rw-r--r--
hash.rb	999 B	-rw-r--r--
immutable_struct.rb	2.05 KB	-rw-r--r--
ivar.rb	7.24 KB	-rw-r--r--
lazy_register.rb	2.28 KB	-rw-r--r--
map.rb	7.72 KB	-rw-r--r--
maybe.rb	8.01 KB	-rw-r--r--
mutable_struct.rb	8.43 KB	-rw-r--r--
mvar.rb	7.19 KB	-rw-r--r--
options.rb	1.25 KB	-rw-r--r--
promise.rb	18.53 KB	-rw-r--r--
scheduled_task.rb	10.73 KB	-rw-r--r--
settable_struct.rb	3.67 KB	-rw-r--r--
synchronization.rb	1.05 KB	-rw-r--r--
timer_task.rb	11.74 KB	-rw-r--r--
tuple.rb	2.98 KB	-rw-r--r--
tvar.rb	5.91 KB	-rw-r--r--
version.rb	72 B	-rw-r--r--

Code Editor : mvar.rb

require 'concurrent/concern/dereferenceable'
require 'concurrent/synchronization'

module Concurrent

# An `MVar` is a synchronized single element container. They are empty or
  # contain one item. Taking a value from an empty `MVar` blocks, as does
  # putting a value into a full one. You can either think of them as blocking
  # queue of length one, or a special kind of mutable variable.
  #
  # On top of the fundamental `#put` and `#take` operations, we also provide a
  # `#mutate` that is atomic with respect to operations on the same instance.
  # These operations all support timeouts.
  #
  # We also support non-blocking operations `#try_put!` and `#try_take!`, a
  # `#set!` that ignores existing values, a `#value` that returns the value
  # without removing it or returns `MVar::EMPTY`, and a `#modify!` that yields
  # `MVar::EMPTY` if the `MVar` is empty and can be used to set `MVar::EMPTY`.
  # You shouldn't use these operations in the first instance.
  #
  # `MVar` is a [Dereferenceable](Dereferenceable).
  #
  # `MVar` is related to M-structures in Id, `MVar` in Haskell and `SyncVar` in Scala.
  #
  # Note that unlike the original Haskell paper, our `#take` is blocking. This is how
  # Haskell and Scala do it today.
  #
  # @!macro copy_options
  #
  # ## See Also
  #
  # 1. P. Barth, R. Nikhil, and Arvind. [M-Structures: Extending a parallel, non- strict, functional language with state](http://dl.acm.org/citation.cfm?id=652538). In Proceedings of the 5th
  #    ACM Conference on Functional Programming Languages and Computer Architecture (FPCA), 1991.
  #
  # 2. S. Peyton Jones, A. Gordon, and S. Finne. [Concurrent Haskell](http://dl.acm.org/citation.cfm?id=237794).
  #    In Proceedings of the 23rd Symposium on Principles of Programming Languages
  #    (PoPL), 1996.
  class MVar < Synchronization::Object
    include Concern::Dereferenceable
    safe_initialization!

# Unique value that represents that an `MVar` was empty
    EMPTY = Object.new

# Unique value that represents that an `MVar` timed out before it was able
    # to produce a value.
    TIMEOUT = Object.new

# Create a new `MVar`, either empty or with an initial value.
    #
    # @param [Hash] opts the options controlling how the future will be processed
    #
    # @!macro deref_options
    def initialize(value = EMPTY, opts = {})
      @value = value
      @mutex = Mutex.new
      @empty_condition = ConditionVariable.new
      @full_condition = ConditionVariable.new
      set_deref_options(opts)
    end

# Remove the value from an `MVar`, leaving it empty, and blocking if there
    # isn't a value. A timeout can be set to limit the time spent blocked, in
    # which case it returns `TIMEOUT` if the time is exceeded.
    # @return [Object] the value that was taken, or `TIMEOUT`
    def take(timeout = nil)
      @mutex.synchronize do
        wait_for_full(timeout)

# If we timed out we'll still be empty
        if unlocked_full?
          value = @value
          @value = EMPTY
          @empty_condition.signal
          apply_deref_options(value)
        else
          TIMEOUT
        end
      end
    end

# acquires lock on the from an `MVAR`, yields the value to provided block,
    # and release lock. A timeout can be set to limit the time spent blocked,
    # in which case it returns `TIMEOUT` if the time is exceeded.
    # @return [Object] the value returned by the block, or `TIMEOUT`
    def borrow(timeout = nil)
      @mutex.synchronize do
        wait_for_full(timeout)

# if we timeoud out we'll still be empty
        if unlocked_full?
          yield @value
        else
          TIMEOUT
        end
      end
    end

# Put a value into an `MVar`, blocking if there is already a value until
    # it is empty. A timeout can be set to limit the time spent blocked, in
    # which case it returns `TIMEOUT` if the time is exceeded.
    # @return [Object] the value that was put, or `TIMEOUT`
    def put(value, timeout = nil)
      @mutex.synchronize do
        wait_for_empty(timeout)

# If we timed out we won't be empty
        if unlocked_empty?
          @value = value
          @full_condition.signal
          apply_deref_options(value)
        else
          TIMEOUT
        end
      end
    end

# Atomically `take`, yield the value to a block for transformation, and then
    # `put` the transformed value. Returns the transformed value. A timeout can
    # be set to limit the time spent blocked, in which case it returns `TIMEOUT`
    # if the time is exceeded.
    # @return [Object] the transformed value, or `TIMEOUT`
    def modify(timeout = nil)
      raise ArgumentError.new('no block given') unless block_given?

@mutex.synchronize do
        wait_for_full(timeout)

# If we timed out we'll still be empty
        if unlocked_full?
          value = @value
          @value = yield value
          @full_condition.signal
          apply_deref_options(value)
        else
          TIMEOUT
        end
      end
    end

# Non-blocking version of `take`, that returns `EMPTY` instead of blocking.
    def try_take!
      @mutex.synchronize do
        if unlocked_full?
          value = @value
          @value = EMPTY
          @empty_condition.signal
          apply_deref_options(value)
        else
          EMPTY
        end
      end
    end

# Non-blocking version of `put`, that returns whether or not it was successful.
    def try_put!(value)
      @mutex.synchronize do
        if unlocked_empty?
          @value = value
          @full_condition.signal
          true
        else
          false
        end
      end
    end

# Non-blocking version of `put` that will overwrite an existing value.
    def set!(value)
      @mutex.synchronize do
        old_value = @value
        @value = value
        @full_condition.signal
        apply_deref_options(old_value)
      end
    end

# Non-blocking version of `modify` that will yield with `EMPTY` if there is no value yet.
    def modify!
      raise ArgumentError.new('no block given') unless block_given?

@mutex.synchronize do
        value = @value
        @value = yield value
        if unlocked_empty?
          @empty_condition.signal
        else
          @full_condition.signal
        end
        apply_deref_options(value)
      end
    end

# Returns if the `MVar` is currently empty.
    def empty?
      @mutex.synchronize { @value == EMPTY }
    end

# Returns if the `MVar` currently contains a value.
    def full?
      !empty?
    end

protected

def synchronize(&block)
      @mutex.synchronize(&block)
    end

private

def unlocked_empty?
      @value == EMPTY
    end

def unlocked_full?
      ! unlocked_empty?
    end

def wait_for_full(timeout)
      wait_while(@full_condition, timeout) { unlocked_empty? }
    end

def wait_for_empty(timeout)
      wait_while(@empty_condition, timeout) { unlocked_full? }
    end

def wait_while(condition, timeout)
      if timeout.nil?
        while yield
          condition.wait(@mutex)
        end
      else
        stop = Concurrent.monotonic_time + timeout
        while yield && timeout > 0.0
          condition.wait(@mutex, timeout)
          timeout = stop - Concurrent.monotonic_time
        end
      end
    end
  end
end

${this.title}

Code Editor : mvar.rb