Class StreamPipelineMachinery

java.lang.Object
org.ek9lang.compiler.phase7.generator.AbstractGenerator
org.ek9lang.compiler.phase7.generator.AbstractStreamStageGenerator
org.ek9lang.compiler.phase7.generator.StreamPipelineMachinery
final class StreamPipelineMachinery extends AbstractStreamStageGenerator
The pipeline machinery shared by the orchestrator and the buffered-emission engine: draining a source to an iterator, planning the per-stage item variables, building the stream loop, and setting up per-stage pre-loop state.

None of it touches StreamStageDispatcher, which is what lets it be a plain collaborator both sides hold. Anything here that DID need the dispatcher would have to live with the engine instead - that dependency is what separates the two.

  • Constructor Details

  • Method Details

    • demand

      StreamDemand demand()
      THE demand rule (head bounds the pull) shared by every loop this machinery and the engine build.

    • drainCondition

      DrainCondition drainCondition(HelperSource source, List<StageState.Counter> heads, DebugInfo debugInfo)
      THE drain-loop condition: source.hasNext() evaluated in its own scope, bounded by heads (rule B): c < l AND ... source.hasNext(), the counters first. With no heads it is the plain call. The returned hasNextVar is the primitive the loop branches on. Shared by every loop over a runtime helper (helperDrainLoop(HelperSource, String, String, List, Supplier, DebugInfo)).

    • drainNextIntoItem

      void drainNextIntoItem(List<IRInstr> body, HelperSource source, String castTypeName, String drainItemVar, DebugInfo debugInfo)
      THE drain-loop body preamble: raw = source.next() (typed Any), cast to the element type, then bound into the persistent drain item variable.

      The item variable is declared in the DRAIN scope, not the body scope, so whatever the body pipes into a buffer outlives the iteration that produced it. Must be called with the body scope already entered - the cast registers against the current scope.

    • helperDrainBody

      StageWrapResult helperDrainBody(HelperSource source, String castTypeName, String itemVar, Supplier<StageWrapResult> downstream, DebugInfo debugInfo)
      THE body of a loop over a runtime helper: its own per-element scope, item = (castType) source.next() (drainNextIntoItem(List, HelperSource, String, String, DebugInfo)), then what downstream builds - generated once that scope is current, so what it makes per element is owned by the iteration (A11).
      Returns:
      the body (scope enter and exit included) and the downstream's post-loop instructions

    • helperDrainLoop

      ConditionAndPostLoop helperDrainLoop(HelperSource source, String castTypeName, String itemVar, List<StageState.Counter> heads, Supplier<StageWrapResult> downstream, DebugInfo debugInfo)
      THE loop over a runtime helper - the twin of buildStreamLoop(IteratorSetup, List, String, ISymbol, boolean, ParseTree, StreamLoopSink, DebugInfo) for an HelperSource rather than an iterator: the drainCondition(HelperSource, List, DebugInfo) (bounded by heads), then the helperDrainBody(HelperSource, String, String, Supplier, DebugInfo). Every buffered emission (the async executor's results, a group/split's buckets, the tail buffer, the drain that materialises a buffer) is this loop; only the item, the heads and the downstream differ. The condition case names the helper's own hasNext() result as its condition result.

    • declareElementSlot

      String declareElementSlot(List<IRInstr> target, boolean typed, String typeName, String scopeId, DebugInfo debugInfo)
      THE declaration of a pipeline's flowing ELEMENT slot - rebound every element (a source item, an emitted item, a drained bucket item, a current group): a compiler-made slot the declaring scope owns, which drops its LAST value at exit and on unwind (A10; EK9_IR_OWNERSHIP_MODEL.md §6), so nothing releases it after the loop. Named as a slot, never _temp: a native back end releases a registered _temp with the value it held when registered, a slot with its current value. typed gives a TYPED_REFERENCE (a join's ASM stackmap).
      Returns:
      the slot's name

    • declareOwnedSlot

      void declareOwnedSlot(List<IRInstr> target, boolean typed, String slot, String typeName, String scopeId, DebugInfo debugInfo)
      declareElementSlot(List, boolean, String, String, DebugInfo) for a slot that already has its name - a stream's own for loop variable.

    • declareInitialisedSlot

      String declareInitialisedSlot(List<IRInstr> target, ISymbol type, String literal, DebugInfo debugInfo)
      THE declaration of a compiler-made, initialised SLOT - a counter or flag rebound as the pipeline runs (a head or skip counter, the async in-flight and submitted counts, the windowed-async more flag): REFERENCE, STORE of the initial value, RETAIN, then the slot registered in the current scope, which releases its LAST value on exit and on unwind (A10; EK9_IR_OWNERSHIP_MODEL.md §6). Each rebind (RELEASE/STORE/ RETAIN) releases the value it replaces. The initial value is the literal, loaded first (AbstractStreamStageGenerator.loadManagedLiteral(List, String, ISymbol, DebugInfo)).
      Returns:
      the slot's name

    • withPipelineScope

      void withPipelineScope(List<EK9Parser.StreamPartContext> stages, boolean typedItem, String itemTypeName, DebugInfo debugInfo, List<IRInstr> instructions, PipelineScopeBody body)
      THE pipeline scope frame: mint and enter a scope, declare the pipeline's element slot in it (declareElementSlot(List, boolean, String, String, DebugInfo)), emit the pre-loop state of stages (generateStageSetup(List, String, DebugInfo)), run body (the loop or drain), then exit the scope - which drops the slot's last value. The streaming pipeline, the buffered for-range emission and its drains all open this frame; never hand-roll it.

    • emitWhileLoop

      void emitWhileLoop(ConditionCaseDetails conditionCase, String loopScopeId, DebugInfo debugInfo, List<IRInstr> instructions)
      THE while loop of a stream loop: conditionCase as a WHILE over loopScopeId, appended to instructions. Every stream drain, pull and emission loop is emitted through this (or emitStreamLoop(ConditionAndPostLoop, String, DebugInfo, List), which adds the loop's post-loop work) - never a hand-built createWhileLoop + controlFlowChainGenerator pair.

    • emitStreamLoop

      void emitStreamLoop(ConditionAndPostLoop loop, String loopScopeId, DebugInfo debugInfo, List<IRInstr> instructions)
      THE emission of a built stream loop: its while (emitWhileLoop(ConditionCaseDetails, String, DebugInfo, List)) then its post-loop instructions (a stage that emits after the loop - tail's drain, join's result).

    • buildStreamLoop

      ConditionAndPostLoop buildStreamLoop(IteratorSetup iteratorSetup, List<StageState.Counter> demandBounds, String itemVar, ISymbol itemType, boolean rawIterator, org.antlr.v4.runtime.tree.ParseTree probeSite, StreamLoopSink sink, DebugInfo debugInfo)
      The unified iterator-driven stream while-loop skeleton: condition scope (hasNext(), with the demand bounds - the segment's heads, or the async window - folded in ahead of it by StreamDemand) + body scope (bind iterator.next() then the StreamLoopSink body-tail) + the ConditionAndPostLoop the caller emits with emitStreamLoop(ConditionAndPostLoop, String, DebugInfo, List). Every iterator-based stream loop (streaming cat/for, and — after migration — the sort/group/split drains) is this skeleton differing only in its demand bounds and sink.

      probeSite is where the loop's LOOP_BODY coverage probe is: the pipeline's source as written (its cat or for), whichever compiler-made loop of that pipeline this is - its id is hashed from that location, so two pipelines are two probes (defect D88). Never null: a probe with no location has no line, and every such probe in one function would share one id.

    • buildUnprobedStreamLoop

      ConditionAndPostLoop buildUnprobedStreamLoop(IteratorSetup iteratorSetup, List<StageState.Counter> demandBounds, String itemVar, ISymbol itemType, boolean rawIterator, StreamLoopSink sink, DebugInfo debugInfo)
      buildStreamLoop(IteratorSetup, List, String, ISymbol, boolean, ParseTree, StreamLoopSink, DebugInfo) with NO loop-body coverage probe - the emission of a sorted bucket buffer (StreamEmissionEngine.bufferSortEmit), which has never carried one.

    • containsBufferBarrier

      boolean containsBufferBarrier(List<EK9Parser.StreamPartContext> stages)
      True if any stage is a pre-loop buffer-barrier (sort/group/split/tail) — these cannot be composed by the windowed async driver and (on an unbounded async) are rejected at phase 3.

    • containsBufferingCollection

      boolean containsBufferingCollection(List<EK9Parser.StreamPartContext> stages)
      True iff stages contains a GROUP/SPLIT/ASYNC collection-buffer stage.

    • drainSourceToIterator

      IteratorSetup drainSourceToIterator(HelperSource source, ISymbol bucketType, DebugInfo debugInfo, List<IRInstr> instructions)
      MATERIALISE a buffering value source (a GROUP/SPLIT tracker's hasNextGroup/nextGroup, or an async executor's hasNext/next) into a fresh raw EK9_LIST buffer and return an IteratorSetup over it. No pre-stages, no sort: each next() yields Any, is cast to bucketType, and piped into the buffer as-is. Nothing is closed here: an async executor is closed by its pipeline's finally region (StreamEmissionEngine). Used to hand a collection-buffer's emitted elements to StreamEmissionEngine.continueOverIterator so a DOWNSTREAM buffer (group|split, group|split|sort, async|split, ...) can drive them — the basis of unlimited buffer-after-buffer composition.

    • drainSourceToIterator

      IteratorSetup drainSourceToIterator(HelperSource source, ISymbol bucketType, boolean keepsUnset, DebugInfo debugInfo, List<IRInstr> instructions)
      As above; keepsUnset appends with += so an UNSET element survives the buffer - a stage never drops one (KeepsUnsetElements). An async result may be unset and passes through like a map's (Steve, 2026-10-21); a group/split bucket is a non-empty List, never unset, so those callers keep |.

    • rawListIterator

      IteratorSetup rawListIterator(String bufferVar, DebugInfo debugInfo, List<IRInstr> instructions)
      THE iterator over a stream's raw org.ek9.lang::List buffer: buffer.iterator() into a fresh managed temp, typed as the raw Iterator over the raw List - so the loops that read it cast each next() (an Any) to the element type. Every buffer a pipeline materialises (a drained helper, a sorted buffer, a for-range buffer) is iterated through this.

    • edgeConsumes

      ISymbol edgeConsumes(EK9Parser.StreamPartContext stage)
      The phase-3-resolved IN-edge (consumes) type of a stream stage, or null if unresolved. Counterpart to edgeProduces(EK9Parser.StreamPartContext) — the element type ENTERING stage k.

    • edgeProduces

      ISymbol edgeProduces(EK9Parser.StreamPartContext stage)
      The phase-3-resolved OUT-edge (produces) type of a stream stage, or null if the stage has no recorded StreamCallSymbol or no produced type set. This is the SINGLE source of truth for "the element type leaving stage k": Phase 3 (StreamAssemblyOrError.streamPartOrError) stamps setProducesSymbolType on EVERY stage's StreamCallSymbol while threading currentStreamType, so the whole typed-edge sequence already exists — read it here rather than re-deriving from the source type carried forward. edgeConsumes(EK9Parser.StreamPartContext) is the IN-edge counterpart.

    • effectivePostSortElementType

      ISymbol effectivePostSortElementType(SortSetup sortSetup, List<EK9Parser.StreamPartContext> allStages, ISymbol sourceType)
      The element type the POST-sort pipeline (and terminal) observe. SORT buffers and re-emits the type it consumes; when a type-changing stage (map/flatten/call) precedes SORT, that is the post-stage type, not the source type. Returns the SORT stage's resolved produced type when SORT is present, else the supplied source type unchanged (so non-sort pipelines and natural-type sorts are byte-identical).

    • generateDynamicStageLimit

      List<IRInstr> generateDynamicStageLimit(EK9Parser.StreamPartContext stageCtx, String limitVar, ISymbol integerType, String scopeId, DebugInfo debugInfo)
      Generate IR to evaluate a dynamic stage operand (function reference or call) and store the Integer result in the given limit variable.

      For function references (head GetThree): creates a function instance and calls it. For function calls (head GetThree()): evaluates the call expression. For Integer variables (head someVar): loads the variable value.

    • generatePipelineItemBinding

      List<IRInstr> generatePipelineItemBinding(String iteratorVar, IAggregateSymbol iteratorType, String pipelineItemVar, ISymbol elementType, DebugInfo debugInfo, boolean rawIterator)
      Generate: pipelineItem = iterator.next() Follows ForInGenerator.generateLoopVariableBinding() pattern.
      Parameters:
      rawIterator - true when iterating a raw org.ek9.lang.Iterator (buffered for-range), where next() returns Any and requires a cast to elementType

    • generateStageSetup

      StageSetupResult generateStageSetup(List<EK9Parser.StreamPartContext> stages, String scopeId, DebugInfo debugInfo)

    • closeSource

      List<IRInstr> closeSource(CallDetails closeCall, DebugInfo debugInfo)
      THE close of a drained stream source (the async executor), in its pipeline's finally region. Ek9Async.close() hands back its report String at +1: it is named and dropped at once - discarded, nothing released it (the per-path verifier's discarded-result rule; Linux LSan: every async pipeline leaked one String per run).

    • indexOfFirstCollectionBuffer

      int indexOfFirstCollectionBuffer(List<EK9Parser.StreamPartContext> stages)
      Index of the first GROUP/SPLIT/ASYNC collection-buffer stage in stages, or -1.

    • indexOfFirstSort

      int indexOfFirstSort(List<EK9Parser.StreamPartContext> stages)
      Index of the first SORT stage in stages, or -1.

    • planPipelineVars

      PipelineVarPlan planPipelineVars(List<EK9Parser.StreamPartContext> stages, String sourceVar, ISymbol sourceType, String scopeId, DebugInfo debugInfo, List<IRInstr> declarationsOut)
      Compute the per-boundary item-variable plan for an in-loop stage list, allocating a fresh variable (declared at the produced type) only where a stage changes the element type. Fresh-variable declarations (REFERENCE + SCOPE_REGISTER) are appended to declarationsOut; the caller emits them once before the loop body. Type-preserving stages reuse the incoming variable, so pipelines without a type-changing stage produce byte-identical IR to before.

    • sortBufferAndGetIterator

      IteratorSetup sortBufferAndGetIterator(EK9Parser.StreamPartContext sortStage, String bufferVar, DebugInfo debugInfo, List<IRInstr> instructions)
      Sort a raw org.ek9.lang::List buffer in place and return an iterator over the sorted result. Shared by both cat-pipeline and for-range pipeline sort paths. Uses Ek9Sort.sort(List) for natural sort or Ek9Sort.sort(List, Any) for comparator sort. No LOAD_FIELD delegate access — works correctly for both pre-built and generated types.
      Parameters:
      sortStage - the SORT stage parse context (may have pipelinePart for comparator)
      bufferVar - variable holding the raw org.ek9.lang::List buffer
      debugInfo - debug information for generated instructions
      instructions - receives the sort call and the iterator creation
      Returns:
      the iterator over the sorted buffer

    • sortReachableBeforeBuffer

      boolean sortReachableBeforeBuffer(List<EK9Parser.StreamPartContext> stages)
      True iff the NEXT buffering SORT in stages is reachable with no intervening collection-buffer (GROUP/SPLIT/ASYNC). Those three stages own the sort that follows them — they buffer and emit in their own pre-loop path, where a downstream SORT is handled separately — so the source-drain sort loop must NOT drain through them (it would treat the GROUP/SPLIT/ASYNC as a no-op pre-sort stage and corrupt the stream). A SORT before any such buffer IS the source-drain loop's to handle. Used as the multi-sort loop condition so sort | map | sort keeps draining but sort | group | sort stops at the group (leaving the trailing sort on postSortStages for the group path).