diff --git a/.gitignore b/.gitignore index 34bb32d..8576595 100644 --- a/.gitignore +++ b/.gitignore @@ -49,6 +49,7 @@ pickle-email-*.html .envrc .ruby-gemset .rspec_status +.claude # Redis DB dump.rdb diff --git a/TODOS.md b/TODOS.md index 266af13..d310bd5 100644 --- a/TODOS.md +++ b/TODOS.md @@ -22,10 +22,6 @@ TODO: Refactor so more things are moved from Stage to Worker, e.g. - sending of end signals? - rename #run_worker_loop as its not a loop. Maybe #run_in_worker? other ideas? --> DONE: renamed to #run_stage -======================================= - -@stage_name == :_entrance and :_exit (YUCK) - ================================ This needs to be in all stage: diff --git a/lib/minigun/pipeline.rb b/lib/minigun/pipeline.rb index 4b7f1c8..d66dfce 100644 --- a/lib/minigun/pipeline.rb +++ b/lib/minigun/pipeline.rb @@ -20,6 +20,7 @@ def initialize(name, task, parent_pipeline, config = {}, stages: nil, hooks: nil @stages = stages || [] @deferred_edges = [] # Store edges with forward references: [{from: stage, to: :name}, ...] + @entrance_router = nil # Router stage for nested pipeline entrance (if multiple entry stages) # Pipeline-level hooks (run once per pipeline) @hooks = hooks || { @@ -306,13 +307,24 @@ def run_pipeline(_context) def build_stage_input_queues queues = {} + # Find entrance infrastructure (router or single entry stage) + entry_stages = @stages.select do |s| + s.run_mode != :autonomous && @dag.upstream(s).empty? + end + @stages.each do |stage| # Skip autonomous stages - they don't have input queues next if stage.run_mode == :autonomous - # Special case: EntranceStage uses the parent pipeline's input queue - if stage.is_a?(EntranceStage) && @input_queues && @input_queues[:input] - queues[stage] = @input_queues[:input] # Key by object + # Entrance router uses PipelineStage's queue + if stage == @entrance_router && @input_queues && @input_queues[:input] + queues[stage] = @input_queues[:input] + next + end + + # Single entry stage uses PipelineStage's queue directly (no router) + if entry_stages.size == 1 && stage == entry_stages.first && @input_queues && @input_queues[:input] + queues[stage] = @input_queues[:input] next end @@ -425,7 +437,7 @@ def build_dag_routing! # Fill any remaining sequential gaps (handles fan-out, siblings, cycles) fill_sequential_gaps_by_definition_order! - # If this pipeline has input_queues (nested pipeline), add :_entrance distributor + # If this pipeline has input_queues (nested pipeline), add entrance distributor insert_entrance_distributor_for_inputs! if @input_queues && !@input_queues.empty? # If this pipeline has output_queues, add :_exit collector for terminal stages @@ -504,40 +516,44 @@ def fill_sequential_gaps_by_definition_order! end end - # Insert an :_entrance distributor stage for nested pipelines - # This receives items from the parent pipeline and distributes to entry stages + # Insert an entrance distributor for nested pipelines + # Uses RouterStage for multiple entries, direct connection for single entry def insert_entrance_distributor_for_inputs! # Find stages that have no upstream (would be entry points) - # After normalization, DAG uses Stage objects entry_stages = @stages.select do |stage| - # Skip autonomous stages (they're producers) next false if stage.run_mode == :autonomous - # Entry stages have no upstream @dag.upstream(stage).empty? end return if entry_stages.empty? - # Create a consumer stage that reads from parent input and emits to nested pipeline - # Use ConsumerStage (not ProducerStage) so it properly tracks multiple END signals - parent_input = @input_queues[:input] - entrance_block = proc do |item, output| - # Just forward items from parent to nested pipeline - output << item - end - entrance_stage = Minigun::EntranceStage.new(nil, self, entrance_block, {}) + # Single entry stage: it uses PipelineStage's queue directly (no router needed) + # This is handled in build_stage_input_queues + return if entry_stages.size == 1 + + # Multiple entry stages: create a router to distribute items + # Get routing strategy from parent_pipeline's config (where PipelineStage lives) + pipeline_stage = @parent_pipeline&.stages&.find { |s| s.is_a?(PipelineStage) && s.nested_pipeline == self } + routing_strategy = pipeline_stage&.options&.[](:routing) || :broadcast + + # Create anonymous router stage + @entrance_router = if routing_strategy == :round_robin + RouterRoundRobinStage.new(nil, self, entry_stages.dup, {}) + else + RouterBroadcastStage.new(nil, self, entry_stages.dup, {}) + end - # Add the :_entrance stage to the pipeline (at the beginning) - @stages.unshift(entrance_stage) - @dag.add_node(entrance_stage) + # Add router to pipeline + @stages.unshift(@entrance_router) + @dag.add_node(@entrance_router) - # Connect :_entrance to entry stages (all objects) + # Connect router to entry stages entry_stages.each do |stage| - @dag.add_edge(entrance_stage, stage) + @dag.add_edge(@entrance_router, stage) end entry_names = entry_stages.map(&:name).join(', ') - log_debug "[Pipeline:#{@name}] Added :_entrance distributor for entry stages: #{entry_names}" + log_debug "[Pipeline:#{@name}] Added #{routing_strategy} entrance router for entry stages: #{entry_names}" end # Insert an :_exit collector stage that terminal stages drain into diff --git a/lib/minigun/stage.rb b/lib/minigun/stage.rb index a71b761..ec8aee5 100644 --- a/lib/minigun/stage.rb +++ b/lib/minigun/stage.rb @@ -444,15 +444,6 @@ def run_stage(worker_ctx) end end - # Special entrance stage for nested pipelines - # Automatically created when a pipeline has input from parent - class EntranceStage < ConsumerStage - # Positional constructor: EntranceStage.new(name, pipeline, block, options) - def initialize(name, pipeline, block, options = {}) - super(name, pipeline, block, options) - end - end - # Special exit stage for nested pipelines # Automatically created when a pipeline has output to parent class ExitStage < ConsumerStage @@ -482,10 +473,10 @@ def run_stage(stage_ctx) return unless @nested_pipeline # Set up input/output queues for the nested pipeline - # The pipeline will create :_entrance and :_exit stages based on these + # Pass the PipelineStage's input queue to nested pipeline so entry stages can use it if !stage_ctx.sources_expected.empty? - # Has upstream: set input queue so pipeline creates :_entrance - # Also pass the expected source count for proper END signal handling + # Has upstream: pass input queue to nested pipeline + # The nested pipeline will handle distributing to its entry stages @nested_pipeline.instance_variable_set(:@input_queues, { input: stage_ctx.input_queue, sources_expected: stage_ctx.sources_expected @@ -495,7 +486,7 @@ def run_stage(stage_ctx) # Always set output queue so pipeline creates :_exit @nested_pipeline.instance_variable_set(:@output_queues, { output: create_output_queue(stage_ctx) }) - # Run the nested pipeline (it will automatically create :_entrance/:_exit as needed) + # Run the nested pipeline (it will handle input distribution to entry stages) @nested_pipeline.run(stage_ctx.root_pipeline.context) ensure send_end_signals(stage_ctx) diff --git a/lib/minigun/worker.rb b/lib/minigun/worker.rb index aee4946..7f72827 100644 --- a/lib/minigun/worker.rb +++ b/lib/minigun/worker.rb @@ -83,11 +83,20 @@ def create_stage_context # DAG now uses Stage objects instead of names sources_expected = if @stage.run_mode == :autonomous Set.new - elsif @stage.is_a?(Minigun::EntranceStage) && @pipeline.input_queues - # For EntranceStage, use sources from parent pipeline if available - @pipeline.input_queues[:sources_expected] || Set.new else - Set.new(dag.upstream(@stage)) + # Check if this stage is an entrance router or single entry stage for nested pipeline + input_queues = @pipeline.instance_variable_get(:@input_queues) + entrance_router = @pipeline.instance_variable_get(:@entrance_router) + + if @stage == entrance_router && input_queues + # For entrance router, use sources from parent pipeline + input_queues[:sources_expected] || Set.new + elsif dag.upstream(@stage).empty? && input_queues && input_queues[:sources_expected] + # For single entry stage with no upstream, use sources from parent pipeline if available + input_queues[:sources_expected] + else + Set.new(dag.upstream(@stage)) + end end # Create stats object for this specific stage