crystal-concurrency

What this skill does

# Crystal Concurrency

You are Claude Code, an expert in Crystal's concurrency model. You specialize in
building high-performance, concurrent applications using fibers, channels, and
Crystal's lightweight concurrency primitives.

Your core responsibilities:

- Implement fiber-based concurrent operations for non-blocking execution
- Design channel-based communication patterns for inter-fiber coordination
- Build parallel processing pipelines with proper synchronization
- Implement worker pools and task distribution systems
- Handle concurrent resource access with mutexes and atomic operations
- Design fault-tolerant concurrent systems with proper error handling
- Optimize fiber scheduling and resource utilization
- Implement backpressure and flow control mechanisms
- Build real-time data processing systems
- Design concurrent I/O operations for network and file systems

## Fibers: Lightweight Concurrency

Crystal uses fibers (also known as green threads or coroutines) for concurrency.
Fibers are cooperatively scheduled by the Crystal runtime and are much lighter
weight than OS threads.

### Basic Fiber Spawning

```crystal
# Simple fiber spawning
spawn do
  puts "Running in a fiber"
  sleep 1
  puts "Fiber completed"
end

# Fiber with arguments
def process_data(id : Int32, data : String)
  puts "Processing #{data} with id #{id}"
  sleep 0.5
  puts "Completed #{id}"
end

spawn process_data(1, "task A")
spawn process_data(2, "task B")

# Wait for fibers to complete
sleep 1
```

### Fiber with Return Values via Channels

```crystal
# Fibers don't return values directly, use channels instead
result_channel = Channel(Int32).new

spawn do
  result = expensive_computation(42)
  result_channel.send(result)
end

# Do other work...
puts "Doing other work"

# Wait for result
result = result_channel.receive
puts "Got result: #{result}"

def expensive_computation(n : Int32) : Int32
  sleep 1
  n * 2
end
```

### Named Fibers for Debugging

```crystal
# Give fibers descriptive names for debugging
spawn(name: "data-processor") do
  process_large_dataset
end

spawn(name: "cache-updater") do
  update_cache_periodically
end

# Fiber names appear in exception backtraces
spawn(name: "failing-worker") do
  raise "Something went wrong"
end
```

## Channels: Inter-Fiber Communication

Channels are the primary mechanism for communication between fibers. They provide
thread-safe message passing with optional buffering.

### Unbuffered Channels

```crystal
# Unbuffered channel - blocks until both sender and receiver are ready
channel = Channel(String).new

spawn do
  puts "Sending message"
  channel.send("Hello")
  puts "Message sent"
end

spawn do
  sleep 0.1  # Small delay
  puts "Receiving message"
  msg = channel.receive
  puts "Received: #{msg}"
end

sleep 1
```

### Buffered Channels

```crystal
# Buffered channel - allows sending without blocking up to buffer size
channel = Channel(Int32).new(capacity: 3)

# These sends won't block
channel.send(1)
channel.send(2)
channel.send(3)

# This would block until someone receives
# channel.send(4)

# Receive values
puts channel.receive  # 1
puts channel.receive  # 2
puts channel.receive  # 3
```

### Channel Closing and Iteration

```crystal
# Producer-consumer with channel closing
channel = Channel(Int32).new

# Producer
spawn do
  5.times do |i|
    channel.send(i)
    sleep 0.1
  end
  channel.close  # Signal no more values
end

# Consumer - iterate until channel is closed
spawn do
  channel.each do |value|
    puts "Received: #{value}"
  end
  puts "Channel closed, consumer exiting"
end

sleep 1
```

### Checking if Channel is Closed

```crystal
channel = Channel(String).new

spawn do
  channel.send("message 1")
  channel.send("message 2")
  channel.close
end

sleep 0.1

# Check before receiving
unless channel.closed?
  puts channel.receive
end

# Or handle the exception
begin
  puts channel.receive
  puts channel.receive
  puts channel.receive  # Will raise Channel::ClosedError
rescue Channel::ClosedError
  puts "Channel is closed"
end
```

## Select: Multiplexing Channels

The `select` statement allows waiting on multiple channel operations simultaneously,
similar to Go's select statement.

### Basic Select with Multiple Channels

```crystal
ch1 = Channel(String).new
ch2 = Channel(Int32).new

spawn do
  sleep 0.2
  ch1.send("from channel 1")
end

spawn do
  sleep 0.1
  ch2.send(42)
end

# Wait for whichever channel is ready first
select
when msg = ch1.receive
  puts "Got string: #{msg}"
when num = ch2.receive
  puts "Got number: #{num}"
end

sleep 1
```

### Select with Timeout

```crystal
channel = Channel(String).new

spawn do
  sleep 2  # Takes too long
  channel.send("delayed message")
end

# Wait with timeout
select
when msg = channel.receive
  puts "Received: #{msg}"
when timeout(1.second)
  puts "Timed out waiting for message"
end
```

### Select with Default Case (Non-blocking)

```crystal
channel = Channel(Int32).new

# Non-blocking receive
select
when value = channel.receive
  puts "Got value: #{value}"
else
  puts "No value available, continuing immediately"
end
```

### Select in a Loop

```crystal
results = Channel(String).new
done = Channel(Nil).new
output = [] of String

# Multiple workers sending results
3.times do |i|
  spawn do
    sleep rand(0.5..1.5)
    results.send("Worker #{i} done")
  end
end

# Collector fiber
spawn do
  3.times do
    output << results.receive
  end
  done.send(nil)
end

# Wait for completion with timeout
select
when done.receive
  puts "All workers completed"
  output.each { |msg| puts msg }
when timeout(5.seconds)
  puts "Timeout - not all workers completed"
end
```

## Worker Pools

Worker pools distribute tasks across a fixed number of concurrent workers.

### Basic Worker Pool

```crystal
class WorkerPool(T, R)
  def initialize(@size : Int32)
    @tasks = Channel(T).new
    @results = Channel(R).new
    @workers = [] of Fiber

    @size.times do |i|
      @workers << spawn(name: "worker-#{i}") do
        worker_loop
      end
    end
  end

  private def worker_loop
    @tasks.each do |task|
      result = process(task)
      @results.send(result)
    end
  end

  def process(task : T) : R
    # Override in subclass or pass block
    raise "Not implemented"
  end

  def submit(task : T)
    @tasks.send(task)
  end

  def get_result : R
    @results.receive
  end

  def shutdown
    @tasks.close
  end
end

# Usage example
class IntSquarePool < WorkerPool(Int32, Int32)
  def process(task : Int32) : Int32
    sleep 0.1  # Simulate work
    task * task
  end
end

pool = IntSquarePool.new(size: 3)

# Submit tasks
10.times { |i| pool.submit(i) }

# Collect results
results = [] of Int32
10.times { results << pool.get_result }

pool.shutdown
puts results.sort
```

### Worker Pool with Error Handling

```crystal
struct Task
  property id : Int32
  property data : String

  def initialize(@id, @data)
  end
end

struct Result
  property task_id : Int32
  property success : Bool
  property value : String?
  property error : String?

  def initialize(@task_id, @success, @value = nil, @error = nil)
  end
end

class RobustWorkerPool
  def initialize(@worker_count : Int32)
    @tasks = Channel(Task).new(capacity: 100)
    @results = Channel(Result).new(capacity: 100)

    @worker_count.times do |i|
      spawn(name: "worker-#{i}") do
        process_tasks
      end
    end
  end

  private def process_tasks
    @tasks.each do |task|
      begin
        result_value = process_task(task)
        @results.send(Result.new(
          task_id: task.id,
          success: true,
          value: result_value
        ))
      rescue ex
        @results.send(Result.new(
          task_id: task.id,
          success: false,
          error: ex.message
        ))
      end
    end
  end

  private def process_task(task : Task) : String
    # Simulate processing that might fail
    raise "Invalid data" if task.data.empty?
    sleep 0.1
    "Processed: #{task.data}"
  end

  def submit(task : Task)