Threads & Thread Safety

Basic Thread Creation and Management

class ThreadManager
  def self.basic_threading_demo
    puts "Main thread: #{Thread.current}"

    # Create a simple thread
    thread1 = Thread.new do
      5.times do |i|
        puts "Thread 1: Count #{i}"
        sleep(0.1)
      end
    end

    # Create another thread with parameters
    thread2 = Thread.new("Background task") do |task_name|
      3.times do |i|
        puts "#{task_name}: Processing #{i}"
        sleep(0.15)
      end
    end

    # Main thread continues execution
    puts "Main thread continues..."

    # Wait for threads to complete
    thread1.join
    thread2.join

    puts "All threads completed"
  end

  def self.thread_lifecycle_demo
    # Thread with return value
    calculator_thread = Thread.new do
      result = (1..1000).sum
      puts "Calculation completed in thread"
      result  # Thread return value
    end

    # Check thread status
    puts "Thread alive? #{calculator_thread.alive?}"
    puts "Thread status: #{calculator_thread.status}"

    # Get thread return value
    result = calculator_thread.value  # Blocks until thread completes
    puts "Thread result: #{result}"
    puts "Thread status after completion: #{calculator_thread.status}"
  end

  def self.thread_exceptions_demo
    # Thread exception handling
    problematic_thread = Thread.new do
      raise "Something went wrong!"
    end

    begin
      problematic_thread.join  # Exception propagates here
    rescue => e
      puts "Caught exception from thread: #{e.message}"
    end

    # Thread with internal exception handling
    safe_thread = Thread.new do
      begin
        risky_operation
      rescue => e
        puts "Exception handled within thread: #{e.message}"
      end
    end

    safe_thread.join
  end

  private

  def self.risky_operation
    raise "Internal error"
  end
end

# Demonstrate threading
ThreadManager.basic_threading_demo
puts "\n" + "="*50 + "\n"
ThreadManager.thread_lifecycle_demo
puts "\n" + "="*50 + "\n"
ThreadManager.thread_exceptions_demo

Thread Communication and Coordination

class ThreadCommunication
  def self.producer_consumer_demo
    queue = Queue.new

    # Producer thread
    producer = Thread.new do
      10.times do |i|
        item = "Item #{i}"
        queue << item
        puts "Produced: #{item}"
        sleep(0.1)
      end
      queue << :done  # Signal completion
    end

    # Consumer thread
    consumer = Thread.new do
      loop do
        item = queue.pop
        break if item == :done

        puts "Consumed: #{item}"
        sleep(0.15)  # Simulate processing time
      end
    end

    producer.join
    consumer.join
    puts "Producer-Consumer demo completed"
  end

  def self.thread_local_variables_demo
    # Thread-local variables
    Thread.main[:name] = "Main Thread"

    threads = 3.times.map do |i|
      Thread.new(i) do |thread_id|
        Thread.current[:name] = "Worker #{thread_id}"
        Thread.current[:processed] = 0

        5.times do |work_unit|
          Thread.current[:processed] += 1
          puts "#{Thread.current[:name]} processed unit #{work_unit}"
          sleep(0.05)
        end

        puts "#{Thread.current[:name]} completed #{Thread.current[:processed]} units"
      end
    end

    threads.each(&:join)
  end

  def self.thread_group_demo
    # Thread groups for organization
    group = ThreadGroup.new

    workers = 3.times.map do |i|
      thread = Thread.new(i) do |worker_id|
        puts "Worker #{worker_id} starting"
        sleep(1)
        puts "Worker #{worker_id} finished"
      end

      group.add(thread)
      thread
    end

    puts "Thread group has #{group.list.length} threads"

    # Wait for all threads in group
    workers.each(&:join)
  end
end

ThreadCommunication.producer_consumer_demo
puts "\n" + "="*40 + "\n"
ThreadCommunication.thread_local_variables_demo
puts "\n" + "="*40 + "\n"
ThreadCommunication.thread_group_demo

Race Condition Examples

class RaceConditionDemo
  def self.unsafe_counter_demo
    counter = 0

    # Create multiple threads that increment the counter
    threads = 10.times.map do
      Thread.new do
        1000.times do
          # This is NOT thread-safe!
          temp = counter
          counter = temp + 1
        end
      end
    end

    threads.each(&:join)

    puts "Expected: 10000"
    puts "Actual: #{counter}"
    puts "Data race occurred!" if counter != 10000
  end

  def self.bank_account_race_demo
    class UnsafeBankAccount
      def initialize(balance = 0)
        @balance = balance
      end

      def balance
        @balance
      end

      def withdraw(amount)
        return false if @balance < amount

        # Simulate processing delay
        sleep(0.001)
        @balance -= amount
        true
      end

      def deposit(amount)
        sleep(0.001)
        @balance += amount
      end
    end

    account = UnsafeBankAccount.new(1000)

    # Multiple threads trying to withdraw
    threads = 5.times.map do
      Thread.new do
        10.times do
          account.withdraw(10) if account.balance >= 10
        end
      end
    end

    threads.each(&:join)

    puts "Final balance: #{account.balance}"
    puts "Expected: 500, but likely different due to race conditions"
  end

  def self.shared_array_demo
    shared_array = []

    threads = 5.times.map do |i|
      Thread.new(i) do |thread_id|
        100.times do |j|
          # Race condition: array modification is not atomic
          shared_array << "Thread-#{thread_id}-Item-#{j}"
        end
      end
    end

    threads.each(&:join)

    puts "Expected items: 500"
    puts "Actual items: #{shared_array.length}"
    puts "Some items may be lost due to race conditions"
  end
end

puts "=== Race Condition Demonstrations ==="
RaceConditionDemo.unsafe_counter_demo
puts "\n" + "-"*40 + "\n"
RaceConditionDemo.bank_account_race_demo
puts "\n" + "-"*40 + "\n"
RaceConditionDemo.shared_array_demo

⚠️ Common Race Condition Patterns

• Read-Modify-Write: Multiple threads reading, modifying, and writing shared data
• Check-Then-Act: Checking a condition then acting on it (condition may change)
• Non-atomic operations: Operations that appear atomic but aren't (like ++ in other languages)
• Shared collections: Multiple threads modifying arrays, hashes, or other collections

Thread-Safe Programming with Mutex

class ThreadSafeExamples
  def self.safe_counter_demo
    counter = 0
    mutex = Mutex.new

    threads = 10.times.map do
      Thread.new do
        1000.times do
          mutex.synchronize do
            # Critical section - only one thread at a time
            temp = counter
            counter = temp + 1
          end
        end
      end
    end

    threads.each(&:join)

    puts "Thread-safe counter result: #{counter}"
    puts "Should always be 10000"
  end

  def self.safe_bank_account_demo
    class ThreadSafeBankAccount
      def initialize(balance = 0)
        @balance = balance
        @mutex = Mutex.new
      end

      def balance
        @mutex.synchronize { @balance }
      end

      def withdraw(amount)
        @mutex.synchronize do
          return false if @balance < amount

          sleep(0.001)  # Simulate processing delay
          @balance -= amount
          true
        end
      end

      def deposit(amount)
        @mutex.synchronize do
          sleep(0.001)
          @balance += amount
        end
      end

      def transfer(other_account, amount)
        # Avoid deadlock by always acquiring locks in same order
        first_lock, second_lock = [@mutex, other_account.instance_variable_get(:@mutex)].sort_by(&:object_id)

        first_lock.synchronize do
          second_lock.synchronize do
            if @balance >= amount
              @balance -= amount
              other_account.instance_variable_set(:@balance,
                other_account.instance_variable_get(:@balance) + amount)
              true
            else
              false
            end
          end
        end
      end
    end

    account1 = ThreadSafeBankAccount.new(1000)
    account2 = ThreadSafeBankAccount.new(1000)

    # Multiple threads performing operations
    threads = []

    # Withdrawal threads
    threads += 5.times.map do
      Thread.new do
        10.times do
          account1.withdraw(10)
        end
      end
    end

    # Transfer threads
    threads += 3.times.map do
      Thread.new do
        5.times do
          account1.transfer(account2, 20)
        end
      end
    end

    threads.each(&:join)

    puts "Account 1 balance: #{account1.balance}"
    puts "Account 2 balance: #{account2.balance}"
    puts "Total: #{account1.balance + account2.balance}"
  end
end

ThreadSafeExamples.safe_counter_demo
puts "\n" + "-"*40 + "\n"
ThreadSafeExamples.safe_bank_account_demo

Advanced Synchronization Patterns

class AdvancedSynchronization
  def self.condition_variable_demo
    require 'thread'

    mutex = Mutex.new
    condition = ConditionVariable.new
    items = []
    finished = false

    # Producer thread
    producer = Thread.new do
      10.times do |i|
        mutex.synchronize do
          items << "Item #{i}"
          puts "Produced: Item #{i}"
          condition.signal  # Wake up waiting consumer
        end
        sleep(0.1)
      end

      mutex.synchronize do
        finished = true
        condition.broadcast  # Wake up all waiting threads
      end
    end

    # Consumer threads
    consumers = 2.times.map do |consumer_id|
      Thread.new do
        loop do
          item = nil

          mutex.synchronize do
            # Wait until there's an item or production is finished
            while items.empty? && !finished
              condition.wait(mutex)
            end

            break if items.empty? && finished

            item = items.shift
          end

          if item
            puts "Consumer #{consumer_id} consumed: #{item}"
            sleep(0.15)  # Simulate processing
          end
        end
      end
    end

    producer.join
    consumers.each(&:join)
    puts "Producer-Consumer with ConditionVariable completed"
  end

  def self.thread_pool_demo
    class SimpleThreadPool
      def initialize(size)
        @size = size
        @jobs = Queue.new
        @pool = Array.new(size) do |i|
          Thread.new do
            Thread.current[:id] = i
            loop do
              job = @jobs.pop
              break if job == :shutdown

              begin
                job.call
              rescue => e
                puts "Error in thread #{Thread.current[:id]}: #{e.message}"
              end
            end
          end
        end
      end

      def schedule(&block)
        @jobs << block
      end

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

    pool = SimpleThreadPool.new(3)

    # Schedule work
    10.times do |i|
      pool.schedule do
        puts "Job #{i} executed by thread #{Thread.current[:id]}"
        sleep(0.1)
      end
    end

    # Wait a bit then shutdown
    sleep(2)
    pool.shutdown
    puts "Thread pool demo completed"
  end

  def self.read_write_lock_demo
    # Simple read-write lock implementation
    class ReadWriteLock
      def initialize
        @readers = 0
        @writer = false
        @mutex = Mutex.new
        @reader_cv = ConditionVariable.new
        @writer_cv = ConditionVariable.new
      end

      def read_lock
        @mutex.synchronize do
          while @writer
            @reader_cv.wait(@mutex)
          end
          @readers += 1
        end
      end

      def read_unlock
        @mutex.synchronize do
          @readers -= 1
          @writer_cv.signal if @readers == 0
        end
      end

      def write_lock
        @mutex.synchronize do
          while @readers > 0 || @writer
            @writer_cv.wait(@mutex)
          end
          @writer = true
        end
      end

      def write_unlock
        @mutex.synchronize do
          @writer = false
          @reader_cv.broadcast
          @writer_cv.signal
        end
      end

      def with_read_lock
        read_lock
        yield
      ensure
        read_unlock
      end

      def with_write_lock
        write_lock
        yield
      ensure
        write_unlock
      end
    end

    data = { counter: 0 }
    rw_lock = ReadWriteLock.new

    # Reader threads
    readers = 5.times.map do |i|
      Thread.new do
        10.times do
          rw_lock.with_read_lock do
            puts "Reader #{i}: counter = #{data[:counter]}"
            sleep(0.01)
          end
        end
      end
    end

    # Writer threads
    writers = 2.times.map do |i|
      Thread.new do
        5.times do
          rw_lock.with_write_lock do
            data[:counter] += 1
            puts "Writer #{i}: incremented counter to #{data[:counter]}"
            sleep(0.02)
          end
        end
      end
    end

    (readers + writers).each(&:join)
    puts "Read-Write lock demo completed"
  end
end

AdvancedSynchronization.condition_variable_demo
puts "\n" + "="*50 + "\n"
AdvancedSynchronization.thread_pool_demo
puts "\n" + "="*50 + "\n"
AdvancedSynchronization.read_write_lock_demo

GVL Impact Demonstration

require 'benchmark'

class GVLDemo
  def self.cpu_bound_task
    # Simulate CPU-intensive work
    1_000_000.times { Math.sqrt(rand(1000)) }
  end

  def self.io_bound_task
    # Simulate I/O-bound work
    sleep(0.1)
  end

  def self.cpu_bound_comparison
    puts "=== CPU-bound Task Comparison ==="

    # Single-threaded execution
    single_threaded_time = Benchmark.realtime do
      4.times { cpu_bound_task }
    end

    # Multi-threaded execution
    multi_threaded_time = Benchmark.realtime do
      threads = 4.times.map do
        Thread.new { cpu_bound_task }
      end
      threads.each(&:join)
    end

    puts "Single-threaded: #{single_threaded_time.round(3)}s"
    puts "Multi-threaded:  #{multi_threaded_time.round(3)}s"
    puts "Speedup: #{(single_threaded_time / multi_threaded_time).round(2)}x"
    puts "Note: Speedup is minimal due to GVL limiting CPU parallelism"
  end

  def self.io_bound_comparison
    puts "\n=== I/O-bound Task Comparison ==="

    # Single-threaded execution
    single_threaded_time = Benchmark.realtime do
      4.times { io_bound_task }
    end

    # Multi-threaded execution
    multi_threaded_time = Benchmark.realtime do
      threads = 4.times.map do
        Thread.new { io_bound_task }
      end
      threads.each(&:join)
    end

    puts "Single-threaded: #{single_threaded_time.round(3)}s"
    puts "Multi-threaded:  #{multi_threaded_time.round(3)}s"
    puts "Speedup: #{(single_threaded_time / multi_threaded_time).round(2)}x"
    puts "Note: Significant speedup due to GVL release during I/O"
  end

  def self.gvl_release_demo
    puts "\n=== GVL Release Points ==="

    start_time = Time.now

    threads = 3.times.map do |i|
      Thread.new do
        puts "Thread #{i} started at #{(Time.now - start_time).round(3)}s"

        # GVL is released during sleep
        sleep(0.1)
        puts "Thread #{i} after sleep at #{(Time.now - start_time).round(3)}s"

        # GVL is held during computation
        1000.times { Math.sqrt(rand(100)) }
        puts "Thread #{i} finished at #{(Time.now - start_time).round(3)}s"
      end
    end

    threads.each(&:join)
  end

  def self.thread_switching_demo
    puts "\n=== Thread Switching Behavior ==="

    mutex = Mutex.new

    threads = 5.times.map do |i|
      Thread.new do
        10.times do |j|
          mutex.synchronize do
            print "#{i}"
            # Small delay to encourage thread switching
            sleep(0.001)
          end
        end
      end
    end

    threads.each(&:join)
    puts "\nNote: Output shows thread interleaving despite GVL"
  end
end

GVLDemo.cpu_bound_comparison
GVLDemo.io_bound_comparison
GVLDemo.gvl_release_demo
GVLDemo.thread_switching_demo

💡 GVL Key Points

• CPU-bound tasks: Limited by GVL - consider processes or native extensions
• I/O-bound tasks: GVL is released - threading provides real concurrency benefits
• C extensions: Can release GVL for true parallelism
• Alternative implementations: JRuby and TruffleRuby don't have GVL

Thread-Safe Patterns and Anti-Patterns

class ThreadingBestPractices
  # ✅ GOOD: Thread-safe singleton pattern
  class ThreadSafeSingleton
    @instance_mutex = Mutex.new

    def self.instance
      return @instance if @instance

      @instance_mutex.synchronize do
        @instance ||= new
      end
    end

    private_class_method :new
  end

  # ✅ GOOD: Immutable objects are thread-safe
  class ImmutableCounter
    def initialize(value = 0)
      @value = value.freeze
    end

    def increment
      self.class.new(@value + 1)
    end

    def value
      @value
    end
  end

  # ✅ GOOD: Thread-local storage
  class ThreadLocalCache
    def self.get(key)
      cache = Thread.current[:cache] ||= {}
      cache[key]
    end

    def self.set(key, value)
      cache = Thread.current[:cache] ||= {}
      cache[key] = value
    end
  end

  # ❌ BAD: Double-checked locking (broken in Ruby)
  class BrokenSingleton
    def self.instance
      return @instance if @instance  # Not thread-safe!

      @mutex.synchronize do
        @instance ||= new  # Race condition possible
      end
    end
  end

  # ✅ GOOD: Proper resource cleanup
  class ResourceManager
    def initialize
      @resources = {}
      @mutex = Mutex.new
    end

    def acquire_resource(id)
      @mutex.synchronize do
        @resources[id] ||= expensive_resource_creation(id)
      end
    end

    def release_resource(id)
      @mutex.synchronize do
        resource = @resources.delete(id)
        resource&.cleanup
      end
    end

    def cleanup_all
      @mutex.synchronize do
        @resources.each_value(&:cleanup)
        @resources.clear
      end
    end

    private

    def expensive_resource_creation(id)
      OpenStruct.new(id: id, cleanup: -> { puts "Cleaning up resource #{id}" })
    end
  end

  def self.demonstrate_patterns
    # Thread-safe singleton
    singleton1 = ThreadSafeSingleton.instance
    singleton2 = ThreadSafeSingleton.instance
    puts "Singleton works: #{singleton1.object_id == singleton2.object_id}"

    # Immutable counter
    counter = ImmutableCounter.new
    new_counter = counter.increment
    puts "Original: #{counter.value}, New: #{new_counter.value}"

    # Thread-local cache
    threads = 3.times.map do |i|
      Thread.new do
        ThreadLocalCache.set(:id, i)
        sleep(0.1)
        puts "Thread #{i} cached value: #{ThreadLocalCache.get(:id)}"
      end
    end
    threads.each(&:join)

    # Resource management
    manager = ResourceManager.new
    resource = manager.acquire_resource(:db_connection)
    manager.release_resource(:db_connection)
  end
end

ThreadingBestPractices.demonstrate_patterns

✅ Threading Best Practices

• Minimize shared state: Use immutable objects and message passing
• Use higher-level abstractions: Queue, ThreadPool, concurrent-ruby gems
• Always use synchronization: for shared mutable state
• Avoid nested locks: to prevent deadlocks
• Handle exceptions: in threads to prevent silent failures
• Use thread-local storage: for per-thread state
• Profile and test: threading code thoroughly

❌ Common Threading Mistakes

• Unprotected shared state: Always synchronize access to mutable shared data
• Deadlocks: Acquire locks in consistent order, use timeouts
• Resource leaks: Always clean up threads and resources
• Ignoring exceptions: Handle exceptions in threads explicitly
• Over-threading: More threads ≠ better performance