Add ClientAPIExecutor with frozen backend spec and working in_process backend#4857
Open
YuanTingHsieh wants to merge 12 commits into
Open
Add ClientAPIExecutor with frozen backend spec and working in_process backend#4857YuanTingHsieh wants to merge 12 commits into
YuanTingHsieh wants to merge 12 commits into
Conversation
EX-2 (Wave 0, interface freeze #2) of the Client API Execution Modes program (see docs/design/client_api_execution_modes_plan.md on PR NVIDIA#4853). New public ClientAPIExecutor at nvflare/app_common/executors/ client_api_executor.py with execution_mode dispatch (in_process / external_process / attach), plus a ClientAPIBackendSpec ABC and a frozen ClientAPIBackendContext the executor hands each backend (config + a back-reference for analytics). The three backends land in follow-up PRs (in_process, external_process, attach); until then each mode fails the job cleanly via system_panic rather than hanging. The constructor is the frozen configuration surface for the program: mode-scoped args are validated symmetrically (an arg set for a mode that ignores it is rejected with a clear error, not silently dropped). Beyond the design's V1 Configuration Surface list it also carries task_script_path/task_script_args (in_process trainer entry point), train/evaluate/submit_model_task_name + train_with_evaluation (power flare.is_train()/is_evaluate()/is_submit_model()), and memory_gc_rounds/ cuda_empty_cache, all forwarded from today's InProcessClientAPIExecutor/ ScriptRunner so existing jobs map without loss. The design doc's Configuration Surface and disposition table are updated to match. Analytics: the executor owns LOG-to-analytics conversion; the local path fires the un-prefixed event + ConvertToFedEvent (today's in-process behavior), the fed path fires "fed.analytix_log_stats" (today's MetricRelay behavior). UnsafeJobError propagates out of execute() so ClientRunner's UNSAFE_JOB handling still fires. No existing class is modified.
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces the new public ClientAPIExecutor entry point and its internal backend contract as an interface-freeze skeleton for the “Client API Execution Modes” program. It establishes a frozen constructor/configuration surface, dispatch scaffolding for in_process / external_process / attach, and a backend context/spec that follow-up PRs will implement.
Changes:
- Add
ClientAPIExecutorskeleton with execution-mode validation, backend creation hooks, clean failure behavior (system_panicon START_RUN), and executor-owned analytics event firing. - Add internal backend contract types:
ClientAPIBackendSpecABC and frozenClientAPIBackendContextsnapshot handed to backends. - Add a comprehensive unit test suite that pins constructor signature/order/defaults and validates mode-scoped argument handling and failure semantics.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
nvflare/app_common/executors/client_api_executor.py |
Adds the public executor skeleton with frozen configuration surface, backend dispatch hooks, and analytics ownership. |
nvflare/app_common/executors/client_api/backend_spec.py |
Introduces the internal backend interface (ClientAPIBackendSpec) and frozen backend context snapshot (ClientAPIBackendContext). |
nvflare/app_common/executors/client_api/__init__.py |
Marks the new client_api package. |
tests/unit_test/app_common/executors/client_api_executor_test.py |
Adds unit tests for validation matrix, dispatch failure behavior, analytics event firing paths, and surface-freeze signature pinning. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Contributor
Greptile SummaryThis PR introduces the frozen
Confidence Score: 5/5Safe to merge; no functional regressions were found in the new executor surface or the DataBus teardown path. The executor skeleton, backend spec, in_process backend, and InProcessClientAPI.close() all correctly address the cross-job DataBus leak, bounded join, and self-unwinding init that the design required. The 300+ tests drive real DataBus round-trips including timeout, abort, multi-round, and zombie-trainer containment. The three findings are minor defensiveness improvements that don't affect correctness. No files require special attention. Important Files Changed
Sequence Diagram%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
participant E as ClientAPIExecutor
participant B as InProcessBackend
participant DB as DataBus (singleton)
participant API as InProcessClientAPI
participant T as Trainer Thread
E->>B: initialize(context, fl_ctx)
B->>DB: subscribe(LOCAL_RESULT, LOG_DATA, ABORT, STOP)
B->>API: InProcessClientAPI(meta)
API->>DB: subscribe(GLOBAL_RESULT, ABORT, STOP)
B->>DB: put_data(CLIENT_API_KEY, api)
B->>T: Thread(task_fn_wrapper.run).start()
T->>DB: get_data(CLIENT_API_KEY) [flare.init()]
E->>B: execute(task_name, shareable, fl_ctx, signal)
B->>DB: fire_event(TOPIC_GLOBAL_RESULT, shareable)
DB->>API: "__receive_callback -> fl_model"
T->>API: "receive() -> fl_model [flare.receive()]"
T->>API: send(result) [flare.send()]
API->>DB: fire_event(TOPIC_LOCAL_RESULT, shareable)
DB->>B: "_local_result_callback -> _local_result"
B->>E: return result Shareable
E->>B: finalize(fl_ctx)
B->>DB: fire_event(TOPIC_STOP)
B->>T: "thread.join(timeout=30s)"
B->>B: _unwind()
B->>API: close() [unsubscribes API callbacks]
B->>DB: unsubscribe(LOCAL_RESULT, LOG_DATA, ABORT, STOP)
B->>DB: put_data(CLIENT_API_KEY, None)
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
participant E as ClientAPIExecutor
participant B as InProcessBackend
participant DB as DataBus (singleton)
participant API as InProcessClientAPI
participant T as Trainer Thread
E->>B: initialize(context, fl_ctx)
B->>DB: subscribe(LOCAL_RESULT, LOG_DATA, ABORT, STOP)
B->>API: InProcessClientAPI(meta)
API->>DB: subscribe(GLOBAL_RESULT, ABORT, STOP)
B->>DB: put_data(CLIENT_API_KEY, api)
B->>T: Thread(task_fn_wrapper.run).start()
T->>DB: get_data(CLIENT_API_KEY) [flare.init()]
E->>B: execute(task_name, shareable, fl_ctx, signal)
B->>DB: fire_event(TOPIC_GLOBAL_RESULT, shareable)
DB->>API: "__receive_callback -> fl_model"
T->>API: "receive() -> fl_model [flare.receive()]"
T->>API: send(result) [flare.send()]
API->>DB: fire_event(TOPIC_LOCAL_RESULT, shareable)
DB->>B: "_local_result_callback -> _local_result"
B->>E: return result Shareable
E->>B: finalize(fl_ctx)
B->>DB: fire_event(TOPIC_STOP)
B->>T: "thread.join(timeout=30s)"
B->>B: _unwind()
B->>API: close() [unsubscribes API callbacks]
B->>DB: unsubscribe(LOCAL_RESULT, LOG_DATA, ABORT, STOP)
B->>DB: put_data(CLIENT_API_KEY, None)
Reviews (12): Last reviewed commit: "Make backend unwind per-step best-effort..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #4857 +/- ##
==========================================
+ Coverage 56.20% 60.55% +4.34%
==========================================
Files 967 976 +9
Lines 91978 92674 +696
==========================================
+ Hits 51699 56121 +4422
+ Misses 40279 36553 -3726
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
FLARE-2698 bullet 2 removes params converters in favor of filters. Since EX-2 is an interface freeze, freezing args we intend to delete would make their later removal a breaking change — so exclude them now. Removed from the ClientAPIExecutor constructor: params_exchange_format, params_transfer_type, server_expected_format, from_nvflare_converter_id, to_nvflare_converter_id (and from ClientAPIBackendContext). Param conversion between the framework-agnostic aggregation representation (numpy) and the framework-native training representation moves to send/receive filters at the client edge (tracked as EX-5 / the converter->filter migration); the executor and Cell layers pass through. Transfer type FULL/DIFF is not a converter and stays a Client API (model_registry) concern. Surface-freeze test updated to the trimmed arg list. 83 tests pass.
Implements the first real backend behind the frozen ClientAPIBackendSpec, porting InProcessClientAPIExecutor's DataBus machinery with the behavior-parity bar "nothing user-visible": trainer script on a thread in the CJ process, tasks over TOPIC_GLOBAL_RESULT, results over TOPIC_LOCAL_RESULT, API discovery via CLIENT_API_KEY. Design-mandated deviations from the legacy executor: - No ParamsConverters / exchange-format knobs (FLARE-2698): the boundary is pass-through (ExchangeFormat.RAW, TransferType.FULL; DIFF returns with the model_registry transfer-type decision). - LOG data routes through the executor-owned fire_log_analytics() (single analytics-event ownership point). - initialize() self-unwinds on failure; finalize() is idempotent and detaches ALL of this job's DataBus state -- including the InProcessClientAPI's own subscriptions via a new close() method (the singleton bus otherwise keeps a dead job's API subscribed, pinning each later job's global model). - execute() is bounded by result_wait_timeout (monotonic clock: a wall-clock step must not fire a spurious abort). After the in-process trainer aborts (script failure, timeout, STOP -- its thread is never relaunched), later tasks fail fast at entry with EXECUTION_EXCEPTION and the recorded first cause; an instant TASK_ABORTED would be misleading (task never delivered, abort_signal never triggered). Stale results from a timed-out task are dropped at the next task's entry. The executor's in_process factory now returns the real backend; the skeleton dispatch tests keep external_process/attach pinned as clean-failure modes. 18 backend tests drive the real DataBus round trip: multi-round execute, abort mid-task, post-timeout fail-fast, bounded wait, unwind on early/late init failure, idempotent finalize (exactly one STOP), successor CLIENT_API_KEY guard, dead-API subscription detach, LOG-to-analytics routing, RAW/FULL meta.
YuanTingHsieh
changed the title
…EX-4) ScriptRunner(script="client.py", execution_mode="in_process") wires the new ClientAPIExecutor instead of the legacy executor stack -- existing examples flip with one argument. Validated end-to-end: the hello-numpy example's unmodified client script ran a 3-round 2-client simulator FedAvg on the new executor with a mathematically correct aggregated model; the unit tests pin that add_to_fed_job produces exactly the executor/args that run validated. Contract: - Default None: legacy behavior byte-for-byte unchanged. - "in_process": new executor; the params boundary is pass-through (no ParamsConverters per FLARE-2698), so the framework format resolution -- including its torch/tensorflow import checks -- is skipped, and the legacy stack args (launch_external_process, executor, task_pipe, launcher, metric_relay, metric_pipe) are rejected as conflicts. - "external_process"/"attach": rejected at job BUILD time with a clear message until their backends land (no deferred START_RUN panic).
pcnudde
previously approved these changes
Jul 8, 2026
finalize() joined the trainer thread unbounded. Safe in the legacy executor (execute() never returns while the trainer lives), but result_wait_timeout makes a live-trainer finalize reachable: a trainer wedged in user code never observes TOPIC_STOP (it is only seen at the next flare call), and the join would hang CJ/simulator teardown forever. finalize() now joins with a 30s bound and abandons with a loud log; the trainer thread is daemon (deliberate divergence from legacy, documented) so an abandoned trainer cannot block process exit either. Regression test drives a genuinely stuck trainer.
YuanTingHsieh
force-pushed
the
yuantingh/client-api-ex2-executor-skeleton
branch
from
4604eed to
171b6c1
Compare
Module and constructor docs now describe the API as it is, not how it got here: dropped the interface-freeze/plan-id/follow-up-PR narration (some of it already stale -- in_process ships in this PR, not a follow-up), the internal ticket references, and the design-divergence changelog section. Kept the user-relevant substance: mode availability, and why the surface has no parameter-conversion args (pass-through boundary; conversion belongs to client-edge filters; FULL/DIFF is a model-registry concern).
finalize() can abandon a still-running daemon trainer after its bounded join; that thread still holds the API object and may resume later. close() detached only the incoming subscriptions -- the zombie's send()/log() could still publish onto the singleton DataBus, where a successor job's backend is now subscribed, feeding it a dead job's result or metric (results carry no job correlation). close() now also marks the API closed; send()/log() DROP with a warning and receive() returns None immediately. Dropping -- not raising -- is deliberate: an exception would propagate into TaskScriptRunner's catch-all, which fires TOPIC_ABORT onto the same singleton bus and would poison the successor, the exact cross-job leak being prevented. Tests cover the dropped send/log, the fast-exit receive, and that an open API still publishes (no over-drop).
_unwind() ran the backend unsubscribes and _client_api.close() in one try block: a failing unsubscribe skipped close(), so the closed gate was never set and an abandoned trainer could still publish results/metrics onto the singleton DataBus (and the dead API stayed subscribed, pinning later jobs' global models). Close the API first -- it is the one step that must not be skipped -- and give every teardown step (close, CLIENT_API_KEY clear, each unsubscribe) its own best-effort block. Tests cover both directions: a failing unsubscribe still closes the API and attempts every remaining unsubscribe, and a failing close() still clears CLIENT_API_KEY and the backend's own subscriptions.
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.
Interface freeze #2 (EX-2) plus the first working backend (EX-3): the frozen
ClientAPIExecutorconstructor +ClientAPIBackendSpec, and the in_process backend implementing it, with tests that drive the real DataBus round trip.Skeleton (EX-2): one public executor for all Client API execution modes (
in_process/external_process/attach), frozen V1 constructor surface (converter/exchange-format args deliberately excluded per FLARE-2698), per-mode constructor validation, executor-owned analytics hook (fire_log_analytics— the single LOG→analytics ownership point), clean-failure dispatch for not-yet-implemented modes.in_process backend (EX-3): ports
InProcessClientAPIExecutor's DataBus machinery behind the frozen spec — trainer script on a CJ thread, tasks overTOPIC_GLOBAL_RESULT, results overTOPIC_LOCAL_RESULT— with the contract obligations the legacy code lacked:initialize()self-unwinds on failure;finalize()is idempotent and detaches all of the job's DataBus state, including the API instance's own subscriptions via a newInProcessClientAPI.close()(the singleton bus otherwise keeps a dead job's API subscribed, pinning each later job's global model);execute()is bounded byresult_wait_timeouton a monotonic clock; after the in-process trainer aborts (its thread is never relaunched), later tasks fail fast at entry withEXECUTION_EXCEPTIONand the recorded first cause instead of a misleading instantTASK_ABORTED; stale results from a timed-out task are dropped;ExchangeFormat.RAW/TransferType.FULL; DIFF returns with the model_registry transfer-type decision);Tests: 101 for the executor + backend (constructor matrix, surface freeze, dispatch, real-DataBus round trip, multi-round, abort mid-task, post-timeout fail-fast, early/late init-failure unwind, idempotent finalize with exactly one STOP, successor
CLIENT_API_KEYguard, dead-API detach, LOG routing, RAW/FULL meta); 301 green across the executor + client in_process surface; legacy executors untouched.Design:
docs/design/client_api_execution_modes.md(#4853). Plan rows: EX-2 + EX-3.