[andygrove]

Which issue does this PR close?

Part of #2652. This is the first of a series of smaller PRs split out of #2657 (pipelined execution).

Rationale for this change

A standalone (in-process) cluster always runs pull-staged. Pipelined execution needs push-staged scheduling, and its end-to-end tests run on standalone clusters, so the standalone cluster needs a push-staged option. That option is also useful by itself, for testing push-staged scheduling without starting separate processes.

Running push-staged in standalone mode exposed a bug. The standalone scheduler advertised the hardcoded endpoint localhost:50050 instead of the port it actually bound. Pull-staged executors never use that endpoint, but push-staged executors report task status to the endpoint each launch names.

What changes are included in this PR?

• A new setting, ballista.standalone.task_scheduling_policy: pull-staged (the default, unchanged behavior) or push-staged.
• new_standalone_scheduler_with_builder binds its listener before it builds the scheduler, advertises localhost:<port>, and uses the configured policy.
• With push-staged, new_standalone_executor_from_builder takes a free port for its task-launch gRPC server and starts through executor_server::startup, instead of running the poll loop.
• A test helper, setup_test_cluster_with_executors, starts a scheduler and several executors in one process.

How are these changes tested?

• A config unit test covers the default, case-insensitive parsing and a rejected value.
• ballista/client/tests/standalone_push_staged.rs runs a scan, an aggregate and a join on a push-staged standalone cluster, and on a push-staged scheduler with two executors. It compares each result with a pull-staged standalone cluster.

Are there any user-facing changes?

There is one new setting, ballista.standalone.task_scheduling_policy, documented in the config reference. The default behavior is unchanged.

[mbutrovich]

First pass, thanks @andygrove!

[mbutrovich]

Could the executor keep the gRPC listener bound instead of releasing the port and binding it again in executor_server::startup? Between the drop on line 117 and the bind in startup, any other bind to port 0 in the process can get the same number. The two new tests run in parallel and each binds several port-0 listeners, so this can happen in this test binary. When it does, startup fails with AddrInUse and the cluster doesn't come up. The repo already ran into this with free_ports in chaos-testing: "Releasing each before taking the next lets the kernel hand the same number back" (cluster.rs). One option is a variant of startup that takes an already-bound TcpIncoming, with startup calling it. That variant would be a new public item in executor_server, so it might be better to keep it pub(crate) if the standalone executor is its only caller.

[andygrove]

Good catch, thanks! The standalone executor now keeps its listener bound and passes it to a new pub(crate) startup_with_incoming. startup binds as before and delegates to it, so nothing new is public.

[mbutrovich]

Could this comment describe what happens if the receiver is dropped? Dropping the notifier does stop the loops, because Shutdown::recv returns once the broadcast sender is gone. Dropping stop_receive doesn't stop them. Instead, the next StopExecutor request panics at stop_sender.send(force).await.unwrap() (executor_server.rs).

Suggested change

	// The executor's loops stop when the notifier or the stop channel is
	// dropped, so keep both for the life of the runtime.
	// The executor's loops stop when the notifier is dropped, and its
	// `stop_executor` handler panics if the stop channel's receiver is
	// dropped, so keep both for the life of the runtime.

[andygrove]

Thanks, applied your wording.

[mbutrovich]

Should these tests check that the cluster really runs push-staged? Their results match pull-staged by design, so they would also pass if the setting were ignored. I tried that locally on the head commit. I hard-coded PullStaged in new_standalone_scheduler_with_builder, forced push_staged to false in the executor, and both tests still passed. A push-staged scheduler rejects poll_work with FailedPrecondition (grpc.rs), so the two-executor test could assert that after the cluster starts:

use ballista_core::serde::protobuf::PollWorkParams;
use ballista_core::serde::protobuf::scheduler_grpc_client::SchedulerGrpcClient;

let mut scheduler = SchedulerGrpcClient::connect(format!("http://{host}:{port}"))
    .await
    .unwrap();
let status = scheduler
    .poll_work(PollWorkParams::default())
    .await
    .unwrap_err();
assert_eq!(status.code(), tonic::Code::FailedPrecondition);

[andygrove]

Nice find. The two-executor test now asserts poll_work gets FailedPrecondition. I checked it against a pull-staged cluster and it fails there with InvalidArgument, so it catches the setting being ignored.

[mbutrovich]

Thanks @andygrove, the updates look good to me!

[andygrove]

Thanks @mbutrovich