Async Gem

Build high-performance applications with modern async patterns, event loops, reactive programming, and concurrent I/O operations.

Home → Ruby → Async Gem

Async Gem

Explore modern async patterns with the Async gem, featuring event loops, reactive programming, and high-performance concurrent I/O operations for scalable Ruby applications.

Introduction to Async Gem

The async gem provides a modern, high-performance approach to concurrent programming in Ruby. Built around an event loop and fiber-based cooperative multitasking, it enables efficient I/O operations and reactive programming patterns without the complexity of traditional threading.

Installation:

# Gemfile
gem 'async', '~> 2.6'
gem 'async-http', '~> 0.60'  # For HTTP operations
gem 'async-io', '~> 1.34'    # For advanced I/O

# Install
bundle install

Key Benefits:

High Performance: Handles thousands of concurrent connections efficiently
Low Overhead: Fiber-based, avoiding thread creation costs
Composable: Easy to build complex async workflows
Integrated: Works well with existing Ruby code and frameworks

Basic Async Operations

Event Loop and Tasks

require 'async'

class AsyncBasics
  def self.simple_async_task
    puts "Starting async demonstration..."

    Async do
      puts "Task 1 starting"
      Async::Task.current.sleep(1)
      puts "Task 1 completed"
    end

    puts "Main thread continues immediately"
  end

  def self.multiple_concurrent_tasks
    Async do |task|
      # Start multiple concurrent tasks
      tasks = []

      5.times do |i|
        tasks << task.async do
          puts "Task #{i} starting at #{Time.now}"
          Async::Task.current.sleep(rand(0.5..2.0))
          puts "Task #{i} completed at #{Time.now}"
          "Result #{i}"
        end
      end

      # Wait for all tasks to complete
      results = tasks.map(&:wait)
      puts "All tasks completed with results: #{results}"
    end
  end

  def self.nested_async_tasks
    Async do |parent_task|
      puts "Parent task starting"

      # Create child tasks
      child1 = parent_task.async do
        puts "Child 1 working..."
        Async::Task.current.sleep(0.5)

        # Nested async operation
        nested_result = Async do
          puts "Nested operation in child 1"
          Async::Task.current.sleep(0.2)
          "Nested result"
        end

        "Child 1 completed with: #{nested_result}"
      end

      child2 = parent_task.async do
        puts "Child 2 working..."
        Async::Task.current.sleep(0.3)
        "Child 2 completed"
      end

      # Wait for children
      result1 = child1.wait
      result2 = child2.wait

      puts "Parent completed: #{result1}, #{result2}"
    end
  end
end

AsyncBasics.simple_async_task
AsyncBasics.multiple_concurrent_tasks
AsyncBasics.nested_async_tasks

Error Handling and Timeouts

class AsyncErrorHandling
  def self.basic_error_handling
    Async do |task|
      # Task that will fail
      failing_task = task.async do
        puts "Starting risky operation..."
        Async::Task.current.sleep(0.5)
        raise "Something went wrong!"
      end

      # Task that will succeed
      success_task = task.async do
        puts "Starting safe operation..."
        Async::Task.current.sleep(0.3)
        "Success!"
      end

      # Handle results
      begin
        result = failing_task.wait
        puts "Unexpected success: #{result}"
      rescue => e
        puts "Caught expected error: #{e.message}"
      end

      begin
        result = success_task.wait
        puts "Success result: #{result}"
      rescue => e
        puts "Unexpected error: #{e.message}"
      end
    end
  end

  def self.timeout_handling
    Async do |task|
      # Task with timeout
      begin
        result = task.with_timeout(1.0) do
          puts "Starting operation that might be slow..."
          Async::Task.current.sleep(2.0)  # This will timeout
          "Operation completed"
        end
        puts "Result: #{result}"
      rescue Async::TimeoutError
        puts "Operation timed out!"
      end

      # Task that completes within timeout
      begin
        result = task.with_timeout(2.0) do
          puts "Starting fast operation..."
          Async::Task.current.sleep(0.5)
          "Fast operation completed"
        end
        puts "Result: #{result}"
      rescue Async::TimeoutError
        puts "Fast operation timed out (unexpected)"
      end
    end
  end

  def self.barrier_pattern
    Async do |task|
      # Create a barrier to wait for multiple tasks
      barrier = Async::Barrier.new

      # Add tasks to barrier
      barrier.async do
        puts "Barrier task 1 starting"
        Async::Task.current.sleep(rand(0.5..1.5))
        puts "Barrier task 1 completed"
        "Result 1"
      end

      barrier.async do
        puts "Barrier task 2 starting"
        Async::Task.current.sleep(rand(0.5..1.5))
        puts "Barrier task 2 completed"
        "Result 2"
      end

      barrier.async do
        puts "Barrier task 3 starting"
        Async::Task.current.sleep(rand(0.5..1.5))
        # This task will fail
        raise "Barrier task 3 failed"
      end

      # Wait for all tasks (will raise if any fail)
      begin
        results = barrier.wait
        puts "All barrier tasks completed: #{results}"
      rescue => e
        puts "Barrier failed: #{e.message}"
        puts "Completed tasks: #{barrier.finished_tasks.map(&:result)}"
      end
    end
  end
end

AsyncErrorHandling.basic_error_handling
AsyncErrorHandling.timeout_handling
AsyncErrorHandling.barrier_pattern

HTTP Operations with Async

Concurrent HTTP Requests

require 'async/http'

class AsyncHTTP
  def self.single_http_request
    Async do
      # Create HTTP client
      Async::HTTP::Internet.new do |internet|
        puts "Making HTTP request..."

        # Note: Using httpbin.org for demonstration
        response = internet.get("https://httpbin.org/delay/1")

        if response.success?
          body = response.read
          puts "Response received: #{body[0..100]}..."
        else
          puts "Request failed: #{response.status}"
        end
      end
    end
  end

  def self.concurrent_http_requests
    urls = [
      "https://httpbin.org/delay/1",
      "https://httpbin.org/json",
      "https://httpbin.org/uuid",
      "https://httpbin.org/ip",
      "https://httpbin.org/user-agent"
    ]

    Async do |task|
      Async::HTTP::Internet.new do |internet|
        # Start all requests concurrently
        request_tasks = urls.map.with_index do |url, index|
          task.async do
            puts "Starting request #{index + 1} to #{url}"
            start_time = Time.now

            response = internet.get(url)
            duration = Time.now - start_time

            {
              url: url,
              status: response.status,
              duration: duration.round(3),
              success: response.success?
            }
          end
        end

        # Wait for all requests to complete
        results = request_tasks.map(&:wait)

        puts "\n=== Request Results ==="
        results.each_with_index do |result, index|
          status = result[:success] ? "✓" : "✗"
          puts "#{status} Request #{index + 1}: #{result[:status]} (#{result[:duration]}s)"
        end

        total_time = results.map { |r| r[:duration] }.sum
        puts "Total concurrent time: #{total_time.round(3)}s"
        puts "Requests would have taken ~#{results.size}s sequentially"
      end
    end
  end

  def self.http_with_retries
    Async do
      Async::HTTP::Internet.new do |internet|
        url = "https://httpbin.org/status/500"  # This will return 500 error
        max_retries = 3

        (1..max_retries).each do |attempt|
          puts "Attempt #{attempt}/#{max_retries} for #{url}"

          begin
            response = internet.get(url)

            if response.success?
              puts "Request succeeded on attempt #{attempt}"
              body = response.read
              puts "Response: #{body[0..100]}..."
              break
            else
              puts "Request failed with status: #{response.status}"

              if attempt < max_retries
                puts "Retrying in #{attempt} seconds..."
                Async::Task.current.sleep(attempt)
              else
                puts "Max retries reached, giving up"
              end
            end
          rescue => e
            puts "Request error on attempt #{attempt}: #{e.message}"

            if attempt < max_retries
              puts "Retrying after error..."
              Async::Task.current.sleep(attempt)
            else
              puts "Max retries reached after errors"
            end
          end
        end
      end
    end
  end
end

AsyncHTTP.single_http_request
AsyncHTTP.concurrent_http_requests
AsyncHTTP.http_with_retries

HTTP Server with Async

require 'async/http/server'
require 'async/http/endpoint'

class AsyncHTTPServer
  def self.create_simple_server
    # Define endpoints
    endpoint = Async::HTTP::Endpoint.parse("http://localhost:9292")

    # Create server app
    app = lambda do |request|
      path = request.path
      method = request.method

      case path
      when "/"
        [200, {}, ["Hello from Async HTTP Server!\n"]]
      when "/async"
        # Demonstrate async operation in handler
        Async do
          # Simulate async work
          Async::Task.current.sleep(0.5)
          [200, {"Content-Type" => "application/json"}, ['{"message": "Async response", "timestamp": "' + Time.now.to_s + '"}']]
        end
      when "/slow"
        # Simulate slow endpoint
        Async::Task.current.sleep(2)
        [200, {}, ["This was a slow response\n"]]
      when "/error"
        [500, {}, ["Internal Server Error\n"]]
      else
        [404, {}, ["Not Found: #{path}\n"]]
      end
    end

    puts "Starting HTTP server on #{endpoint}"
    puts "Try: curl http://localhost:9292/"
    puts "Try: curl http://localhost:9292/async"
    puts "Try: curl http://localhost:9292/slow"

    # Note: This would run indefinitely in a real application
    Async do |task|
      server = Async::HTTP::Server.new(app, endpoint)

      # Run server in background
      server_task = task.async do
        server.run
      end

      # Simulate some requests to our own server
      task.sleep(1)  # Give server time to start

      Async::HTTP::Internet.new do |internet|
        # Test the server
        response = internet.get("http://localhost:9292/")
        puts "Server response: #{response.read}"

        response = internet.get("http://localhost:9292/async")
        puts "Async endpoint response: #{response.read}"
      end

      # Stop server after demo
      task.sleep(1)
      server_task.stop
    end
  end

  def self.websocket_like_behavior
    # Simulate WebSocket-like persistent connections
    Async do |task|
      connections = []

      # Simulate multiple clients connecting
      5.times do |client_id|
        connection_task = task.async do
          puts "Client #{client_id} connected"

          begin
            # Simulate persistent connection behavior
            10.times do |message_num|
              # Simulate receiving/sending messages
              Async::Task.current.sleep(rand(0.1..0.5))
              puts "Client #{client_id} message #{message_num}: Hello server!"

              # Simulate server response
              Async::Task.current.sleep(0.1)
              puts "Server to client #{client_id}: Message #{message_num} received"
            end
          rescue => e
            puts "Client #{client_id} error: #{e.message}"
          ensure
            puts "Client #{client_id} disconnected"
          end
        end

        connections << connection_task
      end

      # Wait for all connections to complete
      connections.each(&:wait)
      puts "All clients disconnected"
    end
  end
end

AsyncHTTPServer.create_simple_server
AsyncHTTPServer.websocket_like_behavior

Reactive Programming Patterns

Event Streams and Processing

class AsyncReactive
  def self.event_stream_processor
    Async do |task|
      # Create a simple event stream using a queue
      event_queue = Async::Queue.new

      # Event producer
      producer = task.async do
        events = [
          { type: :user_login, user_id: 123, timestamp: Time.now },
          { type: :page_view, user_id: 123, page: "/dashboard", timestamp: Time.now },
          { type: :user_logout, user_id: 123, timestamp: Time.now },
          { type: :user_login, user_id: 456, timestamp: Time.now },
          { type: :error, message: "Database connection failed", timestamp: Time.now }
        ]

        events.each do |event|
          puts "Producing event: #{event[:type]}"
          event_queue.enqueue(event)
          Async::Task.current.sleep(0.5)  # Simulate event timing
        end

        event_queue.enqueue(:done)  # Signal completion
      end

      # Event processors
      login_processor = task.async do
        while true
          event = event_queue.dequeue
          break if event == :done

          if event[:type] == :user_login
            puts "  → Login processor: User #{event[:user_id]} logged in"
            # Simulate processing
            Async::Task.current.sleep(0.1)
          end

          # Put event back for other processors
          event_queue.enqueue(event) unless event == :done
        end
      end

      analytics_processor = task.async do
        user_sessions = {}

        while true
          event = event_queue.dequeue
          break if event == :done

          case event[:type]
          when :user_login
            user_sessions[event[:user_id]] = { login_time: event[:timestamp], page_views: 0 }
          when :page_view
            if session = user_sessions[event[:user_id]]
              session[:page_views] += 1
              puts "  → Analytics: User #{event[:user_id]} has #{session[:page_views]} page views"
            end
          when :user_logout
            if session = user_sessions.delete(event[:user_id])
              duration = event[:timestamp] - session[:login_time]
              puts "  → Analytics: User #{event[:user_id]} session duration: #{duration.round(2)}s"
            end
          end
        end

        puts "Analytics processor finished"
      end

      # Wait for producer to finish
      producer.wait

      # Give processors time to finish
      task.sleep(1)
    end
  end

  def self.async_pipeline
    Async do |task|
      # Create processing pipeline stages
      stage1_queue = Async::Queue.new
      stage2_queue = Async::Queue.new
      stage3_queue = Async::Queue.new

      # Stage 1: Data ingestion
      stage1 = task.async do
        data_items = (1..10).map { |i| { id: i, raw_data: "raw_#{i}" } }

        data_items.each do |item|
          puts "Stage 1: Processing item #{item[:id]}"
          processed_item = item.merge(stage1_processed: true, processed_at: Time.now)
          stage1_queue.enqueue(processed_item)
          Async::Task.current.sleep(0.1)
        end

        stage1_queue.enqueue(:done)
        puts "Stage 1 completed"
      end

      # Stage 2: Data transformation
      stage2 = task.async do
        while true
          item = stage1_queue.dequeue
          break if item == :done

          puts "Stage 2: Transforming item #{item[:id]}"
          Async::Task.current.sleep(0.15)  # Simulate processing time

          transformed_item = item.merge(
            transformed_data: item[:raw_data].upcase,
            stage2_processed: true
          )

          stage2_queue.enqueue(transformed_item)
        end

        stage2_queue.enqueue(:done)
        puts "Stage 2 completed"
      end

      # Stage 3: Data output
      stage3 = task.async do
        results = []

        while true
          item = stage2_queue.dequeue
          break if item == :done

          puts "Stage 3: Finalizing item #{item[:id]}"
          Async::Task.current.sleep(0.05)

          final_item = item.merge(
            finalized: true,
            completed_at: Time.now
          )

          results << final_item
        end

        puts "Stage 3 completed with #{results.size} items"
        puts "Sample result: #{results.first}"
        results
      end

      # Wait for pipeline to complete
      stage1.wait
      stage2.wait
      final_results = stage3.wait

      puts "Pipeline processing completed"
      final_results
    end
  end

  def self.backpressure_handling
    Async do |task|
      # Bounded queue to demonstrate backpressure
      bounded_queue = Async::Queue.new(capacity: 3)

      # Fast producer
      producer = task.async do
        20.times do |i|
          puts "Producer: Trying to send item #{i}"
          bounded_queue.enqueue("item_#{i}")
          puts "Producer: Sent item #{i}"
        end
        bounded_queue.enqueue(:done)
        puts "Producer finished"
      end

      # Slow consumer
      consumer = task.async do
        processed = 0

        while true
          item = bounded_queue.dequeue
          break if item == :done

          puts "  Consumer: Processing #{item}"
          Async::Task.current.sleep(0.5)  # Slow processing
          processed += 1
          puts "  Consumer: Finished processing #{item} (#{processed} total)"
        end

        puts "Consumer finished processing #{processed} items"
      end

      # Monitor queue size
      monitor = task.async do
        while producer.running? || consumer.running?
          puts "    Queue size: #{bounded_queue.size}/#{bounded_queue.capacity}"
          Async::Task.current.sleep(0.2)
        end
      end

      producer.wait
      consumer.wait
      monitor.stop
    end
  end
end

AsyncReactive.event_stream_processor
AsyncReactive.async_pipeline
AsyncReactive.backpressure_handling

Real-World Application Example

Async Web Crawler

class AsyncWebCrawler
  def initialize(max_concurrent: 10, delay: 0.1)
    @max_concurrent = max_concurrent
    @delay = delay
    @visited_urls = Set.new
    @results = {}
    @semaphore = Async::Semaphore.new(max_concurrent)
  end

  def crawl(start_urls, max_depth: 2)
    Async do |task|
      url_queue = Async::Queue.new

      # Seed the queue
      start_urls.each { |url| url_queue.enqueue({ url: url, depth: 0 }) }

      # Create crawler workers
      workers = Array.new(@max_concurrent) do |worker_id|
        task.async do
          crawl_worker(worker_id, url_queue, max_depth)
        end
      end

      # Wait for all workers to complete
      workers.each(&:wait)

      {
        crawled_count: @visited_urls.size,
        results: @results,
        visited_urls: @visited_urls.to_a
      }
    end
  end

  private

  def crawl_worker(worker_id, url_queue, max_depth)
    puts "Worker #{worker_id} started"

    while true
      begin
        # Try to get next URL (with timeout to avoid infinite waiting)
        url_info = url_queue.dequeue
        break if url_info.nil?

        url = url_info[:url]
        depth = url_info[:depth]

        # Skip if already visited or max depth reached
        next if @visited_urls.include?(url) || depth > max_depth

        @visited_urls.add(url)

        # Rate limiting
        @semaphore.acquire do
          crawl_page(worker_id, url, depth, url_queue)
        end

        Async::Task.current.sleep(@delay)
      rescue => e
        puts "Worker #{worker_id} error: #{e.message}"
      end
    end

    puts "Worker #{worker_id} finished"
  end

  def crawl_page(worker_id, url, depth, url_queue)
    puts "Worker #{worker_id}: Crawling #{url} (depth: #{depth})"

    # Simulate HTTP request and parsing
    Async::Task.current.sleep(rand(0.1..0.5))

    # Simulate random success/failure
    if rand > 0.1  # 90% success rate
      # Simulate finding links
      found_links = generate_mock_links(url, depth)

      @results[url] = {
        status: :success,
        depth: depth,
        links_found: found_links.size,
        crawled_at: Time.now,
        worker_id: worker_id
      }

      # Add found links to queue if not at max depth
      if depth < 2  # max_depth is typically 2 in our example
        found_links.each do |link|
          url_queue.enqueue({ url: link, depth: depth + 1 })
        end
      end

      puts "  Worker #{worker_id}: Found #{found_links.size} links in #{url}"
    else
      @results[url] = {
        status: :failed,
        depth: depth,
        error: "HTTP error",
        crawled_at: Time.now,
        worker_id: worker_id
      }

      puts "  Worker #{worker_id}: Failed to crawl #{url}"
    end
  end

  def generate_mock_links(base_url, depth)
    # Generate mock links based on the base URL
    return [] if depth >= 2  # Don't generate links at max depth

    num_links = rand(2..5)
    num_links.times.map do |i|
      "#{base_url}/page_#{depth}_#{i}"
    end
  end
end

# Real-world async service integration
class AsyncServiceIntegration
  def self.microservice_communication
    Async do |task|
      # Simulate multiple microservice calls
      services = {
        user_service: "https://httpbin.org/delay/0.5",
        order_service: "https://httpbin.org/delay/0.3",
        inventory_service: "https://httpbin.org/delay/0.7",
        payment_service: "https://httpbin.org/delay/0.4"
      }

      Async::HTTP::Internet.new do |internet|
        # Call all services concurrently
        service_tasks = services.map do |service_name, url|
          task.async do
            puts "Calling #{service_name}..."
            start_time = Time.now

            begin
              response = internet.get(url)
              duration = Time.now - start_time

              {
                service: service_name,
                success: response.success?,
                status: response.status,
                duration: duration.round(3)
              }
            rescue => e
              {
                service: service_name,
                success: false,
                error: e.message,
                duration: (Time.now - start_time).round(3)
              }
            end
          end
        end

        # Wait for all services
        results = service_tasks.map(&:wait)

        puts "\n=== Service Results ==="
        results.each do |result|
          status = result[:success] ? "✓" : "✗"
          puts "#{status} #{result[:service]}: #{result[:duration]}s"
        end

        total_time = results.map { |r| r[:duration] }.max
        puts "Total time (concurrent): #{total_time}s"
        puts "Would take ~#{results.map { |r| r[:duration] }.sum}s sequentially"
      end
    end
  end
end

# Usage examples
puts "=== Async Web Crawler Demo ==="
crawler = AsyncWebCrawler.new(max_concurrent: 3, delay: 0.1)
start_urls = ["https://example.com", "https://test.com"]

results = crawler.crawl(start_urls, max_depth: 1)
puts "Crawled #{results[:crawled_count]} URLs"
puts "Sample results:"
results[:results].first(3).each do |url, info|
  puts "  #{url}: #{info[:status]} (#{info[:links_found]} links)"
end

puts "\n=== Microservice Communication Demo ==="
AsyncServiceIntegration.microservice_communication

Performance and Best Practices

✅ Best Practices

Use for I/O-bound tasks: Async shines with network operations
Implement backpressure: Use bounded queues to prevent memory issues
Handle errors gracefully: Always wrap async operations in error handling
Set appropriate timeouts: Prevent hanging operations
Monitor fiber count: Too many can impact performance
Use barriers for coordination: Wait for multiple operations safely

⚠️ Common Pitfalls

Blocking operations: Avoid File.read, DB calls without async gems
CPU-intensive tasks: These don't benefit from async patterns
Forgetting error handling: Unhandled errors can crash the event loop
Infinite queues: Can lead to memory exhaustion
Missing timeouts: Operations can hang indefinitely
Synchronous gems: Many gems aren't async-aware

Performance Monitoring

class AsyncMonitoring
  def self.performance_metrics
    Async do |task|
      start_time = Time.now
      completed_tasks = 0
      active_tasks = 0

      # Monitor task execution
      monitor = task.async do
        while task.running?
          fiber_count = Fiber.current.instance_variable_get(:@scheduler)&.fiber_count || 0
          puts "Active fibers: #{fiber_count}, Completed: #{completed_tasks}, Active: #{active_tasks}"
          Async::Task.current.sleep(1)
        end
      end

      # Create multiple tasks to monitor
      tasks = 10.times.map do |i|
        task.async do
          active_tasks += 1
          puts "Task #{i} starting"

          # Simulate async work
          Async::Task.current.sleep(rand(0.5..2.0))

          completed_tasks += 1
          active_tasks -= 1
          puts "Task #{i} completed"
        end
      end

      # Wait for all tasks
      tasks.each(&:wait)

      total_time = Time.now - start_time
      puts "All tasks completed in #{total_time.round(2)}s"

      monitor.stop
    end
  end
end

AsyncMonitoring.performance_metrics

🎯 Key Takeaways

• Event Loop Architecture: Single-threaded, fiber-based concurrency for high-performance I/O
• Cooperative Multitasking: Tasks yield control voluntarily, avoiding thread synchronization issues
• Scalable I/O: Handle thousands of concurrent connections efficiently
• Reactive Patterns: Build event-driven, responsive applications with streams and pipelines
• HTTP Integration: Built-in support for concurrent HTTP operations and server development
• Error Resilience: Comprehensive error handling and timeout mechanisms for robust applications

Concurrent-Ruby Gem Quiz: Concurrency

Quick Navigation

Read Topic

Watch Video Tutorial

Video Tutorial

Watch and learn async gem

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

Ruby Language

Rails Framework

Production Ready

Async Gem

Async Gem

Introduction to Async Gem

Installation:

Key Benefits:

Basic Async Operations

Event Loop and Tasks

Error Handling and Timeouts

HTTP Operations with Async

Concurrent HTTP Requests

HTTP Server with Async

Reactive Programming Patterns

Event Streams and Processing

Real-World Application Example

Async Web Crawler

Performance and Best Practices

✅ Best Practices

⚠️ Common Pitfalls

Performance Monitoring

🎯 Key Takeaways

Quick Navigation

Related Topics

Video Tutorial