Multithreading #

require 'benchmark'

# Demonstrating the GIL on CPU-bound tasks
def intensive_calc(n)
  n.times { Math.sqrt(rand) }
end

n = 5_000_000

# Sequential — one at a time
sequential_time = Benchmark.realtime do
  4.times { intensive_calc(n) }
end

# Multi-threaded — 4 threads at once
threaded_time = Benchmark.realtime do
  threads = 4.times.map { Thread.new { intensive_calc(n) } }
  threads.each(&:join)
end

puts "Sequential: #{sequential_time.round(2)}s"
puts "4 threads:  #{threaded_time.round(2)}s"
# The results are almost the same — the GIL prevents true parallelism for CPU-bound work!

# The simplest thread
t = Thread.new { puts "Hello from a new thread!" }
t.join   # wait for the thread to finish before continuing

# Thread with a multi-line block
thread = Thread.new do
  puts "Thread started"
  sleep 1
  puts "Thread finished"
end

puts "Main thread keeps running"
thread.join
puts "Main thread waited for completion"

# value — like join but returns the calculation result
thread = Thread.new { (1..100).sum }
result = thread.value   # wait and get the result
puts result   # => 5050

# join with a timeout — don't wait forever
t = Thread.new { sleep 10 }
finished = t.join(2)   # wait at most 2 seconds
if finished.nil?
  puts "Thread timed out — still running"
  t.kill   # force-stop it
else
  puts "Thread finished on time"
end

# Running many threads and collecting results
urls = ["https://api1.com", "https://api2.com", "https://api3.com"]

# ANTI-PATTERN: sequential — slow
# results = urls.map { |url| fetch_data(url) }

# CORRECT: parallel I/O with threads
results = urls.map { |url| Thread.new { fetch_data(url) } }
            .map(&:value)   # collect all results

puts results.inspect

t = Thread.new { sleep 2 }

puts t.status    # => "sleep"  (currently sleeping)
puts t.alive?    # => true
puts t.stop?     # => true  (stop is also true while sleeping!)

sleep 1
puts t.status    # => "sleep"

t.join
puts t.status    # => false  (finished normally)
puts t.alive?    # => false

# A thread that crashes
t_crash = Thread.new { raise "Intentional crash" }
t_crash.join rescue nil
puts t_crash.status   # => nil  (finished because of an exception)

# Possible statuses: "run", "sleep", "aborting", false, nil

# Thread.current[] — storing values per thread
Thread.new do
  Thread.current[:name] = "Thread A"
  sleep 0.1
  puts Thread.current[:name]   # => "Thread A"
end

Thread.new do
  Thread.current[:name] = "Thread B"
  sleep 0.1
  puts Thread.current[:name]   # => "Thread B"  (unaffected by Thread A)
end.join

# Thread.current[] is useful for storing per-request context
# in web frameworks like Rails (e.g. current_user)

# ANTI-PATTERN: the classic race condition
counter = 0
threads = 10.times.map do
  Thread.new do
    1000.times { counter += 1 }   # counter += 1 is not atomic!
  end
end
threads.each(&:join)
puts counter   # Should be 10000, but might be less!
# counter += 1 is actually: read the value, add 1, write it back
# another thread can slip in between these steps

counter = 0
mutex   = Mutex.new

threads = 10.times.map do
  Thread.new do
    1000.times do
      mutex.synchronize do
        counter += 1   # now safe — only one thread enters this block
      end
    end
  end
end
threads.each(&:join)
puts counter   # => always 10000

# Mutex also has try_lock — doesn't block if already locked
mutex = Mutex.new
acquired = mutex.try_lock
if acquired
  begin
    # do something
  ensure
    mutex.unlock
  end
else
  puts "Mutex is being used by another thread, skipping"
end

mutex_a = Mutex.new
mutex_b = Mutex.new

# Thread 1: locks A first, then B
thread1 = Thread.new do
  mutex_a.synchronize do
    sleep 0.1   # pause so thread 2 has time to lock B
    mutex_b.synchronize do
      puts "Thread 1 finished"
    end
  end
end

# Thread 2: locks B first, then A — DEADLOCK!
thread2 = Thread.new do
  mutex_b.synchronize do
    sleep 0.1
    mutex_a.synchronize do   # ← waits for thread 1 to release A, but thread 1 waits for B
      puts "Thread 2 finished"
    end
  end
end

[thread1, thread2].each(&:join)
# The program hangs forever — deadlock!

# Preventing deadlocks: always lock mutexes in the same order
mutex_a = Mutex.new
mutex_b = Mutex.new

# Both threads lock A first, then B — no deadlock
thread1 = Thread.new do
  mutex_a.synchronize do
    mutex_b.synchronize do
      puts "Thread 1 finished"
    end
  end
end

thread2 = Thread.new do
  mutex_a.synchronize do   # ← same order: A first, then B
    mutex_b.synchronize do
      puts "Thread 2 finished"
    end
  end
end

[thread1, thread2].each(&:join)

mutex  = Mutex.new
cond   = ConditionVariable.new
ready  = false

# Consumer — waits for the producer to be ready
consumer = Thread.new do
  mutex.synchronize do
    cond.wait(mutex) until ready   # wait until ready = true
    puts "Consumer: data received!"
  end
end

# Producer — prepares data then signals
producer = Thread.new do
  sleep 1
  mutex.synchronize do
    ready = true
    cond.signal   # wake up the waiting consumer
    puts "Producer: data sent!"
  end
end

[producer, consumer].each(&:join)

require 'thread'   # not needed in Ruby >= 3.2, already built-in

queue = Queue.new

# Producer — generates data
producer = Thread.new do
  10.times do |i|
    sleep rand(0.1..0.3)
    queue << "item-#{i}"
    puts "Produced: item-#{i}"
  end
  queue << :done   # sentinel value
end

# Consumer — processes data
consumer = Thread.new do
  loop do
    item = queue.pop   # blocks until there's an item
    break if item == :done
    puts "Processed: #{item}"
    sleep rand(0.1..0.2)
  end
end

[producer, consumer].each(&:join)

queue = SizedQueue.new(5)   # at most 5 items

# Producer — will pause when the queue is full
producer = Thread.new do
  20.times do |i|
    queue.push("item-#{i}")   # blocks when the queue already has 5 items
    puts "Produced: item-#{i} | Queue: #{queue.size}/5"
  end
end

# Consumer — processes slower than the producer
consumer = Thread.new do
  20.times do
    item = queue.pop
    sleep 0.2   # slower than the producer
    puts "Processed: #{item}"
  end
end

[producer, consumer].each(&:join)

class ThreadPool
  def initialize(size)
    @size   = size
    @queue  = Queue.new
    @workers = size.times.map do
      Thread.new do
        loop do
          task = @queue.pop
          break if task == :shutdown
          begin
            task.call
          rescue => e
            puts "Worker error: #{e.message}"
          end
        end
      end
    end
  end

  def submit(&block)
    @queue << block
  end

  def shutdown
    @size.times { @queue << :shutdown }
    @workers.each(&:join)
  end
end

# Usage
pool = ThreadPool.new(4)   # 4 worker threads

20.times do |i|
  pool.submit do
    sleep rand(0.1..0.5)
    puts "Task #{i} finished by #{Thread.current.object_id}"
  end
end

pool.shutdown
puts "All tasks finished"

# Basic fiber — explicit control of when to switch
fiber = Fiber.new do
  puts "Fiber: step 1"
  Fiber.yield   # hand control back to the caller
  puts "Fiber: step 2"
  Fiber.yield
  puts "Fiber: step 3"
end

puts "Main: start"
fiber.resume   # => "Fiber: step 1"
puts "Main: back"
fiber.resume   # => "Fiber: step 2"
puts "Main: back again"
fiber.resume   # => "Fiber: step 3"

# Fiber with values passed back and forth
generator = Fiber.new do
  value = 0
  loop do
    value += 1
    Fiber.yield value   # send the value to the caller
  end
end

puts generator.resume   # => 1
puts generator.resume   # => 2
puts generator.resume   # => 3

# Ractor — true parallelism
# Each Ractor has its own GIL → runs truly in parallel

# Heavy computation that can be parallelized with Ractor
def heavy_computation(n)
  (1..n).sum { |i| Math.sqrt(i) }
end

# Sequential — slow
seq_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
seq_results = 4.times.map { heavy_computation(1_000_000) }
seq_time = Process.clock_gettime(Process::CLOCK_MONOTONIC) - seq_time

# Parallel with Ractor — faster on multi-core machines!
ractor_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
ractors = 4.times.map do
  Ractor.new { heavy_computation(1_000_000) }
end
ractor_results = ractors.map(&:take)
ractor_time = Process.clock_gettime(Process::CLOCK_MONOTONIC) - ractor_time

puts "Sequential: #{seq_time.round(2)}s"
puts "Ractor:     #{ractor_time.round(2)}s"
# On a 4-core machine: Ractor ~4x faster!

# Communication between Ractors through message passing
r = Ractor.new do
  message = Ractor.receive   # wait for an incoming message
  "Processed: #{message}"
end

r.send("Hello from main!")
puts r.take   # => "Processed: Hello from main!"

# Ractor Pipeline — a series of Ractors processing data
pipeline = (1..5).map do |stage|
  Ractor.new(stage) do |n|
    loop do
      data = Ractor.receive
      result = data * n   # different transformation at each stage
      Ractor.yield result
    end
  end
end

# Send data to the first pipeline stage
pipeline[0].send(10)
# Data flows through every Ractor

# Approach 1: fork for CPU-bound tasks
pids = 4.times.map do |i|
  fork do
    result = heavy_computation(1_000_000)
    # Send the result to the parent process via a pipe or file
    puts "Worker #{i}: #{result}"
    exit
  end
end
pids.each { |pid| Process.wait(pid) }

# Approach 2: the Parallel gem — a convenient abstraction
# gem install parallel
require 'parallel'

data = (1..20).to_a

# Parallel.map — uses fork automatically
results = Parallel.map(data, in_processes: 4) do |n|
  heavy_computation(n * 100_000)
end

# Parallel with threads (for I/O-bound work)
results = Parallel.map(urls, in_threads: 10) do |url|
  HTTP.get(url).body
end

# 1. Use thread-safe objects from the standard library
require 'thread'

# Queue, SizedQueue — already thread-safe
queue = Queue.new

# Regular Hash and Array are NOT thread-safe for writes
# ANTI-PATTERN:
results = {}
threads = urls.map do |url|
  Thread.new { results[url] = fetch(url) }   # ← race condition!
end
threads.each(&:join)

# CORRECT: collect with join/value and assign after everything finishes
results = urls.map { |url| Thread.new { [url, fetch(url)] } }
            .map(&:value)
            .to_h

# 2. Minimize shared mutable state
# The less data shared between threads, the fewer
# race conditions and synchronization needs

# 3. Handle exceptions inside threads
thread = Thread.new do
  begin
    # code that might fail
    result = risky_process()
  rescue => e
    puts "Thread error: #{e.message}"
    nil   # return nil as a fallback
  end
end

# Without a rescue inside the thread, exceptions are silently ignored!
# Thread.abort_on_exception = true  ← enable for debugging

# 4. Set abort_on_exception for development
Thread.abort_on_exception = true   # the program crashes if any thread raises

# 5. Use timeouts for threads that might hang
require 'timeout'
begin
  Timeout.timeout(5) do
    thread.join   # wait at most 5 seconds
  end
rescue Timeout::Error
  thread.kill
  puts "Thread timed out!"
end