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

    • drainCondition

      DrainCondition drainCondition(String sourceVar, String sourceTypeName, String hasNextMethod, DebugInfo debugInfo)
      THE drain-loop condition: source.hasNext() evaluated in its own scope. Shared by every place that drains an iterator or an async executor into a buffer.

    • drainNextIntoItem

      void drainNextIntoItem(List<IRInstr> body, String sourceVar, String sourceTypeName, String nextMethod, 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.

    • addReferenceAndRegister

      void addReferenceAndRegister(List<IRInstr> target, boolean typed, String var, String type, String scopeId, DebugInfo debugInfo)
      Declare var of type in scopeId (a typed reference when typed, otherwise a plain reference) and SCOPE_REGISTER it. Single-sources the repeated reference-then-register pair used when binding pipeline/loop variables.

    • appendBoundedCondition

      String appendBoundedCondition(List<IRInstr> conditionEvaluation, List<IRInstr> leftInstructions, String leftResult, String leftPrimitive, List<IRInstr> rightInstructions, String rightResult, String scopeId, DebugInfo debugInfo)
      Append a short-circuit LEFT AND RIGHT Boolean condition to conditionEvaluation and return the combined Boolean result variable. RIGHT is evaluated ONLY when LEFT's primitive is true (LOGICAL_AND_BLOCK short-circuit). This is the generic two-operand AND-block; callers build LEFT (the short-circuit decision: head's counter < limit, the window's inFlight < W, etc.) and pass its result+primitive, plus the RIGHT operand's instructions+result (e.g. the managed hasNext()). Mirrors GuardedConditionEvaluator's guard-into-condition fold.

    • appendHeadBoundedCondition

      String appendHeadBoundedCondition(List<IRInstr> conditionEvaluation, List<IRInstr> managedHasNext, String hasNextResult, StageState headState, String scopeId, DebugInfo debugInfo)
      Append a short-circuit counter < limit AND hasNext() condition to conditionEvaluation and return the combined Boolean result variable. The counter ('LEFT') is evaluated first; hasNext() ('RIGHT') is evaluated ONLY when the counter is still below the head limit, so a satisfied head N stops the loop without another (possibly blocking) hasNext()/next() on the source. Mirrors the short-circuit LOGICAL_AND_BLOCK that GuardedConditionEvaluator uses to fold a guard into a condition.

    • buildStreamLoop

      ConditionAndPostLoop buildStreamLoop(IteratorSetup iteratorSetup, StageState headState, String itemVar, ISymbol itemType, boolean rawIterator, EK9Parser.StreamCatContext catCtx, StreamLoopSink sink, DebugInfo debugInfo)
      The unified iterator-driven stream while-loop skeleton: condition scope (hasNext(), with an optional head short-circuit folded in when headState != null) + body scope (bind iterator.next() then the StreamLoopSink body-tail) + the ConditionAndPostLoop the caller feeds to createWhileLoop. Every iterator-based stream loop (streaming cat/for, and — after migration — the sort/group/split drains) is this skeleton differing only in headState and sink.

    • 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(String sourceVar, String sourceTypeName, String hasNextMethod, String nextMethod, CallDetails closeCall, 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. A closeCall (async executor) fires after the drain; trackers pass null. 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.

    • 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).

    • findHeadStageState

      StageState findHeadStageState(List<EK9Parser.StreamPartContext> stages, List<StageState> stageStates)
      Locate the StageState of the first HEAD stage (carries counter/limit), or null if no HEAD present.

    • freshBooleanOperand

      FreshBooleanOperand freshBooleanOperand(String boolVar, DebugInfo debugInfo)
      Load an alloca-backed Boolean variable into a fresh SSA temp (+ its primitive) for use as the LEFT operand of appendBoundedCondition(List, List, String, String, List, String, String, DebugInfo). See FreshBooleanOperand.

    • 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)
      Generate pre-loop state initialization for stateful pipeline stages (HEAD, SKIPPING, TAIL). Creates counter, limit, and one-constant variables for HEAD/SKIP stages. For TAIL, creates a bounded Ek9Tail circular buffer via Ek9Tail.create(tailLimit). Returns pre-loop instructions and per-stage state (null for stateless stages).

    • 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

      SortAndIteratorResult sortBufferAndGetIterator(EK9Parser.StreamPartContext sortStage, String bufferVar, DebugInfo debugInfo)
      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
      Returns:
      instructions for the sort call and iterator creation, plus the new iterator setup

    • 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).