Minigun go BRRRRR

Minigun is a high-performance data processing pipeline framework for Ruby.

Features

• Define multi-stage processing pipelines with a simple, expressive DSL.
• Process data using multiple threads and/or processes for maximum performance.
• Use Copy-On-Write (COW) or IPC forking for efficient parallel processing.
• Direct connections between stages with from and to options.
• Queue-based routing with selective queue subscriptions.
• Batch accumulation for efficient processing.
• Comprehensive error handling and retry mechanisms.
• Optional MessagePack serialization for faster IPC.
• Data compression for large transfers between processes.
• Smart garbage collection for memory optimization.

In many use cases, Minigun can replace queue systems like Resque, Solid Queue, or Sidekiq. Minigun itself is run entire in Ruby's memory, and is database and application agnostic.

TODOs

• Extract examples to examples folder
• Add more examples based on real-world use cases
• Add support for named queues and queue-based routing (already there?)
• Add parallel and sequential blocks for defining parallel and sequential stages
• Add support for custom error handling and retry strategies
• Add support for custom logging and monitoring
• Add support for custom thread and process management (?)

Installation

Add this line to your application's Gemfile:

gem'minigun'

Quick Start

require'minigun'classMyTaskincludeMinigun::DSLpipelinedoproducer:generatedo10.times{ |i| emit(i)}endprocessor:transformdo |number|
emit(number * 2)endaccumulator:batchdo |item|
@items ||= []@items << itemif@items.size >= 5batch=@items.dup@items.clearemit(batch)endendcow_fork:process_batchdo |batch|
# Process the batch in a forked child processbatch.each{ |item| puts"Processing #{item}"}endendend# Run the taskMyTask.new.run

Core Concepts

Pipeline Stages

Minigun has unified its stage types into a cohesive system where each specialized stage is a variation of a common processor implementation:

1. Producer: Generates data for the pipeline. A producer is a processor without input.
2. Processor: Transforms data and passes it to the next stage. It can filter, modify, or route data.
3. Accumulator: Collects and batches items before forwarding them in groups.
4. Consumer: Consumes data without emitting anything further. A consumer is a processor without output.

Fork Variants

For handling batched data processing, two fork implementations are available:

• cow_fork: Uses Copy-On-Write fork to efficiently process batches in separate child processes
• ipc_fork: Uses IPC-style forking for batch processing with different memory characteristics

These are actually aliases for the consumer stage with specific fork configurations.

Custom Stage Classes

Minigun allows you to create custom stage classes to encapsulate complex behavior or implement specialized processing patterns. All stages inherit from the base Stage class and can override its behavior.

Execution Modes

Every stage has a run_mode that determines how it executes within the pipeline. There are three execution strategies:

:autonomous# Generates data independently (ProducerStage):streaming# Processes stream of items in worker loop (Stage, ConsumerStage):composite# Manages internal stages (PipelineStage)

The run_mode method controls critical behaviors like:

• Whether the stage needs an input queue
• Whether it needs an executor for concurrent processing
• How the pipeline routes data to and from the stage
• How the stage participates in disconnection detection

Creating Custom Stages

To create a custom stage class, inherit from Minigun::Stage and implement the required methods:

classCustomStage < Minigun::Stage# Define execution mode (default: :streaming)defrun_mode:streamingend# Define how a single item is processeddefexecute(context,item: nil,input_queue: nil,output_queue: nil)# Your custom processing logicresult=process_item(item)output_queue << resultifoutput_queueend# Optional: Customize the stage executiondefrun_stage(stage_ctx)# Custom execution implementation# See ProducerStage or ConsumerStage for examplesend# Optional: Customize logging typedeflog_type"Custom"endend

Example: Batch Processor Stage

Here's a custom stage that batches items with a timeout:

classTimedBatchStage < Minigun::Stageattr_reader:batch_size,:timeoutdefinitialize(name:,options: {})super@batch_size=options[:batch_size] || 100@timeout=options[:timeout] || 5.0enddefrun_mode:streaming# Processes items from input queueenddefrun_stage(stage_ctx)require'minigun/queue_wrappers'wrapped_input=Minigun::InputQueue.new(stage_ctx.input_queue,stage_ctx.stage_name,stage_ctx.sources_expected)wrapped_output=Minigun::OutputQueue.new(stage_ctx.stage_name,stage_ctx.dag.downstream(stage_ctx.stage_name).map{ |ds|
stage_ctx.stage_input_queues[ds]},stage_ctx.stage_input_queues,stage_ctx.runtime_edges)batch=[]last_flush=Time.nowloopdo# Check for timeoutif !batch.empty? && (Time.now - last_flush) >= @timeoutwrapped_output << batch.dupbatch.clearlast_flush=Time.nowend# Try to get item with timeoutbeginitem=wrapped_input.pop(timeout: 0.1)ifitem == Minigun::AllUpstreamsDone# Flush remaining itemswrapped_output << batchunlessbatch.empty?breakendbatch << item# Flush if batch is fullifbatch.size >= @batch_sizewrapped_output << batch.dupbatch.clearlast_flush=Time.nowendrescueThreadError# Timeout, continue to check for flushendendsend_end_signals(stage_ctx)endend# Use in a pipelineclassMyTaskincludeMinigun::DSLpipelinedoproducer:generatedo100.times{ |i| emit(i)}end# Use custom stage classcustom_stageTimedBatchStage,:batch,batch_size: 10,timeout: 2.0consumer:processdo |batch,output|
puts"Processing batch of #{batch.size} items"endendend

Example: Stateful Filter Stage

Create a stage that filters based on accumulated state:

classDeduplicatorStage < Minigun::Stagedefinitialize(name:,options: {})super@seen=Set.new@mutex=Mutex.newenddefrun_mode:streamingenddefexecute(context,item: nil,input_queue: nil,output_queue: nil)key=extract_key(item)is_new=@mutex.synchronizedoif@seen.include?(key)falseelse@seen.add(key)trueendendoutput_queue << itemifis_new && output_queueendprivatedefextract_key(item)# Override in subclass or pass as optionitem[:id] || itemendend

When to Create Custom Stages

Consider creating custom stage classes when you need:

1. Complex State Management: Stages that maintain sophisticated internal state
2. Specialized Worker Loops: Custom timing, batching, or control flow logic
3. Reusable Patterns: Behavior you want to use across multiple pipelines
4. Framework Extensions: Adding new execution modes or patterns to Minigun
5. Performance Optimization: Fine-tuned control over threading, batching, or memory

For simple transformations, use the standard producer, processor, and consumer DSL methods. For complex, reusable behavior, create custom stage classes.

Stage Connections

Minigun supports two types of stage connections:

1. Sequential Connections: By default, stages are connected in the order they're defined
2. Explicit Connections: Use from and to options to explicitly define connections

# Sequential connectionprocessor:first_stagedo |item|
item + 1endprocessor:second_stagedo |item|
item * 2end# Explicit connectionsprocessor:stage_a,to: [:stage_b,:stage_c]do |item|
itemendprocessor:stage_b,from: :stage_ado |item|
# Process items from stage_aendprocessor:stage_c,from: :stage_ado |item|
# Also process items from stage_aend

Advanced Connection Examples

Branching Pipeline

Create a pipeline that branches based on the type of data:

pipelinedo# Producer emits to multiple processorsproducer:user_producer,to: [:email_processor,:notification_processor]doUser.find_eachdo |user|
emit(user)endend# These processors receive data from the same producerprocessor:email_processor,from: :user_producerdo |user|
generate_email(user)endprocessor:notification_processor,from: :user_producerdo |user|
generate_notification(user)end# Connect the email processor to an accumulatoraccumulator:email_accumulator,from: :email_processordo |email|
@emails ||= []@emails << emailif@emails.size >= 100batch=@emails.dup@emails.clearemit(batch)endend# Process accumulated emailscow_fork:email_sender,from: :email_accumulator,processes: 4do |emails|
emails.each{ |email| send_email(email)}end# Process notifications directlyconsumer:notification_sender,from: :notification_processordo |notification|
send_notification(notification)endend

Diamond-Shaped Pipeline

Create a pipeline that splits and rejoins:

pipelinedoproducer:data_sourcedodata_items.each{ |item| emit(item)}end# Split to parallel processorsprocessor:validate,from: :data_source,to: [:transform_a,:transform_b]do |item|
emit(item)ifitem.valid?end# Parallel transformationsprocessor:transform_a,from: :validate,to: :combinedo |item|
emit(transform_a(item))endprocessor:transform_b,from: :validate,to: :combinedo |item|
emit(transform_b(item))end# Rejoin for final processingprocessor:combine,from: [:transform_a,:transform_b]do |item|
@results ||= []@results << itemif@results.size >= 2emit(combine_results(@results))@results.clearendendconsumer:store_results,from: :combinedo |result|
store_result(result)endend

Queue-Based Routing

You can route items to specific stages by subscribing to named queues:

processor:route,to: [:high_priority,:low_priority]do |item|
ifitem[:priority] == :highemit_to_queue(:high_priority,item)elseemit_to_queue(:low_priority,item)endendprocessor:high_priority,queues: [:high_priority]do |item|
# Process high priority itemsendprocessor:low_priority,queues: [:low_priority]do |item|
# Process low priority itemsend

Advanced Queue Examples

Priority Processing

Create a pipeline with priority lanes for VIP users:

pipelinedoproducer:user_producerdoUser.find_eachdo |user|
emit(user)# Route VIP users to a high priority queueemit_to_queue(:high_priority,user)ifuser.vip?endend# This processor handles both default and high priority usersprocessor:email_processor,threads: 5,queues: [:default,:high_priority]do |user|
email=generate_email(user)emit(email)end# Regular handling for emailsaccumulator:email_accumulator,from: :email_processordo |email|
@emails ||= {}@emails[email.type] ||= []@emails[email.type] << email# Emit batches by email type when they reach the threshold@emails.eachdo |type,batch|
ifbatch.size >= 50emit_to_queue(type,batch.dup)batch.clearendendend# Handle newsletter emails separatelyconsumer:newsletter_sender,queues: [:newsletter]do |emails|
send_newsletter_batch(emails)end# Handle transaction emails separatelyconsumer:transaction_sender,queues: [:transaction]do |emails|
send_transaction_batch(emails)end# Handle all other typesconsumer:general_sender,queues: [:default]do |emails|
send_email_batch(emails)endend

Load Balancing

Distribute work across multiple queues for better load balancing:

pipelinedoproducer:data_sourcedolarge_dataset.each_with_indexdo |item,i|
# Round-robin distribute across multiple queuesqueue=[:queue_1,:queue_2,:queue_3][i % 3]emit_to_queue(queue,item)endend# Process queue 1 with specific settingsprocessor:worker_1,queues: [:queue_1],threads: 3do |item|
process_with_worker_1(item)end# Process queue 2 with different settingsprocessor:worker_2,queues: [:queue_2],threads: 5do |item|
process_with_worker_2(item)end# Process queue 3 with yet different settingsprocessor:worker_3,queues: [:queue_3],threads: 2do |item|
process_with_worker_3(item)end# All results go to the same accumulatoraccumulator:result_collector,from: [:worker_1,:worker_2,:worker_3]do |result|
@results ||= []@results << resultif@results.size >= 100batch=@results.dup@results.clearemit(batch)endendconsumer:store_results,from: :result_collectordo |batch|
store_batch(batch)endend

Execution Strategies

Minigun provides multiple execution strategies for running pipeline stages, each optimized for different use cases. You can configure execution at the task level or per-stage.

Available Executors

1. Inline Executor (:inline)

The simplest executor - runs everything sequentially in the main process.

classSimpleTaskincludeMinigun::Taskexecution:inline# Run everything in the main processpipelinedoproducer:generate{10.times{ |i| emit(i)}}processor:transform{ |n| emit(n * 2)}consumer:output{ |n| putsn}endend

Characteristics:

• No concurrency
• Minimal overhead
• Easy to debug
• Best for simple, fast operations

Use when:

• Operations are very fast
• You need to debug the pipeline
• Data volume is small

---

2. Thread Pool Executor (:thread)

Runs stages concurrently using a thread pool. This is the most common executor.

classThreadedTaskincludeMinigun::Taskexecution:thread,max: 10# Use up to 10 threadspipelinedoproducer:fetch_urls{urls.each{ |url| emit(url)}}processor:download,threads: 5do |url|
# 5 threads concurrently downloadingemit(HTTP.get(url))endconsumer:save{ |content| File.write(...,content)}endend

Characteristics:

• Concurrent execution within a single process
• Shared memory (no serialization overhead)
• Subject to Ruby GVL (Global VM Lock)
• Low overhead for creating workers

Use when:

• Operations are I/O bound (network, disk, database)
• You need shared memory access
• Operations are thread-safe
• You want lightweight concurrency

---

3. COW Fork Pool Executor (:cow_fork)

Forks a new process for EACH item using Copy-On-Write memory sharing.

classCowForkTaskincludeMinigun::Taskexecution:cow_fork,max: 4# Up to 4 concurrent forkspipelinedoproducer:generate{100.times{ |i| emit(i)}}# Each item gets its own forked processprocessor:heavy_compute,execution: :cow_forkdo |item|
# This runs in a fresh forked process with COW memoryresult=expensive_computation(item)emit(result)endconsumer:save{ |result| save_result(result)}endend

How It Works:

1. Parent process pulls item from input queue
2. Forks a new child process (memory shared via COW)
3. Child processes one item and writes to output queue
4. Child exits immediately
5. Parent reaps completed children and continues
6. Maintains up to max concurrent child processes

Characteristics:

• Fork per item - ephemeral processes
• Copy-On-Write memory sharing (no serialization)
• Child inherits parent's memory state
• Memory pages shared until modified
• Automatic memory cleanup when process exits
• Each item processed in complete isolation

Use when:

• You have large read-only data structures
• Operations are CPU-intensive
• You want to avoid GVL limitations
• Operations might leak memory (cleaned up automatically)
• Each item needs fresh process state

Example with Large Shared Data:

classDataProcessorincludeMinigun::Taskdefinitialize# Large lookup table (50MB)@lookup_table=load_huge_datasetendexecution:cow_fork,max: 8pipelinedoproducer:generate{ids.each{ |id| emit(id)}}# Each fork gets COW access to @lookup_table# No serialization - memory is shared until written toprocessor:processdo |id|
# Can access @lookup_table directly - no copy!result=complex_calculation(id,@lookup_table)emit(result)endconsumer:save{ |result| save(result)}endend

---

4. IPC Fork Pool Executor (:ipc_fork)

Creates persistent worker processes that communicate via Inter-Process Communication (IPC).

classIpcForkTaskincludeMinigun::Taskexecution:ipc_fork,max: 4# Create 4 persistent workerspipelinedoproducer:generate{1000.times{ |i| emit(i)}}# Workers stay alive and process multiple itemsprocessor:compute,execution: :ipc_forkdo |item|
result=expensive_operation(item)emit(result)endconsumer:save{ |result| save_result(result)}endend

How It Works:

1. Parent spawns max persistent worker processes on startup
2. Workers communicate with parent via bidirectional pipes
3. Parent distributes items from input queue to workers (round-robin)
4. Workers pull items via IPC, process them, and send results back
5. Results are pushed to output queue (routed based on DAG)
6. Workers stay alive until stage completes
7. Parent coordinates shutdown when input queue is exhausted

Characteristics:

• Persistent workers - like ThreadPoolExecutor but with processes
• Explicit IPC via pipes (data is serialized)
• Strong process isolation
• Workers handle multiple items throughout their lifetime
• Overhead of process creation amortized across many items
• Data serialization overhead (uses Marshal or MessagePack)

Use when:

• Operations are CPU-intensive and long-running
• You need true parallelism (no GVL)
• Setup cost per item is high (e.g., loading models, establishing connections)
• You want persistent worker pools like Puma or Unicorn
• You're processing many items and want to amortize fork cost

IPC Optimizations:

Minigun provides several optimizations for IPC communication:

classOptimizedIpcTaskincludeMinigun::Task# Optional: Install msgpack gem for faster serialization# gem 'msgpack'execution:ipc_fork,max: 4,pipe_timeout: 60,# Timeout for pipe operationsuse_compression: true# Compress large transferspipelinedoproducer:generate{large_items.each{ |item| emit(item)}}processor:processdo |item|
# Item is deserialized from IPC# Process and emit resultemit(transform(item))endconsumer:save{ |result| save(result)}endend

Optimizations:

• MessagePack: Automatically used if msgpack gem is installed (faster than Marshal)
• Compression: Large data (>1KB) automatically compressed with Zlib
• Garbage Collection: Optimized GC before forking and during processing

---

Comparison Table

Executor	Concurrency	Process Model	Memory Sharing	Serialization	Best For
:inline	None	Single process	N/A	None	Simple, fast operations
:thread	Threads	Single process	Shared memory	None	I/O-bound operations
:cow_fork	Processes	Fork per item	COW (shared)	None	CPU-bound with large read-only data
:ipc_fork	Processes	Persistent workers	Isolated	Marshal/MessagePack	CPU-bound long-running operations

---

Choosing an Executor

Use :inline when:

• Debugging or testing
• Operations are trivial (< 1ms)
• Single-threaded is sufficient

Use :thread when:

• Operations are I/O-bound (databases, networks, files)
• You need shared memory
• Operations are thread-safe
• You want lightweight concurrency

Use :cow_fork when:

• Operations are CPU-intensive
• You have large read-only data structures
• You want true parallelism without GVL
• Each item needs complete process isolation
• Memory leaks are a concern (auto-cleanup)

Use :ipc_fork when:

• Operations are CPU-intensive AND long-running
• Setup cost per item is significant
• You want persistent worker pools
• You need strong process isolation
• You're processing many items

---

Per-Stage Execution

You can mix execution strategies within a single pipeline:

classHybridTaskincludeMinigun::Taskexecution:thread,max: 10# Default: thread poolpipelinedo# Runs in thread poolproducer:fetch_urls,threads: 5dourls.each{ |url| emit(url)}end# Runs in thread poolprocessor:download,threads: 10do |url|
emit(HTTP.get(url))end# Override: use COW fork for CPU-intensive workprocessor:process_images,execution: :cow_fork,max: 4do |html|
images=extract_images(html)emit(process_with_opencv(images))end# Override: use persistent IPC workers for ML inferenceprocessor:classify,execution: :ipc_fork,max: 2do |images|
# Workers load ML model once, reuse for all itemsemit(@model.predict(images))end# Back to thread poolconsumer:savedo |results|
database.insert(results)endendend

---

Configuration Options

classConfiguredTaskincludeMinigun::Task# Global configurationmax_threads10# Maximum threads per processmax_processes4# Maximum forked processesmax_retries3# Maximum retry attempts for errorsbatch_size100# Default batch sizeconsumer_type:cow# Default consumer fork implementation (:cow or :ipc)# Advanced IPC optionspipe_timeout30# Timeout for IPC pipe operations (seconds)use_compressiontrue# Enable compression for large IPC transfersgc_probability0.1# Probability of GC during batch processing (0.0-1.0)# Stage-specific configurationpipelinedoproducer:sourcedo# Generate dataendprocessor:transform,threads: 5do |item|
# Process with 5 threadsendaccumulator:batch,max_queue: 1000,max_all: 2000do |item|
# Batch with custom limitsendconsumer:sink,fork: :ipc,processes: 2do |batch|
# Consume with 2 IPC processesendendend

Queue Sizing and Backpressure

Minigun uses SizedQueue (bounded queues) by default for automatic backpressure. This prevents memory bloat when producers are faster than consumers.

Global Default:

# Set global default queue sizeMinigun.configuredo |config|
config.default_queue_size=1000# Default is 1000end

Per-Stage Queue Size:

pipelinedoproducer:fast_sourcedo |output|
# Produces data very quicklyend# Small queue for tight backpressureprocessor:slow_transform,queue_size: 50do |item,output|
# Slow processing creates backpressure on producersleep0.1output << item * 2end# Large queue for buffering burstsconsumer:batch_sink,queue_size: 5000do |item,output|
# Can handle bursty workloadsend# Unbounded queue (use with caution!)consumer:emergency_overflow,queue_size: Float::INFINITYdo |item,output|
# No backpressure - can grow without boundendend

Unbounded Queues:

Set queue_size to 0, nil, or Float::INFINITY for unbounded Queue instead of SizedQueue:

# All three are equivalentprocessor:stage1,queue_size: 0processor:stage2,queue_size: nilprocessor:stage3,queue_size: Float::INFINITY

When to Use Each:

• Bounded (default 1000): Best for most cases. Provides automatic backpressure.
• Small (50-100): Tight coupling between stages, immediate backpressure.
• Large (5000+): Buffer for bursty producers, smooth out spikes.
• Unbounded: Only when you're certain producers won't overwhelm consumers (e.g., rate-limited APIs).

Hooks

Minigun supports hooks for various lifecycle events:

classTaskWithHooksincludeMinigun::Taskbefore_rundo# Called before the pipeline startsendafter_rundo# Called after the pipeline completesendbefore_forkdo# Called in the parent process before forkingendafter_forkdo# Called in the child process after forkingendend

Example Use Cases

Data ETL Pipeline

classDataETLincludeMinigun::Taskpipelinedoproducer:extractdo# Extract data from sourcedatabase.each_batch(1000)do |batch|
emit(batch)endendprocessor:transformdo |batch|
# Transform the databatch.map{ |row| transform_row(row)}endconsumer:loaddo |batch|
# Load data to destinationdestination.insert_batch(batch)endendend

Parallel Web Crawler

classWebCrawlerincludeMinigun::Taskmax_threads20pipelinedoproducer:seed_urlsdoinitial_urls.each{ |url| emit(url)}endprocessor:fetch_pagesdo |url|
response=HTTP.get(url){url: url,content: response.body}endprocessor:extract_linksdo |page|
links=extract_links_from_html(page[:content])# Emit new links for crawlinglinks.each{ |link| emit(link)}# Pass the page content for processingpageendaccumulator:batch_pagesdo |page|
@pages ||= []@pages << pageif@pages.size >= 10batch=@pages.dup@pages.clearemit(batch)endendcow_fork:process_pagesdo |batch|
# Process pages in parallel using forked processesbatch.each{ |page| process_content(page)}endendend

License

The gem is available as open source under the terms of the MIT License.