feat(engine)!: execute loop and iteration in engine frames#197
Draft
laipz8200 wants to merge 9 commits into
Draft
feat(engine)!: execute loop and iteration in engine frames#197laipz8200 wants to merge 9 commits into
laipz8200 wants to merge 9 commits into
Conversation
Move loop and iteration execution into the main graph engine frame model. Add frame-aware task routing and event handling so container bodies run without creating nested engines. Extend DSL loading for container node types and add workflow event scenarios for complete graphs.
Rename ReadyQueueProtocol to ReadyQueue and update direct imports.
Add field comments for loop and iteration run contexts.
Keep nested iteration outputs when flatten_output is disabled.
Route LoopNode and IterationNode through the normal BaseNode.run() generator path. Add explicit start/resume ready tasks, WorkerPool-owned suspended invocation storage, and semantic container await/result models so child-frame execution can complete without occupying a worker thread.
Avoid firing GraphEngineLayer.on_node_run_end when a cooperative container node yields an await request. The hook now runs only when the invocation actually completes after resume.
Protect ContainerExecution bookkeeping with a reentrant lock because worker threads start container awaits while the dispatcher completes child frames. Add a regression test that blocks complete_frame and verifies concurrent start_container_await waits.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Important
!Related Issue
close #198
Summary
Breaking Changes
LoopNodeandIterationNodeno longer execute child graphs from their own_run()methods. Their bodies are now interpreted byGraphEnginethrough frame-aware execution.frame_idwithnode_id. Custom ready queue, dispatcher, worker, or event-handler integrations must preserve the frame when scheduling or handling node work.GraphRuntimeState. Custom node factories used with loop or iteration graphs must providewith_runtime_state(graph_runtime_state).NodeExecution.execution_idis now required when the domain model is created, andNodeExecution.mark_started()no longer accepts an execution id.ensure_execution_id; the method has been removed. The worker binds the frame-scoped execution id before callingnode.run().Downstream Adaptation
GraphEngine.run()orgraphon.dsl.loads(...).run()instead of callingLoopNode.run()orIterationNode.run()directly.in_loop_idorin_iteration_id.ReadyTask(frame_id, node_id)values. Userootfor normal top-level work.with_runtime_state()method that returns an equivalent factory bound to the supplied runtime state.NodeExecutiondirectly, provideexecution_idup front. PreferGraphExecution.get_or_create_node_execution()when possible.node_execution.mark_started(execution_id)withnode_execution.mark_started().node.ensure_execution_id()usage with worker-style binding: get the frame-scoped node execution, callnode.bind_execution_id(node_execution.execution_id), then callnode.run().Checklist
!