Class: Concurrent::MVar

Inherits:
Synchronization::Object show all
Includes:
Concern::Dereferenceable
Defined in:
lib/concurrent/mvar.rb

Overview

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.

## Copy Options

Object references in Ruby are mutable. This can lead to serious problems when the Concern::Dereferenceable#value of an object is a mutable reference. Which is always the case unless the value is a ‘Fixnum`, `Symbol`, or similar “primative” data type. Each instance can be configured with a few options that can help protect the program from potentially dangerous operations. Each of these options can be optionally set when the oject instance is created:

  • ‘:dup_on_deref` When true the object will call the `#dup` method on the `value` object every time the `#value` methid is called (default: false)

  • ‘:freeze_on_deref` When true the object will call the `#freeze` method on the `value` object every time the `#value` method is called (default: false)

  • ‘:copy_on_deref` When given a `Proc` object the `Proc` will be run every time the `#value` method is called. The `Proc` will be given the current `value` as its only argument and the result returned by the block will be the return value of the `#value` call. When `nil` this option will be ignored (default: nil)

When multiple deref options are set the order of operations is strictly defined. The order of deref operations is:

  • ‘:copy_on_deref`

  • ‘:dup_on_deref`

  • ‘:freeze_on_deref`

Because of this ordering there is no need to ‘#freeze` an object created by a provided `:copy_on_deref` block. Simply set `:freeze_on_deref` to `true`. Setting both `:dup_on_deref` to `true` and `:freeze_on_deref` to `true` is as close to the behavior of a “pure” functional language (like Erlang, Clojure, or Haskell) as we are likely to get in Ruby. ## See Also

    1. Barth, R. Nikhil, and Arvind. [M-Structures: Extending a parallel, non- strict, functional language with state](dl.acm.org/citation.cfm?id=652538). In Proceedings of the 5th

    ACM Conference on Functional Programming Languages and Computer Architecture (FPCA), 1991.

    1. Peyton Jones, A. Gordon, and S. Finne. [Concurrent Haskell](dl.acm.org/citation.cfm?id=237794).

    In Proceedings of the 23rd Symposium on Principles of Programming Languages (PoPL), 1996.

Constant Summary collapse

EMPTY =

Unique value that represents that an ‘MVar` was empty

Object.new
TIMEOUT =

Unique value that represents that an ‘MVar` timed out before it was able to produce a value.

Object.new

Instance Method Summary collapse

Methods included from Concern::Dereferenceable

#value

Methods inherited from Synchronization::Object

attr_atomic, attr_volatile, ensure_safe_initialization_when_final_fields_are_present, new, safe_initialization!, safe_initialization?, volatile_cas_fields

Constructor Details

#initialize(value = EMPTY, opts = {}) ⇒ MVar

Create a new ‘MVar`, either empty or with an initial value.

Parameters:

  • opts (Hash) (defaults to: {}) —

    the options controlling how the future will be processed

Options Hash (opts):

  • :dup_on_deref (Boolean) — default: false —

    Call ‘#dup` before returning the data from Concern::Dereferenceable#value

  • :freeze_on_deref (Boolean) — default: false —

    Call ‘#freeze` before returning the data from Concern::Dereferenceable#value

  • :copy_on_deref (Proc) — default: nil —

    When calling the Concern::Dereferenceable#value method, call the given proc passing the internal value as the sole argument then return the new value returned from the proc.



54
55
56
57
58
59
60
 
# File 'lib/concurrent/mvar.rb', line 54

def initialize(value = EMPTY, opts = {})
  @value = value
  @mutex = Mutex.new
  @empty_condition = ConditionVariable.new
  @full_condition = ConditionVariable.new
  set_deref_options(opts)
end

Instance Method Details

#borrow(timeout = nil) ⇒ Object

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.

Returns:

  • (Object) —

    the value returned by the block, or ‘TIMEOUT`



86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/concurrent/mvar.rb', line 86

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

#empty? ⇒ Boolean

Returns if the ‘MVar` is currently empty.

Returns:

  • (Boolean) 


195
196
197
 
# File 'lib/concurrent/mvar.rb', line 195

def empty?
  @mutex.synchronize { @value == EMPTY }
end

#full? ⇒ Boolean

Returns if the ‘MVar` currently contains a value.

Returns:

  • (Boolean) 


200
201
202
 
# File 'lib/concurrent/mvar.rb', line 200

def full?
  !empty?
end

#modify(timeout = nil) ⇒ Object

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.

Returns:

  • (Object) —

    the transformed value, or ‘TIMEOUT`

Raises:

  • (ArgumentError) 


123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/concurrent/mvar.rb', line 123

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

#modify! ⇒ Object

Non-blocking version of ‘modify` that will yield with `EMPTY` if there is no value yet.

Raises:

  • (ArgumentError) 


179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/concurrent/mvar.rb', line 179

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

#put(value, timeout = nil) ⇒ Object

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.

Returns:

  • (Object) —

    the value that was put, or ‘TIMEOUT`



103
104
105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/concurrent/mvar.rb', line 103

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

#set!(value) ⇒ Object

Non-blocking version of ‘put` that will overwrite an existing value.



169
170
171
172
173
174
175
176
 
# File 'lib/concurrent/mvar.rb', line 169

def set!(value)
  @mutex.synchronize do
    old_value = @value
    @value = value
    @full_condition.signal
    apply_deref_options(old_value)
  end
end

#take(timeout = nil) ⇒ Object

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.

Returns:

  • (Object) —

    the value that was taken, or ‘TIMEOUT`



66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/concurrent/mvar.rb', line 66

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

#try_put!(value) ⇒ Object

Non-blocking version of ‘put`, that returns whether or not it was successful.



156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/concurrent/mvar.rb', line 156

def try_put!(value)
  @mutex.synchronize do
    if unlocked_empty?
      @value = value
      @full_condition.signal
      true
    else
      false
    end
  end
end

#try_take! ⇒ Object

Non-blocking version of ‘take`, that returns `EMPTY` instead of blocking.



142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/concurrent/mvar.rb', line 142

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