Master production-ready concurrency tools with the concurrent-ruby gem, featuring promises, actors, thread pools, and atomic operations for robust concurrent applications.
The concurrent-ruby gem provides industrial-strength concurrency primitives for Ruby applications.
It offers thread-safe data structures, synchronization primitives, and high-level abstractions that make
concurrent programming safer and more predictable.
# Gemfile
gem 'concurrent-ruby', '~> 1.2'
# Install
bundle installPromises and Futures provide elegant asynchronous programming patterns with built-in error handling and composition.
require 'concurrent-ruby'
class PromiseExamples
def self.basic_promise
# Create a promise that resolves with a value
promise = Concurrent::Promise.execute do
sleep(1)
"Hello from promise!"
end
puts "Promise state: #{promise.state}" # :pending
puts "Promise value: #{promise.value}" # Blocks until resolved
puts "Promise state: #{promise.state}" # :fulfilled
end
def self.promise_with_error_handling
# Promise that might fail
risky_promise = Concurrent::Promise.execute do
sleep(0.5)
raise "Something went wrong!" if rand > 0.5
"Success!"
end
# Chain error handling
safe_promise = risky_promise.rescue do |error|
puts "Caught error: #{error.message}"
"Default value"
end
puts "Final result: #{safe_promise.value}"
end
def self.promise_chaining
# Chain multiple asynchronous operations
result = Concurrent::Promise.execute do
puts "Step 1: Fetching user data"
sleep(0.2)
{ id: 123, name: "John Doe" }
end.then do |user|
puts "Step 2: Fetching user preferences for #{user[:name]}"
sleep(0.3)
user.merge(preferences: { theme: "dark", notifications: true })
end.then do |user_with_prefs|
puts "Step 3: Generating recommendations"
sleep(0.1)
user_with_prefs.merge(recommendations: ["Ruby", "Rails", "Concurrent Programming"])
end
puts "Final user data: #{result.value}"
end
def self.multiple_promises
# Execute multiple promises in parallel
promises = [
Concurrent::Promise.execute { sleep(0.3); "Result 1" },
Concurrent::Promise.execute { sleep(0.1); "Result 2" },
Concurrent::Promise.execute { sleep(0.2); "Result 3" }
]
# Wait for all to complete
results = promises.map(&:value)
puts "All results: #{results}"
# Or use zip for cleaner syntax
combined = Concurrent::Promise.zip(*promises)
puts "Combined results: #{combined.value}"
end
end
PromiseExamples.basic_promise
PromiseExamples.promise_with_error_handling
PromiseExamples.promise_chaining
PromiseExamples.multiple_promisesclass AdvancedPromisePatterns
def self.timeout_and_fallback
# Promise with timeout and fallback
slow_service = Concurrent::Promise.execute do
sleep(3) # Simulates slow service
"Slow response"
end
fast_fallback = Concurrent::Promise.execute do
sleep(0.5)
"Fast fallback"
end
# Race promises - first one to complete wins
result = Concurrent::Promise.any(slow_service, fast_fallback)
puts "Winner: #{result.value}"
end
def self.circuit_breaker_pattern
class ServiceClient
def initialize
@failure_count = 0
@last_failure_time = nil
@circuit_open = false
end
def call_service
if circuit_open?
return Concurrent::Promise.reject(StandardError.new("Circuit breaker open"))
end
Concurrent::Promise.execute do
# Simulate service call
if rand > 0.7 # 30% failure rate
@failure_count += 1
@last_failure_time = Time.now
@circuit_open = true if @failure_count >= 3
raise "Service unavailable"
else
@failure_count = 0
@circuit_open = false
"Service response: #{Time.now}"
end
end
end
private
def circuit_open?
@circuit_open && @last_failure_time && (Time.now - @last_failure_time) < 5
end
end
client = ServiceClient.new
# Make multiple calls
10.times do |i|
promise = client.call_service
promise.then do |result|
puts "Call #{i} succeeded: #{result}"
end.rescue do |error|
puts "Call #{i} failed: #{error.message}"
end
sleep(0.5)
end
end
def self.promise_memoization
# Memoized async operations
class AsyncMemoizer
def initialize
@cache = Concurrent::Map.new
end
def fetch(key, &block)
@cache.compute_if_absent(key) do
Concurrent::Promise.execute(&block)
end
end
end
memoizer = AsyncMemoizer.new
# Multiple calls for the same key will share the same promise
3.times do |i|
promise = memoizer.fetch("expensive_operation") do
puts "Executing expensive operation..."
sleep(1)
"Expensive result"
end
promise.then do |result|
puts "Call #{i} got result: #{result}"
end
end
sleep(2) # Wait for completion
end
end
AdvancedPromisePatterns.timeout_and_fallback
AdvancedPromisePatterns.circuit_breaker_pattern
AdvancedPromisePatterns.promise_memoizationActors provide isolated, message-driven computation with built-in supervision for fault tolerance.
class CounterActor < Concurrent::Actor::RestartingContext
def initialize(initial_value = 0)
@count = initial_value
end
def on_message(message)
case message
when :increment
@count += 1
@count
when :decrement
@count -= 1
@count
when :get
@count
when :reset
@count = 0
@count
when Integer
@count = message
@count
else
puts "Unknown message: #{message}"
end
end
end
# Bank account actor with validation
class BankAccountActor < Concurrent::Actor::RestartingContext
def initialize(initial_balance = 0)
@balance = initial_balance.to_f
@transaction_log = []
end
def on_message(message)
case message
when { type: :deposit, amount: amount }
deposit(amount)
when { type: :withdraw, amount: amount }
withdraw(amount)
when { type: :balance }
@balance
when { type: :history }
@transaction_log.dup
when { type: :transfer, to: to_account, amount: amount }
transfer(to_account, amount)
else
{ error: "Unknown message: #{message}" }
end
end
private
def deposit(amount)
if amount > 0
@balance += amount
log_transaction(:deposit, amount)
{ success: true, balance: @balance }
else
{ error: "Deposit amount must be positive" }
end
end
def withdraw(amount)
if amount > 0 && amount <= @balance
@balance -= amount
log_transaction(:withdrawal, amount)
{ success: true, balance: @balance }
elsif amount > @balance
{ error: "Insufficient funds" }
else
{ error: "Withdrawal amount must be positive" }
end
end
def transfer(to_account, amount)
if amount > 0 && amount <= @balance
@balance -= amount
log_transaction(:transfer_out, amount, to: to_account.path)
result = to_account.ask({ type: :deposit, amount: amount })
if result[:success]
{ success: true, balance: @balance }
else
# Rollback on failure
@balance += amount
log_transaction(:transfer_rollback, amount)
{ error: "Transfer failed: #{result[:error]}" }
end
else
{ error: "Invalid transfer amount" }
end
end
def log_transaction(type, amount, metadata = {})
@transaction_log << {
type: type,
amount: amount,
timestamp: Time.now,
balance_after: @balance
}.merge(metadata)
end
end
# Usage examples
def demonstrate_actors
# Create counter actor
counter = Concurrent::Actor.spawn(CounterActor, :counter, 10)
puts "Initial count: #{counter.ask(:get)}"
puts "After increment: #{counter.ask(:increment)}"
puts "After decrement: #{counter.ask(:decrement)}"
puts "Setting to 100: #{counter.ask(100)}"
# Create bank accounts
alice_account = Concurrent::Actor.spawn(BankAccountActor, :alice, 1000)
bob_account = Concurrent::Actor.spawn(BankAccountActor, :bob, 500)
puts "Alice balance: #{alice_account.ask(type: :balance)}"
puts "Bob balance: #{bob_account.ask(type: :balance)}"
# Perform transactions
result = alice_account.ask(type: :withdraw, amount: 100)
puts "Alice withdrawal: #{result}"
result = alice_account.ask(type: :transfer, to: bob_account, amount: 200)
puts "Transfer result: #{result}"
puts "Alice final balance: #{alice_account.ask(type: :balance)}"
puts "Bob final balance: #{bob_account.ask(type: :balance)}"
# Get transaction history
history = alice_account.ask(type: :history)
puts "Alice transaction history:"
history.each_with_index do |transaction, i|
puts " #{i + 1}. #{transaction[:type]}: #{transaction[:amount]} at #{transaction[:timestamp]}"
end
end
demonstrate_actorsclass SupervisorActor < Concurrent::Actor::RestartingContext
def initialize
@workers = {}
@worker_count = 0
end
def on_message(message)
case message
when { type: :spawn_worker, worker_class: worker_class, args: args }
spawn_worker(worker_class, args)
when { type: :stop_worker, worker_id: worker_id }
stop_worker(worker_id)
when { type: :list_workers }
@workers.keys
when { type: :send_to_worker, worker_id: worker_id, message: worker_message }
send_to_worker(worker_id, worker_message)
when { type: :broadcast, message: broadcast_message }
broadcast(broadcast_message)
else
{ error: "Unknown supervisor message: #{message}" }
end
end
private
def spawn_worker(worker_class, args)
@worker_count += 1
worker_id = "worker_#{@worker_count}"
begin
worker = Concurrent::Actor.spawn(worker_class, worker_id.to_sym, *args)
@workers[worker_id] = worker
# Monitor worker for termination
worker.ask(:ping).rescue do |error|
puts "Worker #{worker_id} failed during creation: #{error}"
@workers.delete(worker_id)
end
{ success: true, worker_id: worker_id }
rescue => e
{ error: "Failed to spawn worker: #{e.message}" }
end
end
def stop_worker(worker_id)
if worker = @workers.delete(worker_id)
worker.ask(:terminate)
{ success: true }
else
{ error: "Worker not found: #{worker_id}" }
end
end
def send_to_worker(worker_id, message)
if worker = @workers[worker_id]
begin
result = worker.ask(message)
{ success: true, result: result }
rescue => e
puts "Worker #{worker_id} failed, restarting..."
restart_worker(worker_id)
{ error: "Worker failed: #{e.message}" }
end
else
{ error: "Worker not found: #{worker_id}" }
end
end
def broadcast(message)
results = {}
@workers.each do |worker_id, worker|
begin
results[worker_id] = worker.ask(message)
rescue => e
results[worker_id] = { error: e.message }
restart_worker(worker_id)
end
end
results
end
def restart_worker(worker_id)
old_worker = @workers[worker_id]
return unless old_worker
begin
# Attempt to restart with the same configuration
new_worker = Concurrent::Actor.spawn(WorkerActor, worker_id.to_sym)
@workers[worker_id] = new_worker
puts "Restarted worker: #{worker_id}"
rescue => e
puts "Failed to restart worker #{worker_id}: #{e.message}"
@workers.delete(worker_id)
end
end
end
class WorkerActor < Concurrent::Actor::RestartingContext
def initialize(name = "unnamed")
@name = name
@processed_count = 0
end
def on_message(message)
case message
when :ping
"pong from #{@name}"
when { type: :process, data: data }
process_data(data)
when { type: :status }
{ name: @name, processed: @processed_count }
when { type: :fail }
raise "Intentional failure for testing"
when :terminate
terminate!
else
{ error: "Unknown message: #{message}" }
end
end
private
def process_data(data)
@processed_count += 1
sleep(rand(0.1..0.5)) # Simulate processing time
# Simulate occasional failures
raise "Processing failed" if rand < 0.1
{
success: true,
processed_data: data.to_s.upcase,
count: @processed_count
}
end
end
# Demonstration
def demonstrate_supervision
supervisor = Concurrent::Actor.spawn(SupervisorActor, :supervisor)
# Spawn some workers
3.times do |i|
result = supervisor.ask(type: :spawn_worker, worker_class: WorkerActor, args: ["worker_#{i}"])
puts "Spawned worker: #{result}"
end
# List workers
workers = supervisor.ask(type: :list_workers)
puts "Active workers: #{workers}"
# Send work to specific workers
workers.each_with_index do |worker_id, i|
result = supervisor.ask(
type: :send_to_worker,
worker_id: worker_id,
message: { type: :process, data: "task_#{i}" }
)
puts "Work result: #{result}"
end
# Broadcast a status check
status_results = supervisor.ask(type: :broadcast, message: { type: :status })
puts "Worker statuses: #{status_results}"
# Test failure handling
puts "Testing worker failure..."
failure_result = supervisor.ask(
type: :send_to_worker,
worker_id: workers.first,
message: { type: :fail }
)
puts "Failure test result: #{failure_result}"
sleep(1) # Allow restart to complete
# Check if worker was restarted
final_workers = supervisor.ask(type: :list_workers)
puts "Workers after restart: #{final_workers}"
end
demonstrate_supervisionclass ConcurrentCollectionsDemo
def self.demonstrate_map
# Thread-safe hash with atomic operations
concurrent_map = Concurrent::Map.new
# Spawn multiple threads to modify the map
threads = 10.times.map do |i|
Thread.new do
# Atomic put if absent
concurrent_map.put_if_absent("key_#{i}", "value_#{i}")
# Atomic compute operations
concurrent_map.compute("counter") do |old_value|
(old_value || 0) + 1
end
# Atomic replace
concurrent_map.replace("key_#{i}", "updated_value_#{i}")
end
end
threads.each(&:join)
puts "Final map size: #{concurrent_map.size}"
puts "Counter value: #{concurrent_map['counter']}"
puts "Sample entries: #{concurrent_map.to_h.first(3)}"
end
def self.demonstrate_array
# Thread-safe array
concurrent_array = Concurrent::Array.new
# Multiple threads adding elements
threads = 5.times.map do |i|
Thread.new do
10.times do |j|
concurrent_array << "item_#{i}_#{j}"
end
end
end
threads.each(&:join)
puts "Array size: #{concurrent_array.size}"
puts "First 5 elements: #{concurrent_array.first(5)}"
# Safe iteration
concurrent_array.each_with_index do |item, index|
puts "#{index}: #{item}" if index < 3
end
end
def self.demonstrate_set
# Thread-safe set
concurrent_set = Concurrent::Set.new
# Multiple threads adding overlapping elements
threads = 5.times.map do |i|
Thread.new do
10.times do |j|
concurrent_set.add("element_#{j % 5}") # Intentional duplicates
end
end
end
threads.each(&:join)
puts "Set size (should be 5): #{concurrent_set.size}"
puts "Set contents: #{concurrent_set.to_a}"
end
end
ConcurrentCollectionsDemo.demonstrate_map
ConcurrentCollectionsDemo.demonstrate_array
ConcurrentCollectionsDemo.demonstrate_setclass AtomicOperationsDemo
def self.demonstrate_atomic_reference
# Atomic reference for thread-safe object updates
atomic_ref = Concurrent::AtomicReference.new({ count: 0, message: "initial" })
threads = 10.times.map do |i|
Thread.new do
5.times do
# Atomic compare and set
loop do
current = atomic_ref.get
new_value = {
count: current[:count] + 1,
message: "updated by thread #{i}"
}
break if atomic_ref.compare_and_set(current, new_value)
# Retry if another thread modified the value
end
end
end
end
threads.each(&:join)
final_value = atomic_ref.get
puts "Final atomic reference: #{final_value}"
end
def self.demonstrate_atomic_fixnum
# Atomic integer operations
atomic_counter = Concurrent::AtomicFixnum.new(0)
threads = 10.times.map do
Thread.new do
100.times do
atomic_counter.increment
end
end
end
threads.each(&:join)
puts "Atomic counter final value: #{atomic_counter.value}" # Should be 1000
# Atomic operations
puts "Increment and get: #{atomic_counter.increment}"
puts "Decrement and get: #{atomic_counter.decrement}"
puts "Add 10: #{atomic_counter.update { |v| v + 10 }}"
puts "Compare and set (1001 to 2000): #{atomic_counter.compare_and_set(1010, 2000)}"
end
def self.demonstrate_atomic_boolean
# Atomic boolean for thread-safe flags
shutdown_flag = Concurrent::AtomicBoolean.new(false)
ready_flag = Concurrent::AtomicBoolean.new(false)
# Worker thread
worker = Thread.new do
puts "Worker starting..."
ready_flag.make_true
until shutdown_flag.true?
puts "Working..."
sleep(0.5)
end
puts "Worker shutting down..."
end
# Wait for worker to be ready
sleep(0.1) until ready_flag.true?
puts "Worker is ready!"
# Let it work for a bit
sleep(2)
# Signal shutdown
puts "Signaling shutdown..."
shutdown_flag.make_true
worker.join
puts "Worker stopped"
end
end
AtomicOperationsDemo.demonstrate_atomic_reference
AtomicOperationsDemo.demonstrate_atomic_fixnum
AtomicOperationsDemo.demonstrate_atomic_booleanclass ExecutorDemo
def self.demonstrate_thread_pool_executor
# Fixed thread pool
executor = Concurrent::ThreadPoolExecutor.new(
min_threads: 2,
max_threads: 4,
max_queue: 10,
idle_time: 60
)
# Submit multiple tasks
futures = 10.times.map do |i|
Concurrent::Future.execute(executor: executor) do
puts "Task #{i} executing on #{Thread.current.name}"
sleep(rand(0.1..0.5))
"Result #{i}"
end
end
# Wait for all tasks to complete
results = futures.map(&:value)
puts "All results: #{results}"
# Shutdown executor
executor.shutdown
executor.wait_for_termination(5)
end
def self.demonstrate_cached_thread_pool
# Cached thread pool - creates threads as needed
executor = Concurrent::CachedThreadPool.new
# Submit burst of tasks
futures = 20.times.map do |i|
Concurrent::Future.execute(executor: executor) do
puts "Burst task #{i} on #{Thread.current.name}"
sleep(0.1)
i * i
end
end
results = futures.map(&:value)
puts "Burst results sum: #{results.sum}"
executor.shutdown
executor.wait_for_termination(5)
end
def self.demonstrate_single_thread_executor
# Single thread executor - guarantees serial execution
executor = Concurrent::SingleThreadExecutor.new
# These will execute one at a time
10.times do |i|
executor.post do
puts "Serial task #{i} at #{Time.now.strftime('%H:%M:%S.%L')}"
sleep(0.1)
end
end
sleep(2) # Wait for tasks to complete
executor.shutdown
executor.wait_for_termination(5)
end
def self.demonstrate_immediate_executor
# Immediate executor - runs tasks in the calling thread
executor = Concurrent::ImmediateExecutor.new
puts "Before submitting tasks (#{Thread.current.name})"
3.times do |i|
executor.post do
puts "Immediate task #{i} on #{Thread.current.name}"
end
end
puts "After submitting tasks"
end
end
ExecutorDemo.demonstrate_thread_pool_executor
ExecutorDemo.demonstrate_cached_thread_pool
ExecutorDemo.demonstrate_single_thread_executor
ExecutorDemo.demonstrate_immediate_executorclass ConcurrentWebScraper
def initialize(max_workers: 5, request_delay: 0.1)
@executor = Concurrent::ThreadPoolExecutor.new(
min_threads: 1,
max_threads: max_workers,
max_queue: 100
)
@results = Concurrent::Map.new
@errors = Concurrent::Array.new
@request_delay = request_delay
@rate_limiter = Concurrent::Semaphore.new(max_workers)
end
def scrape_urls(urls)
# Create futures for all URLs
futures = urls.map do |url|
Concurrent::Future.execute(executor: @executor) do
scrape_single_url(url)
end
end
# Wait for all to complete with timeout
Concurrent::Future.zip(*futures).value(30) # 30 second timeout
{
results: @results.to_h,
errors: @errors.to_a,
success_count: @results.size,
error_count: @errors.size
}
rescue Concurrent::TimeoutError
{
results: @results.to_h,
errors: @errors.to_a + ["Scraping timed out"],
success_count: @results.size,
error_count: @errors.size + 1
}
ensure
@executor.shutdown
@executor.wait_for_termination(10)
end
private
def scrape_single_url(url)
@rate_limiter.acquire
begin
# Simulate HTTP request
sleep(@request_delay)
# Simulate occasional failures
if rand < 0.1
raise "HTTP error for #{url}"
end
# Simulate parsing
content = "Content from #{url} - #{rand(1000)} words"
@results[url] = {
content: content,
scraped_at: Time.now,
thread: Thread.current.name
}
puts "Successfully scraped: #{url}"
rescue => e
error_info = {
url: url,
error: e.message,
timestamp: Time.now
}
@errors << error_info
puts "Failed to scrape #{url}: #{e.message}"
ensure
@rate_limiter.release
end
end
end
# Usage example
scraper = ConcurrentWebScraper.new(max_workers: 3, request_delay: 0.2)
urls = [
"https://example1.com",
"https://example2.com",
"https://example3.com",
"https://example4.com",
"https://example5.com"
]
puts "Starting concurrent scraping..."
results = scraper.scrape_urls(urls)
puts "\n=== Scraping Results ==="
puts "Successful: #{results[:success_count]}"
puts "Failed: #{results[:error_count]}"
puts "\nSuccessful scrapes:"
results[:results].each do |url, data|
puts " #{url}: #{data[:content][0..50]}... (#{data[:thread]})"
end
unless results[:errors].empty?
puts "\nErrors:"
results[:errors].each do |error|
puts " #{error[:url] || 'Unknown'}: #{error[:error] || error}"
end
endWatch and learn concurrent-ruby gem