Skip to content

Conversation

aglinxinyuan
Copy link
Contributor

@aglinxinyuan aglinxinyuan commented Jun 16, 2025

Main changes for the Basic For Loop:

  1. Loop Operator:
  • LoopStart Operator
  • LoopEnd Operator
  1. ECM:
  • IterationEnd
  1. DCM:
  • NextIteration

Life Cycle of Basic For Loop:
Untitled Diagram drawio (1)

@aglinxinyuan aglinxinyuan changed the base branch from master to xinyuan-rename-ecm June 17, 2025 02:06
Base automatically changed from xinyuan-rename-ecm to master June 17, 2025 03:47
@aglinxinyuan aglinxinyuan self-assigned this Jun 22, 2025
@aglinxinyuan aglinxinyuan changed the title Add Support for Basic For Loop feat(core): support for loop Jul 3, 2025
@aglinxinyuan aglinxinyuan changed the title feat(core): support for loop feat: support for loop Jul 3, 2025
@aglinxinyuan aglinxinyuan marked this pull request as ready for review July 7, 2025 01:17
Copy link
Contributor

@Copilot Copilot AI left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR adds support for basic for-loop semantics by introducing LoopStart and LoopEnd operators, new iteration control messages, and the necessary execution and RPC plumbing in both Scala and Python runtimes.

  • Introduced LoopStart/LoopEnd executors and descriptors to buffer and replay tuples across iterations
  • Defined EndIteration and NextIteration control messages and integrated them into DataProcessor, OutputManager, and RPC handlers
  • Updated Python runner, protobuf definitions, and internal models to handle iteration lifecycle

Reviewed Changes

Copilot reviewed 25 out of 28 changed files in this pull request and generated 6 comments.

Show a summary per file
File Description
core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/sleep/SleepOpExec.scala Added Sleep operator executor
core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/sleep/SleepOpDesc.scala Defined Sleep operator descriptor
core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/loop/LoopStartOpExec.scala Implemented LoopStart executor
core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/loop/LoopStartOpDesc.scala Defined LoopStart descriptor
core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/loop/LoopEndOpExec.scala Implemented LoopEnd executor
core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/loop/LoopEndOpDesc.scala Defined LoopEnd descriptor
core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/LogicalOp.scala Imported new loop and sleep operators
core/workflow-core/src/main/scala/edu/uci/ics/amber/core/tuple/TupleLike.scala Added FinalizeIteration marker
core/workflow-core/src/main/scala/edu/uci/ics/amber/core/executor/OperatorExecutor.scala Added reset lifecycle method
core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/worker/promisehandlers/StartChannelHandler.scala Refactored start-channel handling
core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/worker/promisehandlers/NextIterationHandler.scala Added NextIteration RPC handler
core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/worker/promisehandlers/EndIterationHandler.scala Added EndIteration RPC handler
core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala Updated end-channel to finalize loops
core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/worker/DataProcessorRPCHandlerInitializer.scala Registered iteration handlers
core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/worker/DataProcessor.scala Integrated iteration control flow and executor reset
core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/messaginglayer/OutputManager.scala Added finalizeIteration API
core/amber/src/main/python/proto/edu/uci/ics/amber/engine/architecture/rpc/init.py Extended Python protobuf with EndIteration
core/amber/src/main/python/core/runnables/main_loop.py Handled EndIteration in main loop
core/amber/src/main/python/core/runnables/data_processor.py Treated EndIteration as InternalMarker
core/amber/src/main/python/core/models/operator.py Cleared table buffer in on_finish
core/amber/src/main/python/core/models/internal_marker.py Added EndIteration marker class
core/amber/src/main/python/core/architecture/rpc/async_rpc_handler_initializer.py Registered Python EndIterationHandler
core/amber/src/main/python/core/architecture/handlers/control/end_iteration_handler.py Implemented Python EndIteration handler
core/amber/src/main/protobuf/edu/uci/ics/amber/engine/architecture/rpc/workerservice.proto Added EndIteration/NextIteration RPCs
core/amber/src/main/protobuf/edu/uci/ics/amber/engine/architecture/rpc/controlcommands.proto Added EndIterationRequest message
Comments suppressed due to low confidence (1)

core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/sleep/SleepOpDesc.scala:34

  • [nitpick] The JSON schema title "n" is ambiguous. Consider using a more descriptive title like "time" or "duration" to clarify the property purpose.
  @JsonSchemaTitle("n")

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Projects
None yet
Development

Successfully merging this pull request may close these issues.

1 participant