Ruby Logo

Async Patterns & Best Practices

Master producer-consumer patterns, thread pools, synchronization primitives, and concurrency best practices for scalable applications.

Home → Ruby → Async Patterns & Best Practices

Async Patterns & Best Practices

Master essential concurrency patterns including producer-consumer, thread pools, and synchronization primitives for building robust, scalable Ruby applications.

Producer-Consumer Pattern

The producer-consumer pattern decouples data generation from data processing, allowing for flexible, scalable architectures.

Basic Producer-Consumer with Queue

require 'thread'

class ProducerConsumer
  def initialize(queue_size = 10)
    @queue = SizedQueue.new(queue_size)
    @producers = []
    @consumers = []
    @running = false
  end

  def start
    @running = true
  end

  def stop
    @running = false
    @producers.each(&:join)

    # Signal consumers to stop
    @consumers.size.times { @queue << :stop }
    @consumers.each(&:join)
  end

  def add_producer(name, &block)
    producer = Thread.new do
      Thread.current.name = "Producer-#{name}"

      while @running
        begin
          item = block.call
          @queue << item if item
          sleep(0.1)  # Simulate production time
        rescue => e
          puts "Producer #{name} error: #{e.message}"
        end
      end

      puts "Producer #{name} stopped"
    end

    @producers << producer
  end

  def add_consumer(name, &block)
    consumer = Thread.new do
      Thread.current.name = "Consumer-#{name}"

      loop do
        item = @queue.pop
        break if item == :stop

        begin
          block.call(item)
        rescue => e
          puts "Consumer #{name} error: #{e.message}"
        end
      end

      puts "Consumer #{name} stopped"
    end

    @consumers << consumer
  end

  def queue_size
    @queue.size
  end

  def queue_stats
    {
      size: @queue.size,
      max_size: @queue.max,
      num_producers: @producers.size,
      num_consumers: @consumers.size
    }
  end
end

# Usage example
system = ProducerConsumer.new(5)

# Add producers
system.add_producer("DataGenerator") do
  "Data-#{Time.now.to_f}"
end

system.add_producer("RandomNumbers") do
  rand(1..100)
end

# Add consumers
system.add_consumer("Logger") do |item|
  puts "Logged: #{item}"
end

system.add_consumer("Processor") do |item|
  puts "Processed: #{item.inspect}"
  sleep(0.05)  # Simulate processing time
end

system.start

# Let it run for a bit
sleep(2)
puts "Queue stats: #{system.queue_stats}"

sleep(1)
system.stop

Advanced Producer-Consumer with Priorities

class PriorityProducerConsumer
  def initialize
    @high_priority_queue = Queue.new
    @normal_priority_queue = Queue.new
    @low_priority_queue = Queue.new
    @consumers = []
    @running = false
    @stats = {
      high: 0,
      normal: 0,
      low: 0,
      processed: 0
    }
    @stats_mutex = Mutex.new
  end

  def start
    @running = true
  end

  def stop
    @running = false
    # Send stop signals to all queues
    [@high_priority_queue, @normal_priority_queue, @low_priority_queue].each do |queue|
      @consumers.size.times { queue << { type: :stop } }
    end
    @consumers.each(&:join)
  end

  def produce(priority, data)
    return unless @running

    task = {
      id: SecureRandom.uuid,
      data: data,
      priority: priority,
      created_at: Time.now
    }

    queue = case priority
            when :high
              @high_priority_queue
            when :normal
              @normal_priority_queue
            when :low
              @low_priority_queue
            else
              @normal_priority_queue
            end

    queue << task

    @stats_mutex.synchronize do
      @stats[priority] += 1
    end
  end

  def add_consumer(name)
    consumer = Thread.new do
      Thread.current.name = "Consumer-#{name}"

      while @running
        task = next_task
        break if task[:type] == :stop

        begin
          process_task(task)

          @stats_mutex.synchronize do
            @stats[:processed] += 1
          end
        rescue => e
          puts "Consumer #{name} error processing task #{task[:id]}: #{e.message}"
        end
      end

      puts "Consumer #{name} stopped"
    end

    @consumers << consumer
  end

  def stats
    @stats_mutex.synchronize { @stats.dup }
  end

  private

  def next_task
    # Check high priority first
    return @high_priority_queue.pop(true) unless @high_priority_queue.empty?

    # Then normal priority
    return @normal_priority_queue.pop(true) unless @normal_priority_queue.empty?

    # Finally low priority
    return @low_priority_queue.pop(true) unless @low_priority_queue.empty?

    # If all queues are empty, block on the high priority queue
    @high_priority_queue.pop
  rescue ThreadError
    # Non-blocking pop failed, try again
    retry
  end

  def process_task(task)
    processing_time = case task[:priority]
                     when :high
                       0.01  # High priority tasks are processed quickly
                     when :normal
                       0.05
                     when :low
                       0.1   # Low priority tasks take longer
                     end

    sleep(processing_time)
    puts "Processed #{task[:priority]} priority task: #{task[:data]} (ID: #{task[:id][0..8]})"
  end
end

# Usage example
priority_system = PriorityProducerConsumer.new

# Add consumers
3.times { |i| priority_system.add_consumer(i + 1) }

priority_system.start

# Produce tasks with different priorities
10.times do |i|
  priority_system.produce(:low, "Low priority task #{i}")
  priority_system.produce(:normal, "Normal task #{i}")

  if i % 3 == 0
    priority_system.produce(:high, "URGENT: High priority task #{i}")
  end

  sleep(0.05)
end

sleep(2)
puts "Final stats: #{priority_system.stats}"
priority_system.stop

Thread Pool Implementation

Thread pools manage a fixed number of worker threads to efficiently handle multiple tasks without the overhead of creating/destroying threads.

Robust Thread Pool Implementation

class ThreadPool
  attr_reader :size, :active_count, :pending_count

  def initialize(size = 5)
    @size = size
    @queue = Queue.new
    @threads = []
    @active_count = 0
    @active_mutex = Mutex.new
    @shutdown = false
    @stats = {
      completed: 0,
      failed: 0,
      total_submitted: 0
    }
    @stats_mutex = Mutex.new

    create_threads
  end

  def submit(&block)
    raise "ThreadPool is shutdown" if @shutdown

    task_id = SecureRandom.uuid
    task = {
      id: task_id,
      block: block,
      submitted_at: Time.now
    }

    @queue << task

    @stats_mutex.synchronize do
      @stats[:total_submitted] += 1
    end

    task_id
  end

  def submit_with_callback(&block)
    promise = Promise.new

    submit do
      begin
        result = block.call
        promise.fulfill(result)
      rescue => e
        promise.reject(e)
      end
    end

    promise
  end

  def shutdown(wait: true)
    @shutdown = true

    # Signal all threads to stop
    @size.times { @queue << :shutdown }

    @threads.each(&:join) if wait
  end

  def pending_count
    @queue.size
  end

  def stats
    @stats_mutex.synchronize { @stats.dup }
  end

  def healthy?
    !@shutdown && @threads.all?(&:alive?)
  end

  private

  def create_threads
    @size.times do |i|
      thread = Thread.new do
        Thread.current.name = "ThreadPool-Worker-#{i}"
        worker_loop
      end

      @threads << thread
    end
  end

  def worker_loop
    loop do
      task = @queue.pop
      break if task == :shutdown

      increment_active_count

      begin
        start_time = Time.now
        task[:block].call
        duration = Time.now - start_time

        @stats_mutex.synchronize do
          @stats[:completed] += 1
        end

        puts "Task #{task[:id][0..8]} completed in #{duration.round(3)}s"
      rescue => e
        @stats_mutex.synchronize do
          @stats[:failed] += 1
        end

        puts "Task #{task[:id][0..8]} failed: #{e.message}"
      ensure
        decrement_active_count
      end
    end
  end

  def increment_active_count
    @active_mutex.synchronize { @active_count += 1 }
  end

  def decrement_active_count
    @active_mutex.synchronize { @active_count -= 1 }
  end
end

# Promise class for async results
class Promise
  def initialize
    @mutex = Mutex.new
    @condition = ConditionVariable.new
    @fulfilled = false
    @rejected = false
    @value = nil
    @reason = nil
  end

  def fulfill(value)
    @mutex.synchronize do
      return if @fulfilled || @rejected

      @value = value
      @fulfilled = true
      @condition.broadcast
    end
  end

  def reject(reason)
    @mutex.synchronize do
      return if @fulfilled || @rejected

      @reason = reason
      @rejected = true
      @condition.broadcast
    end
  end

  def value(timeout = nil)
    @mutex.synchronize do
      wait_until = timeout ? Time.now + timeout : nil

      while !@fulfilled && !@rejected
        if timeout
          remaining = wait_until - Time.now
          raise "Promise timeout" if remaining <= 0
          @condition.wait(@mutex, remaining)
        else
          @condition.wait(@mutex)
        end
      end

      raise @reason if @rejected
      @value
    end
  end

  def fulfilled?
    @fulfilled
  end

  def rejected?
    @rejected
  end
end

# Usage example
pool = ThreadPool.new(3)

# Submit regular tasks
5.times do |i|
  pool.submit do
    sleep(rand(0.1..0.5))
    puts "Task #{i} completed by #{Thread.current.name}"
  end
end

# Submit task with promise
promise = pool.submit_with_callback do
  sleep(0.2)
  "Important result: #{rand(100)}"
end

# Wait for the result
begin
  result = promise.value(1.0)  # 1 second timeout
  puts "Promise result: #{result}"
rescue => e
  puts "Promise failed: #{e.message}"
end

sleep(1)
puts "Pool stats: #{pool.stats}"
puts "Active threads: #{pool.active_count}"
puts "Pending tasks: #{pool.pending_count}"

pool.shutdown

Advanced Synchronization Primitives

Read-Write Lock Implementation

class ReadWriteLock
  def initialize
    @mutex = Mutex.new
    @readers_condition = ConditionVariable.new
    @writers_condition = ConditionVariable.new
    @readers = 0
    @writers = 0
    @waiting_writers = 0
  end

  def read_lock
    @mutex.synchronize do
      while @writers > 0 || @waiting_writers > 0
        @readers_condition.wait(@mutex)
      end
      @readers += 1
    end
  end

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

  def write_lock
    @mutex.synchronize do
      @waiting_writers += 1
      while @readers > 0 || @writers > 0
        @writers_condition.wait(@mutex)
      end
      @waiting_writers -= 1
      @writers += 1
    end
  end

  def write_unlock
    @mutex.synchronize do
      @writers -= 1
      @writers_condition.signal
      @readers_condition.broadcast
    end
  end

  def with_read_lock(&block)
    read_lock
    begin
      block.call
    ensure
      read_unlock
    end
  end

  def with_write_lock(&block)
    write_lock
    begin
      block.call
    ensure
      write_unlock
    end
  end

  def stats
    @mutex.synchronize do
      {
        readers: @readers,
        writers: @writers,
        waiting_writers: @waiting_writers
      }
    end
  end
end

# Thread-safe cache using read-write lock
class ThreadSafeCache
  def initialize
    @cache = {}
    @rwlock = ReadWriteLock.new
  end

  def get(key)
    @rwlock.with_read_lock do
      @cache[key]
    end
  end

  def set(key, value)
    @rwlock.with_write_lock do
      @cache[key] = value
    end
  end

  def delete(key)
    @rwlock.with_write_lock do
      @cache.delete(key)
    end
  end

  def keys
    @rwlock.with_read_lock do
      @cache.keys.dup
    end
  end

  def size
    @rwlock.with_read_lock do
      @cache.size
    end
  end

  def clear
    @rwlock.with_write_lock do
      @cache.clear
    end
  end
end

# Usage example
cache = ThreadSafeCache.new

# Create reader threads
readers = 5.times.map do |i|
  Thread.new do
    Thread.current.name = "Reader-#{i}"

    10.times do |j|
      key = "key#{rand(10)}"
      value = cache.get(key)
      puts "#{Thread.current.name} read #{key}: #{value}"
      sleep(0.01)
    end
  end
end

# Create writer threads
writers = 2.times.map do |i|
  Thread.new do
    Thread.current.name = "Writer-#{i}"

    5.times do |j|
      key = "key#{rand(10)}"
      value = "value-#{i}-#{j}"
      cache.set(key, value)
      puts "#{Thread.current.name} wrote #{key}: #{value}"
      sleep(0.05)
    end
  end
end

# Wait for all threads
(readers + writers).each(&:join)

puts "Final cache size: #{cache.size}"
puts "Cache keys: #{cache.keys}"

Semaphore Implementation

class Semaphore
  def initialize(initial_count)
    @count = initial_count
    @mutex = Mutex.new
    @condition = ConditionVariable.new
  end

  def acquire(permits = 1)
    @mutex.synchronize do
      while @count < permits
        @condition.wait(@mutex)
      end
      @count -= permits
    end
  end

  def release(permits = 1)
    @mutex.synchronize do
      @count += permits
      @condition.broadcast
    end
  end

  def with_permit(permits = 1, &block)
    acquire(permits)
    begin
      block.call
    ensure
      release(permits)
    end
  end

  def available_permits
    @mutex.synchronize { @count }
  end

  def try_acquire(permits = 1, timeout: nil)
    @mutex.synchronize do
      if timeout
        deadline = Time.now + timeout
        while @count < permits && Time.now < deadline
          remaining = deadline - Time.now
          return false if remaining <= 0
          @condition.wait(@mutex, remaining)
        end
      end

      if @count >= permits
        @count -= permits
        true
      else
        false
      end
    end
  end
end

# Connection pool using semaphore
class ConnectionPool
  def initialize(max_connections = 5)
    @connections = max_connections.times.map { |i| "Connection-#{i}" }
    @available = Queue.new
    @connections.each { |conn| @available << conn }
    @semaphore = Semaphore.new(max_connections)
    @in_use = Set.new
    @mutex = Mutex.new
  end

  def with_connection(timeout: 5, &block)
    if @semaphore.try_acquire(1, timeout: timeout)
      connection = @available.pop

      @mutex.synchronize do
        @in_use.add(connection)
      end

      begin
        puts "Acquired connection: #{connection}"
        result = block.call(connection)
        result
      ensure
        @mutex.synchronize do
          @in_use.delete(connection)
        end

        @available << connection
        @semaphore.release
        puts "Released connection: #{connection}"
      end
    else
      raise "Connection timeout: no connections available within #{timeout} seconds"
    end
  end

  def stats
    @mutex.synchronize do
      {
        total_connections: @connections.size,
        available: @semaphore.available_permits,
        in_use: @in_use.size
      }
    end
  end
end

# Usage example
pool = ConnectionPool.new(3)

# Create multiple threads trying to use connections
threads = 8.times.map do |i|
  Thread.new do
    Thread.current.name = "Client-#{i}"

    begin
      pool.with_connection(timeout: 2) do |conn|
        puts "#{Thread.current.name} using #{conn}"
        sleep(rand(0.5..1.0))  # Simulate work
        "Result from #{Thread.current.name}"
      end
    rescue => e
      puts "#{Thread.current.name} failed: #{e.message}"
    end
  end
end

# Monitor pool stats
monitor = Thread.new do
  5.times do
    puts "Pool stats: #{pool.stats}"
    sleep(0.5)
  end
end

threads.each(&:join)
monitor.join

Best Practices & Patterns

✅ Do's

  • Use bounded queues to prevent memory exhaustion
  • Implement proper error handling in all concurrent code
  • Monitor thread pool health and provide observability
  • Use timeouts to prevent indefinite blocking
  • Prefer composition over inheritance for concurrent classes
  • Test concurrent code thoroughly under load
  • Document thread safety guarantees clearly

❌ Don'ts

  • Don't create unbounded threads - use thread pools
  • Don't ignore exceptions in background threads
  • Don't share mutable state without synchronization
  • Don't use sleep for synchronization
  • Don't mix blocking and non-blocking operations carelessly
  • Don't forget to clean up resources properly
  • Don't make assumptions about execution order

Performance Monitoring Pattern

class ConcurrencyMonitor
  def initialize
    @metrics = {
      thread_count: 0,
      active_threads: 0,
      completed_tasks: 0,
      failed_tasks: 0,
      average_task_duration: 0.0,
      queue_sizes: {}
    }
    @mutex = Mutex.new
    @task_durations = []
  end

  def track_thread_creation
    @mutex.synchronize do
      @metrics[:thread_count] += 1
    end
  end

  def track_thread_start
    @mutex.synchronize do
      @metrics[:active_threads] += 1
    end
  end

  def track_thread_end
    @mutex.synchronize do
      @metrics[:active_threads] -= 1
    end
  end

  def track_task_completion(duration)
    @mutex.synchronize do
      @metrics[:completed_tasks] += 1
      @task_durations << duration

      # Keep only last 1000 durations for average calculation
      @task_durations = @task_durations.last(1000)
      @metrics[:average_task_duration] = @task_durations.sum / @task_durations.size
    end
  end

  def track_task_failure
    @mutex.synchronize do
      @metrics[:failed_tasks] += 1
    end
  end

  def track_queue_size(queue_name, size)
    @mutex.synchronize do
      @metrics[:queue_sizes][queue_name] = size
    end
  end

  def snapshot
    @mutex.synchronize { @metrics.dup }
  end

  def report
    stats = snapshot

    puts "=== Concurrency Report ==="
    puts "Total threads created: #{stats[:thread_count]}"
    puts "Currently active threads: #{stats[:active_threads]}"
    puts "Completed tasks: #{stats[:completed_tasks]}"
    puts "Failed tasks: #{stats[:failed_tasks]}"
    puts "Average task duration: #{stats[:average_task_duration].round(3)}s"

    if stats[:queue_sizes].any?
      puts "Queue sizes:"
      stats[:queue_sizes].each do |name, size|
        puts "  #{name}: #{size}"
      end
    end

    success_rate = if stats[:completed_tasks] + stats[:failed_tasks] > 0
                    (stats[:completed_tasks].to_f / (stats[:completed_tasks] + stats[:failed_tasks]) * 100).round(2)
                  else
                    0.0
                  end
    puts "Success rate: #{success_rate}%"
    puts "========================="
  end
end

🎯 Key Takeaways

  • • Producer-Consumer: Decouples data generation from processing, enabling scalable architectures
  • • Thread Pools: Manage thread lifecycle efficiently, preventing resource exhaustion
  • • Synchronization: Use appropriate primitives (Mutex, ReadWriteLock, Semaphore) for different scenarios
  • • Error Handling: Always handle exceptions gracefully in concurrent environments
  • • Monitoring: Track performance metrics and resource usage for production systems
  • • Testing: Concurrent code requires specialized testing approaches and stress testing

Quick Navigation

Related Topics

Video Tutorial

Watch and learn async patterns & best practices

Pro Tip: After reading through the content above, watch this video to reinforce your understanding and see the concepts in action!