Threads and the GVL

# frozen_string_literal: true

# Thread.new starts the block at once on a new thread; the arguments are
# passed to the block, which keeps them out of the closure.
isbns = %w[9780306406157 9780140449136]
lookups = isbns.map do |isbn|
  Thread.new(isbn) do |digits|
    Thread.current.name = "lookup-#{digits[-4..]}"
    sleep 0.01
    digits.reverse
  end
end

lookups.map(&:value) # => ["7516046030879", "6319440410879"]
lookups.map(&:name) # => ["lookup-6157", "lookup-9136"]
lookups.map(&:status) # => [false, false]
lookups.map(&:alive?) # => [false, false]

# join(limit) gives up after the limit and returns nil; the thread keeps running.
slow = Thread.new { sleep 0.2 }
slow.join(0.01) # => nil
slow.status # => "sleep"
slow.join # => #<Thread:0x... thread_basics.rb:21 dead>
slow.status # => false

# A thread that ends with an exception reports nil from status.
failed = Thread.new do
  Thread.current.report_on_exception = false
  raise IOError, "disk full"
end
sleep 0.05
failed.status # => nil
begin
  failed.join
rescue IOError => e
  e.message # => "disk full"
end

Thread.main.equal?(Thread.current) # => true
Thread.list.size # => 1

# frozen_string_literal: true

# status: "run" while running, "sleep" while blocked or sleeping, false when finished
# normally, nil when ended by an exception, "aborting" while being killed.
napper = Thread.new { Thread.stop }
Thread.pass until napper.stop?
napper.status # => "sleep"
napper.stop? # => true
napper.alive? # => true

# wakeup marks the thread runnable; run also schedules it right away.
napper.wakeup.equal?(napper) # => true
napper.join
napper.status # => false

Thread.current.status # => "run"
Thread.main.priority # => 0

# Threads belong to a ThreadGroup, inherited from the creating thread.
group = ThreadGroup.new
member = Thread.new { sleep }
group.add(member).equal?(group) # => true
group.list.first.equal?(member) # => true
member.group.equal?(group) # => true
ThreadGroup::Default.list.include?(Thread.main) # => true
member.kill
member.join

# A name makes inspect and debugging output readable.
named = Thread.new { sleep }
named.name = "returns-desk"
Thread.pass until named.stop?
puts named.inspect.sub(/0x\h+/, "0x...") # prints #<Thread:0x...@returns-desk thread_lifecycle.rb:30 sleep>
named.kill

# Thread.exit, Thread.current.exit and Thread#kill all end the target thread.
Thread.new { Thread.exit }.value # => nil

#<Thread:0x...@returns-desk thread_lifecycle.rb:30 sleep>

# frozen_string_literal: true

# Processes have separate memory; threads share it. After fork, only the thread
# that called fork exists in the child.
counter = 0
helper = Thread.new { sleep }
Thread.pass until helper.stop?

reader, writer = IO.pipe
pid = fork do
  reader.close
  counter += 1
  writer.puts "child: threads=#{Thread.list.size}, counter=#{counter}"
  writer.close
end
writer.close
puts reader.read
Process.wait(pid)
puts "parent: threads=#{Thread.list.size}, counter=#{counter}"

thread_counter = 0
Thread.new { thread_counter += 1 }.join
puts "after a thread: counter=#{thread_counter}"

child: threads=1, counter=1
parent: threads=2, counter=0
after a thread: counter=1

# frozen_string_literal: true

log = Thread::Queue.new
stop = false
workers = %i[a b].map do |name|
  Thread.new do
    last = nil
    until stop
      now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
      log << [name, now] if last.nil? || now - last > 0.02
      last = now
    end
  end
end
sleep 1
stop = true
workers.each(&:join)
log.close
events = Array.new(log.size) { log.pop }.sort_by(&:last)
base = events.first.last
turns = events.map { |name, time| "#{name}@#{((time - base) * 1000).round}ms" }
puts turns.first(8).join(" ")

a@0ms b@110ms a@220ms b@330ms a@443ms b@558ms a@668ms b@778ms

# frozen_string_literal: true

require "zlib"

# Pure Ruby CPU work: a naive checksum over many strings.
def checksum_work(rounds)
  rounds.times.sum { |i| (i.to_s * 3).sum }
end

# Waiting work: sleep stands in for a blocking read from a slow service.
def lookup_work = sleep(0.2)

# Native work that releases the GVL: compressing 8 MB of random bytes with zlib.
def compress_work(payload) = Zlib::Deflate.deflate(payload).bytesize

def seconds
  started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  yield
  Process.clock_gettime(Process::CLOCK_MONOTONIC) - started
end

def in_threads(count, &)
  Array.new(count) { Thread.new(&) }.each(&:join)
end

jobs = 4
payload = Random.new(1).bytes(8_000_000)

wait_serial = seconds { jobs.times { lookup_work } }
wait_threads = seconds { in_threads(jobs) { lookup_work } }

ruby_serial = seconds { jobs.times { checksum_work(300_000) } }
ruby_threads = seconds { in_threads(jobs) { checksum_work(300_000) } }

zlib_serial = seconds { jobs.times { compress_work(payload) } }
zlib_threads = seconds { in_threads(jobs) { compress_work(payload) } }

puts format("waiting:   threads take %<r>.2f times the serial time", r: wait_threads / wait_serial)
puts format("pure Ruby: threads take %<r>.2f times the serial time", r: ruby_threads / ruby_serial)
puts format("zlib:      threads take %<r>.2f times the serial time", r: zlib_threads / zlib_serial)

waiting:   threads take 0.25 times the serial time
pure Ruby: threads take 1.03 times the serial time
zlib:      threads take 0.32 times the serial time

# frozen_string_literal: true

# Counts operating-system threads: /proc on Linux, ps -M on macOS.
def native_threads
  if File.directory?("/proc/self/task")
    Dir.children("/proc/self/task").size
  else
    IO.popen(["ps", "-M", "-p", Process.pid.to_s], &:readlines).size - 1
  end
end

threads = Array.new(200) { Thread.new { sleep } }
Thread.pass until threads.all?(&:stop?)
puts "ruby threads: #{Thread.list.size}, native threads: #{native_threads}"

ruby threads: 201, native threads: 202

ruby threads: 201, native threads: 3

# frozen_string_literal: true

# A check-then-act sequence is not atomic: the GVL can switch threads between
# the read and the write. Thread.pass makes the switch deterministic here.
class LoanCounter
  attr_reader :count

  def initialize
    @count = 0
    @mutex = Thread::Mutex.new
  end

  def unsafe_increment
    current = @count
    Thread.pass
    @count = current + 1
  end

  # The same sequence inside Mutex#synchronize cannot interleave.
  def safe_increment
    @mutex.synchronize { unsafe_increment }
  end
end

def run_threads(times, &)
  Array.new(times) { Thread.new(&) }.each(&:join)
end

unsafe = LoanCounter.new
run_threads(10) { 100.times { unsafe.unsafe_increment } }
unsafe.count.between?(1, 999) # => true

safe = LoanCounter.new
run_threads(10) { 100.times { safe.safe_increment } }
safe.count # => 1000

# Mutex API: lock state can be inspected, but never use it to decide what to do.
mutex = Thread::Mutex.new
mutex.locked? # => false
mutex.synchronize { [mutex.locked?, mutex.owned?] } # => [true, true]
mutex.try_lock # => true
mutex.try_lock # => false
mutex.unlock.equal?(mutex) # => true

# A Mutex is not reentrant: locking it twice in one thread raises ThreadError.
mutex.synchronize do
  mutex.synchronize { :never }
rescue ThreadError => e
  e.message # => "deadlock; recursive locking"
end

begin
  mutex.unlock
rescue ThreadError => e
  e.message # => "Attempt to unlock a mutex which is not locked"
end

# Mutex#sleep releases the lock while it sleeps and reacquires it on waking.
# It returns nil when the timeout elapsed, and the slept seconds when woken early.
mutex.synchronize do
  mutex.sleep(0.01) # => nil
  [mutex.locked?, mutex.owned?] # => [true, true]
end

# frozen_string_literal: true

# Since Ruby 3.0 a Mutex belongs to a fiber, not to a thread.
mutex = Thread::Mutex.new
mutex.lock

other_fiber = Fiber.new do
  mutex.lock
rescue ThreadError => e
  e.message
end
other_fiber.resume # => "deadlock; lock already owned by another fiber belonging to the same thread"

mutex.owned? # => true
Fiber.new { mutex.owned? }.resume # => false
mutex.unlock

# Thread.ignore_deadlock (3.0) disables the "No live threads left" check.
Thread.ignore_deadlock # => false
Thread.report_on_exception # => true
Thread.abort_on_exception # => false

# frozen_string_literal: true

mutex = Thread::Mutex.new
events = Thread::Queue.new

# A signal handler runs in the main thread between instructions, where locking
# a Mutex is refused. Hand the work to a thread through a Queue instead.
Signal.trap("USR1") do
  mutex.synchronize { :never }
rescue ThreadError => e
  events << "mutex in trap: #{e.message}"
end

Signal.trap("USR2") { events << "queued from trap" }

Process.kill("USR1", Process.pid)
Process.kill("USR2", Process.pid)
2.times { puts events.pop }

mutex in trap: can't be called from trap context
queued from trap

# frozen_string_literal: true

require "monitor"

# A Monitor is a reentrant mutex: the owner may enter it again.
class ReturnDesk
  include MonitorMixin

  def initialize
    super # MonitorMixin#initialize must run
    @returned = []
    @arrived = new_cond
  end

  def give_back(title)
    synchronize do
      @returned << title
      @arrived.signal
    end
  end

  # Re-enters the monitor already held by this method.
  def give_back_all(titles)
    synchronize { titles.each { |title| give_back(title) } }
  end

  # wait_while releases the monitor while it waits and re-checks the block on every wake-up.
  def next_return
    synchronize do
      @arrived.wait_while { @returned.empty? }
      @returned.shift
    end
  end

  def size = synchronize { @returned.size }
end

desk = ReturnDesk.new
reader = Thread.new { desk.next_return }
desk.give_back_all(%w[Dune Emma Ulysses])
reader.value # => "Dune"
desk.size # => 2

monitor = Monitor.new
monitor.enter
monitor.mon_locked? # => true
monitor.mon_owned? # => true
monitor.try_enter # => true
monitor.exit
monitor.exit
monitor.mon_locked? # => false

begin
  monitor.exit
rescue ThreadError => e
  e.message # => "current fiber not owner"
end

# An existing object can be extended with the mixin instead of subclassing.
tally = [].extend(MonitorMixin)
tally.synchronize { tally << 1 }
tally.mon_locked? # => false

# frozen_string_literal: true

# A gate that opens once: waiters sleep on a ConditionVariable until the flag changes.
class OpeningGate
  def initialize
    @mutex = Thread::Mutex.new
    @opened = Thread::ConditionVariable.new
    @open = false
    @log = []
  end

  attr_reader :log

  def wait_for_opening(name)
    @mutex.synchronize do
      # Always wait in a loop: wake-ups can be spurious, and another thread
      # may have changed the state again before this one got the mutex back.
      @opened.wait(@mutex) until @open
      @log << name
    end
  end

  def open
    @mutex.synchronize do
      @open = true
      @opened.broadcast
    end
  end

  def wait_with_timeout(seconds)
    @mutex.synchronize do
      @opened.wait(@mutex, seconds) # with a timeout, wait returns when time is up, signalled or not
      @open
    end
  end
end

gate = OpeningGate.new
gate.wait_with_timeout(0.01) # => false

readers = %w[ana ben cleo].map { |name| Thread.new { gate.wait_for_opening(name) } }
Thread.pass until readers.all?(&:stop?)
gate.open
readers.each(&:join)
gate.log.sort # => ["ana", "ben", "cleo"]

# signal wakes one waiter, broadcast wakes all of them. Calling either with no
# waiters is harmless but the signal is not remembered.
cv = Thread::ConditionVariable.new
cv.signal.equal?(cv) # => true
cv.broadcast.equal?(cv) # => true

# frozen_string_literal: true

# Producer-consumer: one producer feeds a Queue, a pool of workers drains it.
ISBNS = %w[9780306406157 9780306406158 9780140449136 not-an-isbn 9781234567897].freeze

def valid_isbn?(value)
  digits = value.delete("-")
  return false unless digits.match?(/\A\d{13}\z/)

  sum = digits.chars.each_with_index.sum { |d, i| d.to_i * (i.even? ? 1 : 3) }
  (sum % 10).zero?
end

jobs = Thread::Queue.new
results = Thread::Queue.new

workers = Array.new(3) do
  Thread.new do
    # pop blocks while the queue is empty and returns nil once it is closed and drained.
    while (isbn = jobs.pop)
      results << [isbn, valid_isbn?(isbn)]
    end
  end
end

ISBNS.each { |isbn| jobs << isbn }
jobs.close
workers.each(&:join)
results.close

checked = []
while (pair = results.pop)
  checked << pair
end
checked.size # => 5
checked.to_h.select { |_, valid| valid }.keys.sort # => ["9780140449136", "9780306406157", "9781234567897"]

jobs.closed? # => true
jobs.empty? # => true
jobs.num_waiting # => 0
jobs.length # => 0

begin
  jobs << "9780306406157"
rescue ClosedQueueError => e
  e.message # => "queue closed"
end

# frozen_string_literal: true

# A SizedQueue applies back-pressure: push blocks while the queue is full.
buffer = Thread::SizedQueue.new(2)
produced = []
producer = Thread.new do
  %w[a b c d e].each do |item|
    buffer.push(item)
    produced << item
  end
  buffer.close
end

Thread.pass until buffer.size == 2 && producer.stop?
produced.join # => "ab"
buffer.max # => 2
buffer.num_waiting # => 1

consumed = []
while (item = buffer.pop)
  consumed << item
end
producer.join
consumed.join # => "abcde"

# Timeouts (3.2): push and pop give up and return nil instead of blocking forever.
full = Thread::SizedQueue.new(1)
full.push(:first)
full.push(:second, timeout: 0.01) # => nil
full.pop # => :first
full.pop(timeout: 0.01) # => nil

empty = Thread::Queue.new
empty.pop(timeout: 0) # => nil
empty.push(1)
empty.pop(timeout: 0) # => 1

# non_block = true raises instead of waiting.
begin
  empty.pop(true)
rescue ThreadError => e
  e.message # => "queue empty"
end

begin
  full.push(:a, true)
  full.push(:b, true)
rescue ThreadError => e
  e.message # => "queue full"
end

# A Queue can be created from any enumerable, and a closed queue refuses new items.
Thread::Queue.new([1, 2, 3]).size # => 3
closed = Thread::Queue.new.close
begin
  closed.push(1)
rescue ClosedQueueError => e
  [e.class, e.message] # => [ClosedQueueError, "queue closed"]
end
closed.pop # => nil

# Queues cannot be frozen (3.3): freezing would make them unusable.
begin
  Thread::Queue.new.freeze
rescue TypeError => e
  e.message # => "cannot freeze #<Thread::Queue:0x...>"
end

# frozen_string_literal: true

# Two threads take two locks in opposite orders. Ruby notices that every
# thread is asleep and nothing can wake them, and raises fatal in the main thread.
loans = Thread::Mutex.new
returns = Thread::Mutex.new
ready = Thread::Queue.new

first = Thread.new do
  loans.synchronize do
    ready << :first
    sleep 0.05
    returns.synchronize { :never }
  end
end

second = Thread.new do
  returns.synchronize do
    ready << :second
    sleep 0.05
    loans.synchronize { :never }
  end
end

2.times { ready.pop }
[first, second].each(&:join)

deadlock.rb:26:in 'Thread#join': No live threads left. Deadlock? (fatal)
3 threads, 3 sleeps current:0x... main thread:0x...
* #<Thread:0x... sleep_forever>
   rb_thread_t:0x... native:0x... int:0
   deadlock.rb:26:in 'Thread#join'
   deadlock.rb:26:in 'Array#each'
   deadlock.rb:26:in '<main>'
* #<Thread:0x... deadlock.rb:9 sleep_forever>
   rb_thread_t:0x... native:0x... int:0 mutex:3 cond:1
    depended by: tb_thread_id:0x...
   deadlock.rb:13:in 'Thread::Mutex#synchronize'
   deadlock.rb:13:in 'block (2 levels) in <main>'
   deadlock.rb:10:in 'Thread::Mutex#synchronize'
   deadlock.rb:10:in 'block in <main>'
* #<Thread:0x... deadlock.rb:17 sleep_forever>
   rb_thread_t:0x... native:0x... int:0
   deadlock.rb:21:in 'Thread::Mutex#synchronize'
   deadlock.rb:21:in 'block (2 levels) in <main>'
   deadlock.rb:18:in 'Thread::Mutex#synchronize'
   deadlock.rb:18:in 'block in <main>'

	from deadlock.rb:26:in 'Array#each'
	from deadlock.rb:26:in '<main>'

# frozen_string_literal: true

# ||= is a read, a test and a write: not atomic, so two threads can both
# find the ivar empty and both run the expensive block.
class Branch
  LOCK = Thread::Mutex.new

  # One shared default instance, created under a lock like Shelf.config could be.
  def self.default
    LOCK.synchronize { @default ||= new }
  end

  attr_reader :builds

  def initialize
    @builds = Thread::Queue.new
    @mutex = Thread::Mutex.new
  end

  def index
    @index ||= build_index
  end

  def safe_index
    @mutex.synchronize { @safe_index ||= build_index }
  end

  private

  def build_index
    @builds << :built
    sleep 0.05
    { "9780306406157" => "Dune" }.freeze
  end
end

def build_count(branch, method)
  Array.new(5) { Thread.new { branch.public_send(method) } }.each(&:join)
  branch.builds.size
end

build_count(Branch.new, :index) # => 5
build_count(Branch.new, :safe_index) # => 1

# Class-level state is shared by every thread: the same rule applies to
# Shelf.config, class variables and constants holding mutable objects.
Array.new(4) { Thread.new { Branch.default } }.map(&:value).uniq.size # => 1

# frozen_string_literal: true

require "singleton"

# Singleton makes .new private and adds .instance, which guards creation with a mutex.
class Config
  include Singleton

  # The hand-written variant: @unsafe ||= new is a check-then-act, so two
  # threads can each build an instance when new is slow.
  def self.unsafe_instance
    @unsafe_instance ||= new
  end

  def initialize
    super
    sleep 0.05
  end
end

hand_rolled = Array.new(4) { Thread.new { Config.unsafe_instance } }.map(&:value)
hand_rolled.uniq.size # => 4

guarded = Array.new(4) { Thread.new { Config.instance } }.map(&:value)
guarded.uniq.size # => 1

begin
  Config.new
rescue NoMethodError => e
  e.message # => "private method 'new' called for class Config"
end

# frozen_string_literal: true

Book = Data.define(:isbn, :title)

# A trimmed copy of Shelf::Catalog: every access to the store goes through one mutex.
class Catalog
  include Enumerable

  def initialize
    @books = []
    @mutex = Thread::Mutex.new
  end

  def add(book)
    @mutex.synchronize { @books << book }
    self
  end

  # Iterate over a snapshot so that a block that calls #add cannot deadlock or
  # see a half-updated list.
  def each(&)
    return enum_for(:each) unless block_given?

    @mutex.synchronize { @books.dup }.each(&)
  end
end

catalog = Catalog.new
writers = Array.new(4) do |n|
  Thread.new do
    25.times { |i| catalog.add(Book.new("#{n}-#{i}", "Title #{n}-#{i}")) }
  end
end
writers.each(&:join)
catalog.count # => 100
catalog.map(&:isbn).uniq.size # => 100

# A reader may add while iterating: each works on a snapshot and the lock is not held.
catalog.each { |book| catalog.add(book) if book.isbn == "0-0" }
catalog.count # => 101

# Had each yielded while holding @mutex, the add above would raise ThreadError
# ("deadlock; recursive locking").

# frozen_string_literal: true

class Cancelled < StandardError; end

# Thread#raise queues an exception in the target thread. Since Ruby 4.0 it
# accepts cause:, like Kernel#raise.
reason = IOError.new("lender went away")
waiting = Thread.new do
  Thread.current.report_on_exception = false
  sleep
rescue Cancelled => e
  [e.message, e.cause&.message]
end
Thread.pass until waiting.stop?
waiting.raise(Cancelled, "loan cancelled", cause: reason)
waiting.value # => ["loan cancelled", "lender went away"]

# Without cause:, the cause is whatever was being handled in the target thread (here nothing).
plain = Thread.new do
  sleep
rescue Cancelled => e
  e.cause
end
Thread.pass until plain.stop?
plain.raise(Cancelled)
plain.value # => nil

# Thread#kill runs ensure clauses and ends the thread; value is then nil.
log = []
victim = Thread.new do
  sleep
ensure
  log << :cleaned_up
end
Thread.pass until victim.stop?
victim.kill.join.equal?(victim) # => true
log.dup # => [:cleaned_up]
victim.value # => nil
victim.status # => false

# Thread.handle_interrupt defers asynchronous exceptions (raise, kill, Timeout)
# until the protected section ends.
steps = []
worker = Thread.new do
  Thread.current.report_on_exception = false
  Thread.handle_interrupt(Cancelled => :never) do
    steps << :critical_start
    sleep 0.05
    steps << :critical_end
    Thread.pending_interrupt? # => true
  end
  steps << :never_reached
rescue Cancelled
  steps << :cancelled_after_section
end
Thread.pass until steps.include?(:critical_start)
worker.raise(Cancelled)
worker.join
steps.dup # => [:critical_start, :critical_end, :cancelled_after_section]

# frozen_string_literal: true

require "timeout"

class LookupTimeout < StandardError; end

def attempt(label)
  puts "#{label}: returned #{yield.inspect}"
rescue Timeout::Error, LookupTimeout, ArgumentError => e
  puts "#{label}: #{e.class}: #{e.message}"
end

# Timeout.timeout raises inside the block's thread when time is up, and the
# caller sees Timeout::Error.
attempt("slow lookup") { Timeout.timeout(0.05) { sleep 1 } }
attempt("fast lookup") { Timeout.timeout(0.5) { :found } }

# A nil or zero limit means no limit; negative limits raise ArgumentError.
attempt("nil limit") { Timeout.timeout(nil) { :unbounded } }
attempt("negative") { Timeout.timeout(-1) { :never } }

# Inside the block the default exception is Timeout::ExitException, an
# Exception (not a StandardError) so that ordinary rescue clauses do not see it.
Timeout::ExitException.superclass # => Exception
Timeout::Error.superclass # => RuntimeError

attempt("rescue StandardError inside") do
  Timeout.timeout(0.05) do
    sleep 1
  rescue StandardError
    :swallowed
  end
end

# Anything that rescues Exception inside the block, or a custom StandardError
# class chosen for the timeout, can swallow it and the block then carries on.
attempt("rescue Exception inside") do
  Timeout.timeout(0.05) do
    sleep 1
  rescue Exception # rubocop:disable Lint/RescueException
    :swallowed
  end
end

attempt("custom class, caller") { Timeout.timeout(0.05, LookupTimeout, "catalogue too slow") { sleep 1 } }
attempt("custom class, swallowed") do
  Timeout.timeout(0.05, LookupTimeout) do
    sleep 1
  rescue StandardError
    :swallowed
  end
end

# Prefer an operation-level timeout when the API has one.
Thread::Queue.new.pop(timeout: 0.05) # => nil