Migrate scala control messages to protobuf #2950

shengquan-ni · 2024-10-17T10:00:32Z

This PR migrates all Scala control messages and their corresponding handlers to use gRPC.

Main Changes (For both Scala and Python Engine)

Rewrote AsyncRPCServer and AsyncRPCClient
- Adjusted both to fit into the generated gRPC interface, while keeping RPCs sent and received through actor messages.
Centralized RPC Definitions
- All RPC messages and method definitions are now located in:
```
src/main/protobuf/edu/uci/ics/amber/engine/architecture/rpc
```

Steps to Create New RPCs

Create a request proto message in controlcommands.proto with all necessary inputs.
Create a response proto message in controlreturns.proto.
Define the RPC:
- For controller-handled RPCs, add the definition to controllerservice.proto.
- For worker-handled RPCs, add it to workerservice.proto.
Generate Scala code by running protocGenerate in the SBT shell.
Generate Python code by running ./scripts/python-proto-gen.sh from core folder.
Implement the RPC handler:
- Scala Engine: in either ControllerAsyncRPCHandlerInitializer or WorkerAsyncRPCHandlerInitializer, and move it to the promisehandlers folder for better organization.
- Python Engine: create a new handler under python/core/architecture/handlers/control and override corresponding async method for the new rpc defined in worker service.

Changes in Sending RPCs (Scala Engine)

Example: Using the StartWorkflow RPC, which is handled by the controller.

Before this PR:

send(StartWorkflow(), CONTROLLER).map { ret =>
  // handle the return value
}

After this PR:

controllerInterface.startWorkflow(EmptyRequest(), mkContext(CONTROLLER)).map { resp =>
    // handle the response
}

…xera into shengquan-rpc-refactor

Yicong-Huang

LGTM! Had some tests locally and left some comments in code.

core/amber/build.sbt

core/amber/project/plugins.sbt

core/amber/requirements.txt

core/amber/src/main/protobuf/edu/uci/ics/amber/engine/architecture/rpc/controlcommands.proto

core/amber/src/main/protobuf/edu/uci/ics/amber/engine/architecture/rpc/controlreturns.proto

core/amber/src/main/protobuf/edu/uci/ics/amber/engine/common/actormessage.proto

core/amber/src/main/protobuf/edu/uci/ics/amber/engine/architecture/rpc/controlreturns.proto

core/amber/src/main/python/core/architecture/handlers/actorcommand/backpressure_handler.py

Yicong-Huang

LGTM! Had some tests locally and left some comments in code.

...src/main/scala/edu/uci/ics/amber/engine/architecture/messaginglayer/WorkerTimerService.scala

core/amber/src/main/python/core/architecture/handlers/control/control_handler_base.py

...i/ics/amber/engine/architecture/controller/promisehandlers/TakeGlobalCheckpointHandler.scala

…xera into shengquan-rpc-refactor

core/amber/src/main/python/core/architecture/handlers/control/no_op_handler.py

core/amber/src/main/python/core/architecture/handlers/control/resume_worker_handler.py

core/amber/src/main/python/core/architecture/rpc/async_rpc_server.py

core/amber/src/main/python/core/runnables/main_loop.py

core/amber/src/main/python/core/runnables/test_console_message.py

…xera into shengquan-rpc-refactor

This bug is caused by a wrong import of the `RUNNING` state introduced in #2950. Co-authored-by: Xinyuan Lin <xinyual3@uci.edu>

As discussed in #2950, we plan to remove obsolete RPCs and reconfiguration-related RPCs in this PR. We will bring reconfiguration back after #2950. Removed: 1. QueryCurrentInputTuple. 2. ShutdownDPThread. Disabled, will be added later: 1. Reconfiguration. 2. UpdateExecutor. 3. UpdateMultipleExecutors.

This PR migrates all Scala control messages and their corresponding handlers to use gRPC. --- ### Main Changes (For both Scala and Python Engine) 1. **Rewrote `AsyncRPCServer` and `AsyncRPCClient`** - Adjusted both to fit into the generated gRPC interface, while keeping RPCs sent and received through actor messages. 2. **Centralized RPC Definitions** - All RPC messages and method definitions are now located in: ``` src/main/protobuf/edu/uci/ics/amber/engine/architecture/rpc ``` --- ### Steps to Create New RPCs 1. **Create a request proto message** in `controlcommands.proto` with all necessary inputs. 2. **Create a response proto message** in `controlreturns.proto`. 3. **Define the RPC**: - For **controller-handled RPCs**, add the definition to `controllerservice.proto`. - For **worker-handled RPCs**, add it to `workerservice.proto`. 4. **Generate Scala code** by running `protocGenerate` in the SBT shell. 5. **Generate Python code** by running `./scripts/python-proto-gen.sh` from `core` folder. 6. **Implement the RPC handler**: - Scala Engine: in either `ControllerAsyncRPCHandlerInitializer` or `WorkerAsyncRPCHandlerInitializer`, and move it to the `promisehandlers` folder for better organization. - Python Engine: create a new handler under `python/core/architecture/handlers/control` and override corresponding `async` method for the new rpc defined in worker service. --- ### Changes in Sending RPCs (Scala Engine) Example: Using the `StartWorkflow` RPC, which is handled by the controller. - **Before this PR:** ```scala send(StartWorkflow(), CONTROLLER).map { ret => // handle the return value } ``` - **After this PR:** ```scala controllerInterface.startWorkflow(EmptyRequest(), mkContext(CONTROLLER)).map { resp => // handle the response } ``` Co-authored-by: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com>

This bug is caused by a wrong import of the `RUNNING` state introduced in #2950. Co-authored-by: Xinyuan Lin <xinyual3@uci.edu>

This PR fixes a bug where the frontend incorrectly shows an operator as still running, even though the backend execution has completed. <img width="80%" alt="截屏2025-06-05 下午3 44 23" src="https://github.com/user-attachments/assets/63865886-1b24-4f36-8ebf-155c2748f98e" /> ### Timeline for this bug This issue was originally identified and fixed in #2411. However, it re-emerged after we migrated our RPC layer to gRPC in #2950. A minor mistake during the migration caused the originally chained future to be overridden by an `EmptyResponse` as the return value. ### Root Cause The core issue is that the `controllerInitiatedQueryStats` call returns immediately instead of waiting for the worker's response. Previously, we relied on this call to collect worker stats after execution, and used the updated stats to infer workflow completion. Because the call now returns immediately, the stats are never updated. Additionally, our current design infers region completion from port completion and workflow completion from region completion. This decouples workflow execution state from the actual state of the workers. As a result, a workflow may be marked as complete even though the frontend still shows an operator as running, when in fact, the worker has already finished execution. ### Example Scenario Here’s how the issue manifests when the last operator finishes: 1. A port finishes, triggering a handler that marks the port as completed. 2. A `QueryStats` control message is sent to the worker that reported port completion (response pending). 3. We check if all ports in the region are complete—if so, the region is marked complete. 4. Separately, we receive a worker execution completion event and send another `QueryStats` message (response also pending). 5. Since all ports are marked complete, the workflow is marked complete and the entire execution is terminated. 6. Both pending `QueryStats` messages are lost because the worker is already shut down. ### The Fix The solution is straightforward: correctly chain and return the future so that the control message awaits the worker’s response before proceeding. ### How the fix is verified The bug is not always reproducible because sometimes the execution is killed after updating the states. So I looked at the WebSocket event sequence. In the correct behavior, setting the workflow state to `complete` should be after all the stats updates. Before the fix: <img width="100%" alt="截屏2025-06-05 下午3 50 34" src="https://github.com/user-attachments/assets/da996255-743d-4cda-8a07-85bd9a9d1d3d" /> After the fix: <img width="825" alt="截屏2025-06-05 下午3 52 05" src="https://github.com/user-attachments/assets/6b64660b-547f-4f19-8769-16a7f0676031" />

shengquan-ni and others added 15 commits October 9, 2024 15:14

update

6cf1e75

update

d86a017

update

36291c6

update

351e41d

update

e1da80f

update

3fb852f

update

59182e2

update

b0ae882

update

57e00bb

update

419c3be

update

bab902c

Merge branch 'master' into shengquan-rpc-refactor

7b5f5b1

update proto

c686a73

fix merge conflict

5fad8b9

Merge branch 'master' into shengquan-rpc-refactor

428a1f5

shengquan-ni self-assigned this Oct 17, 2024

shengquan-ni and others added 8 commits October 17, 2024 03:03

fix format

fbd9ec1

Merge branch 'shengquan-rpc-refactor' of https://github.com/Texera/te…

1f61066

…xera into shengquan-rpc-refactor

Update WorkflowExecutionService.scala

869791a

Update AmberClient.scala

3614b5a

update

320a741

update

fe21178

update

5d013d6

Merge branch 'master' into shengquan-rpc-refactor

9922c15

shengquan-ni force-pushed the shengquan-rpc-refactor branch 2 times, most recently from 8d5f5eb to 9922c15 Compare October 20, 2024 23:32

Yicong-Huang and others added 3 commits October 22, 2024 14:14

Merge branch 'master' into shengquan-rpc-refactor

3719ea6

update

99694d1

update

72ea461

shengquan-ni mentioned this pull request Oct 25, 2024

Remove unused RPCs and disable reconfiguration #2970

Merged

shengquan-ni added 2 commits October 28, 2024 23:29

Update backpressure_handler.py

df8e3fb

Update requirements.txt

98915d7

shengquan-ni requested a review from Yicong-Huang October 29, 2024 06:37

shengquan-ni added engine refactor Refactor the code labels Oct 29, 2024

Merge branch 'master' into shengquan-rpc-refactor

2803d36

Yicong-Huang approved these changes Oct 30, 2024

View reviewed changes

Yicong-Huang reviewed Oct 30, 2024

View reviewed changes

...src/main/scala/edu/uci/ics/amber/engine/architecture/messaginglayer/WorkerTimerService.scala Outdated Show resolved Hide resolved

Yicong-Huang reviewed Oct 30, 2024

View reviewed changes

core/amber/src/main/python/core/architecture/handlers/control/control_handler_base.py Show resolved Hide resolved

...i/ics/amber/engine/architecture/controller/promisehandlers/TakeGlobalCheckpointHandler.scala Show resolved Hide resolved

shengquan-ni added 3 commits October 30, 2024 16:29

Merge branch 'master' into shengquan-rpc-refactor

41979a6

address comments

bd5e9a6

Merge branch 'shengquan-rpc-refactor' of https://github.com/Texera/te…

040bb21

…xera into shengquan-rpc-refactor

Yicong-Huang reviewed Oct 31, 2024

View reviewed changes

Yicong-Huang and others added 8 commits October 30, 2024 23:23

mvp for decorators

e1ec1f4

remove unsupported use cases

b3e609c

clean up

027a635

fix format

f5676da

update

26dbfc2

Merge branch 'master' into shengquan-rpc-refactor

d30ace9

add noop back

7afeaaf

Merge branch 'shengquan-rpc-refactor' of https://github.com/Texera/te…

7956a3d

…xera into shengquan-rpc-refactor

shengquan-ni merged commit 84eba2b into master Oct 31, 2024

shengquan-ni deleted the shengquan-rpc-refactor branch October 31, 2024 22:18

shengquan-ni mentioned this pull request Nov 1, 2024

Fix nullpointer exception on result panel #2986

Merged

shengquan-ni added a commit that referenced this pull request Nov 1, 2024

Fix nullpointer exception on result panel (#2986)

ed3190a

This bug is caused by a wrong import of the `RUNNING` state introduced in #2950. Co-authored-by: Xinyuan Lin <xinyual3@uci.edu>

PurelyBlank pushed a commit that referenced this pull request Dec 4, 2024

Fix nullpointer exception on result panel (#2986)

896a16b

This bug is caused by a wrong import of the `RUNNING` state introduced in #2950. Co-authored-by: Xinyuan Lin <xinyual3@uci.edu>

shengquan-ni mentioned this pull request Jun 5, 2025

Fix operators showing "Running" state after workflow completion #3463

Merged

Migrate scala control messages to protobuf #2950

Migrate scala control messages to protobuf #2950

Uh oh!

Conversation

shengquan-ni commented Oct 17, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Main Changes (For both Scala and Python Engine)

Steps to Create New RPCs

Changes in Sending RPCs (Scala Engine)

Uh oh!

Yicong-Huang left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Yicong-Huang left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

shengquan-ni commented Oct 17, 2024 •

edited

Loading