nvflare.private.fed.server.server_runner module

class ServerRunner(config: ServerRunnerConfig, job_id: str, engine: ServerEngineSpec)

Bases: TBI

Server runner class.

Parameters:

• config (ServerRunnerConfig) – configuration of server runner

• job_id (str) – The number to distinguish each experiment

• engine (ServerEngineSpec) – server engine

ABORT_RETURN_CODES = ['RUN_MISMATCH', 'TASK_UNKNOWN', 'UNSAFE_JOB']

abort(fl_ctx: FLContext, turn_to_cold: bool = False)

get_persist_state(fl_ctx: FLContext) → dict

Generate data from state to be persisted.

Parameters:

fl_ctx – FLContext

Returns:

A dict serializable persist data

handle_dead_job(client_name: str, fl_ctx: FLContext)

handle_event(event_type: str, fl_ctx: FLContext)

Handles events.

Parameters:

• event_type (str) – event type fired by workflow.

• fl_ctx (FLContext) – FLContext information.

process_submission(client: Client, task_name: str, task_id: str, result: Shareable, fl_ctx: FLContext)

Process task result submitted from a client.

NOTE: the Engine will create a new fl_ctx and call this method:

with engine.new_context() as fl_ctx:

name, id, data = runner.process_submission(client, fl_ctx)

Parameters:

• client – Client object

• task_name – task name

• task_id – task id

• result – task result

• fl_ctx – FLContext

process_task_request(client: ~nvflare.apis.client.Client, fl_ctx: ~nvflare.apis.fl_context.FLContext) -> (<class 'str'>, <class 'str'>, <class 'nvflare.apis.shareable.Shareable'>)

Process task request from a client.

NOTE: the Engine will create a new fl_ctx and call this method:

with engine.new_context() as fl_ctx:

name, id, data = runner.process_task_request(client, fl_ctx) …

Parameters:

• client (Client) – client object

• fl_ctx (FLContext) – FL context

Returns:

A tuple of (task name, task id, and task data)

restore(state_data: dict, fl_ctx: FLContext)

Restore the state from persisted data.

Parameters:

• state_data – serialized persist data

• fl_ctx – FLContext

run()

class ServerRunnerConfig(heartbeat_timeout: int, task_request_interval: float, workflows: [], task_data_filters: dict, task_result_filters: dict, handlers=None, components=None)

Bases: object

Configuration for ServerRunner.

Parameters:

• heartbeat_timeout (int) – Client heartbeat timeout in seconds

• task_request_interval (float) – Task request interval in seconds

• workflows (list) – A list of workflow

• task_data_filters (dict) – A dict of {task_name: list of filters apply to data (pre-process)}

• task_result_filters (dict) – A dict of {task_name: list of filters apply to result (post-process)}

• handlers (list, optional) – A list of event handlers

• components (dict, optional) – A dict of extra python objects {id: object}

add_component(comp_id: str, component: object)