Skip to content

Mutex guards, Queue hands over, ConditionVariable waits

Level: 301 · anyone whose threads share a counter, a list or a piece of work

One line: Mutex#synchronize makes count += 1 exact across ten threads, a Queue moves work from one thread to another and ends the consumer's loop with nil once it is closed, and a ConditionVariable lets a thread sleep until another thread says the state it needs is there — Python has the same three as Lock, queue.Queue and Condition, with RLock and Queue.join() alongside.

count += 1 is a read, an add and a write, and the GVL does not keep another thread from running between them, so a counter shared by threads can lose updates — the Concurrency library measures that loss in six languages, and this page measures only the cure. Mutex#synchronize takes the lock, runs the block, releases the lock in an ensure, and returns the block's value (rows 1 and 7); locked? says whether anyone holds it and owned? whether this thread does (row 2); try_lock takes it or returns false without waiting (row 3). Two ThreadErrors mark the edges: unlocking a mutex you do not hold (row 4), and locking a Mutex you already hold, which Ruby refuses as a recursive deadlock rather than letting it hang (row 5). A Monitor from the monitor library is the re-entrant version (row 6).

A Queue is a thread-safe list whose pop blocks while it is empty. close is how a producer says there is nothing more: after it, pop hands out what is left and then returns nil, so while (item = q.pop) is the consumer loop and needs no sentinel (row 8); pushing after close raises ClosedQueueError (row 9). pop(timeout:) returns nil when the time runs out and pop(true) raises ThreadError instead of waiting (row 10). A SizedQueue also blocks the producer while it is full, which is how five items pass through a queue of two (row 11) — backpressure in one class. A ConditionVariable is for the case a queue does not fit: a thread that must sleep until some shared state is right. wait(mutex) releases the mutex and sleeps; signal wakes one waiter, which re-takes the mutex before it continues (row 12). The wait is always in a loop over the predicate, because a waiter may wake without a signal — a spurious wakeup — and because another thread may have changed the state again (Not machine-checked here.). Ruby's Queue has no task_done or join (row 13); a program waits for the workers, not for the queue.

Verified output of mutex_queue_and_condition_variable_rb.rb — regenerated by tools/run_examples.py, never hand-typed.

 1. 10 threads x 1000 synchronize { count += 1 }         10000
 2. locked?/owned? outside -> inside synchronize         [false, false] -> [true, true]
 3. try_lock while I hold it / while it is free          false / true
 4. unlock when not locked                               ThreadError: Attempt to unlock a mutex which is not locked
 5. synchronize inside synchronize (Mutex)               ThreadError: deadlock; recursive locking
 6. the same on a Monitor (re-entrant)                   :ok
 7. synchronize returns                                  :the_blocks_value
 8. pop until close makes it return nil                  [1, 2, 3], closed? true
 9. push after close                                     ClosedQueueError: queue closed
10. pop(timeout: 0.2) / pop(true) on an empty queue      nil / ThreadError: queue empty
11. SizedQueue.new(2): max, 5 items through it           2, [0, 1, 2, 3, 4]
12. ConditionVariable wait/signal, in order              ["waiter: entering wait", "signaller: set ready, signalling", "waiter: woke, ready is true"]
13. Queue#task_done / Queue#join exist?                  false / false

Reading the output

Row 1 is the whole reason for a mutex: ten threads each added 1 a thousand times inside synchronize, and the total is exactly 10000. Rows 2 and 3 are the lock's questions — outside the block [false, false], inside [true, true], and try_lock is false while this thread already holds the mutex and true when it is free. Rows 4 and 5 are the two ThreadErrors, with their messages: Attempt to unlock a mutex which is not locked and deadlock; recursive locking. Row 6 shows that a Monitor lets the same thread enter twice, and row 7 that synchronize is an expression whose value is the block's.

Rows 8 to 11 are the queue. The consumer thread collected [1, 2, 3] and its loop ended by itself when pop returned nil after close; closed? is true; a fourth push raises ClosedQueueError: queue closed. On an empty, open queue, pop(timeout: 0.2) came back with nil after a fifth of a second and pop(true) raised ThreadError: queue empty at once. The SizedQueue of two let five items through in order, the producer blocking each time the queue was full. Row 12 is the handshake: the waiter entered wait, the signaller set ready and signalled, and the waiter woke with ready already true — the same three lines on every run, because the mutex orders them. Row 13 prints false / false: there is no task_done.

Compared with Python

Python's threading.Lock gives the same row 1 — 10000 — and the same non-blocking acquire(blocking=False) in row 3, but the lock has no owner: locked() is all it can say (row 2), and any thread may release it (Not machine-checked here.). Where Ruby raises ThreadError on a second synchronize in the same thread, a Python Lock simply blocks forever; row 5 proves that with acquire(timeout=0.2) returning False instead of hanging. RLock is the re-entrant one (row 6), and with lock: is a statement with no value (row 7), where Ruby's synchronize returns the block's. Releasing an unlocked lock is a RuntimeError (row 4).

queue.Queue blocks on get the same way, and Queue(maxsize=2) is the SizedQueue (row 11), but there is no close in Python 3.12: the consumer loop ends on a sentinel value (row 8), and nothing stops a put after it (row 9). Python 3.13 added Queue.shutdown(), which makes get raise queue.ShutDown once the queue is drained — closer to Ruby's close, but not available on the Ubuntu runner, so the twin does not use it (Not machine-checked here.). A timed-out get raises queue.Empty rather than returning None, and so does get_nowait() (row 10). Condition.wait_for(predicate) is Ruby's cv.wait(lock) until ready written as one call (row 12). Row 13 is the one thing Python's queue has and Ruby's lacks: task_done() and join(), so a producer can wait until every item it queued has been processed, not merely taken. threading.Event — a flag one thread sets and others wait on — has no direct Ruby class; a closed Queue or a ConditionVariable plays the part.

Verified output of mutex_queue_and_condition_variable_py.py — regenerated by tools/run_examples.py, never hand-typed.

 1. 10 threads x 1000 `with lock: count += 1`            10000
 2. locked() outside -> inside `with lock` (no owned?)   False -> True
 3. acquire(blocking=False) while held / while free      False / True
 4. release() when not locked                            RuntimeError
 5. acquire again inside `with lock` (Lock)              acquire(timeout=0.2) -> False: it would block forever
 6. the same on an RLock (re-entrant)                    'ok'
 7. `with lock:` returns                                 nothing: with is a statement, not an expression
 8. get until a None sentinel (no close in 3.12)         [1, 2, 3], closed? n/a
 9. put after the sentinel                               accepted; qsize() is 1
10. get(timeout=0.2) / get_nowait() on an empty queue    Empty / Empty
11. Queue(maxsize=2): maxsize, 5 items through it        2, [0, 1, 2, 3, 4]
12. Condition wait_for/notify, in order                  ['waiter: entering wait', 'signaller: set ready, notifying', 'waiter: woke, ready is True']
13. Queue.task_done / Queue.join exist?                  True / True: join() returned after 3 task_done() calls

The Concurrency library measures the bug this page guards against, Is total += n safe on two threads? ↗, and the cures side by side in Keeping every update ↗.

Try it

  1. Two producers push 1..5 and 6..10 through a SizedQueue of capacity 3; two consumers pop until the queue is closed and add each item into a total guarded by a Mutex. Print the total, the number of items and whether the queue ended closed and empty — and nothing about which consumer took which item.
  2. Replace the ConditionVariable handshake of row 12 with a Queue used as a one-shot signal (q.pop waits for q << :go). Then write down what a ConditionVariable can express that a Queue cannot: a predicate over shared state, and broadcast to every waiter at once.
Solution to 1

mutex_queue_and_condition_variable_kata_rb.rb in full — pasted here by tools/run_examples.py from the file CI runs.

# Kata: two producers push 1..5 and 6..10 through a SizedQueue of capacity 3;
# two consumers pop until the queue is closed and add what they get into a
# total that a Mutex guards. The total and the count are the same on every
# run; which consumer took which item is not, so it is not printed.
sq = SizedQueue.new(3)
producers = [(1..5), (6..10)].map do |range|
  Thread.new { range.each { |i| sq << i } }
end

lock = Mutex.new
total = 0
count = 0
consumers = 2.times.map do
  Thread.new do
    while (item = sq.pop)
      lock.synchronize do
        total += item
        count += 1
      end
    end
  end
end

producers.each(&:join)
sq.close
consumers.each(&:join)

puts "capacity of the queue:  #{sq.max}"
puts "items consumed:         #{count}"
puts "sum of the items:       #{total}"
puts "closed and drained:     #{sq.closed? && sq.empty?}"

Verified output of mutex_queue_and_condition_variable_kata_rb.rb — regenerated by tools/run_examples.py, never hand-typed.

capacity of the queue:  3
items consumed:         10
sum of the items:       55
closed and drained:     true

See also