[Q&A] Maintaining executor state across workflows

[ssingh-dual]

Python version (python3 -V)

3.12.3

NVFlare version (python3 -m pip list | grep "nvflare")

2.5.2

NVFlare branch (if running examples, please use the branch that corresponds to the NVFlare version, git branch)

2.5 main

Operating system

Ubuntu 24.04.1 LTS

Have you successfully run any of the following examples?

• hello-numpy-sag with simulator
• hello-pt with simulator
• hello-numpy-sag with POC
• hello-pt with POC

Please describe your question

I'm building a job that consists of 2 workflows, each performing the scatter_and_gather workflow (say, with num_executor=1 and num_aggregator=1). For each workflow, the executor class creates a new instance of executor.
Is there a way to use the same instance of executor across the two workflows? Or to share few member variable values across the two instances. Below describes what I'm trying to achieve:

class Executors(Executor):
    def __init__(self):
        super().__init__()
        self.key = None
    def execute():
        self.key = 1

During execution of workflow#1, the key of the executor instance is set to 1. Is there a way where the executor instance in workflow#2 retains the key value=1?

I might be able to use FLContext to hold key value as a property but that makes the key value available to all executors. If I have a set of executors, I'd associate each instance of executor with a private key and the following workflows should withhold the keys for the same set of executors.

[YuanTingHsieh]

@ssingh-dual thanks for the question.

There are two key concepts to understand here:

1. Server Side: The server side has Controllers that manage the workflow. In the case of SAG (Scatter and Gather), you can find an example of this here.

2. Client Side: The client side has Executors that receive corresponding tasks, execute them, and return the results.

For your question:
You want to use the same Executor on the client side across different SAG workflows, which is entirely feasible.

To achieve this, you simply need to configure the config_fed_server to use two or more SAG controllers. Here's an example configuration:

% cat qa_job/app/config/config_fed_server.conf 
{
  format_version = 2
  task_data_filters = []
  task_result_filters = []
  model_class_path = "net.Net"
  workflows = [
    {
      id = "scatter_and_gather1"
      path = "nvflare.app_common.workflows.scatter_and_gather.ScatterAndGather"
      args {
        min_clients = 2
        num_rounds = 2
        start_round = 0
        wait_time_after_min_received = 0
        aggregator_id = "aggregator"
        persistor_id = "persistor"
        shareable_generator_id = "shareable_generator"
        train_task_name = "train1"
        train_timeout = 0
      }
    }
    {
      id = "scatter_and_gather2"
      path = "nvflare.app_common.workflows.scatter_and_gather.ScatterAndGather"
      args {
        min_clients = 2
        num_rounds = 2
        start_round = 0
        wait_time_after_min_received = 0
        aggregator_id = "aggregator"
        persistor_id = "persistor"
        shareable_generator_id = "shareable_generator"
        train_task_name = "train2"
        train_timeout = 0
      }
    }
  ]
}

In this configuration, you define two SAG workflows with different train_task_name values (train1 and train2).

On the client side, you can use the same Executor for both tasks by specifying them in the tasks list. Here’s the corresponding client configuration:

executors = [
  {
    tasks = [
      "train1", "train2"
    ]
    executor {
      path = "nvflare.app_opt.pt.client_api_launcher_executor.PTClientAPILauncherExecutor"
      args {
        launcher_id = "launcher"
        pipe_id = "pipe"
        heartbeat_timeout = 60
        params_exchange_format = "numpy"
        params_transfer_type = "DIFF"
        train_with_evaluation = true
      }
    }
  }
]

Notice that the task names (train1 and train2) in the client configuration correspond to those defined in the server's workflow configuration.

Note: Since these workflows are part of the same job, the Executor instance will be the same across both tasks, allowing the same Executor to handle multiple tasks from different workflows.