feat(framework) Add ClientApp multiprocess execution

src/py/flwr/client/app.py

@@ -15,6 +15,7 @@
 """Flower client app."""
 
 import signal
+import subprocess
 import sys
 import time
 from dataclasses import dataclass
@@ -35,6 +36,7 @@
 from flwr.common.address import parse_address
 from flwr.common.constant import (
     MISSING_EXTRA_REST,
+    RUN_ID_NUM_BYTES,
     TRANSPORT_TYPE_GRPC_ADAPTER,
     TRANSPORT_TYPE_GRPC_BIDI,
     TRANSPORT_TYPE_GRPC_RERE,
@@ -48,14 +50,15 @@
 from flwr.common.typing import Fab, Run, UserConfig
 from flwr.proto.clientappio_pb2_grpc import add_ClientAppIoServicer_to_server
 from flwr.server.superlink.fleet.grpc_bidi.grpc_server import generic_create_grpc_server
+from flwr.server.superlink.state.utils import generate_rand_int_from_bytes
 
 from .grpc_adapter_client.connection import grpc_adapter
 from .grpc_client.connection import grpc_connection
 from .grpc_rere_client.connection import grpc_request_response
 from .message_handler.message_handler import handle_control_message
 from .node_state import NodeState
 from .numpy_client import NumPyClient
-from .process.clientappio_servicer import ClientAppIoServicer
+from .process.clientappio_servicer import ClientAppIoInputs, ClientAppIoServicer
 
 ADDRESS_CLIENTAPPIO_API_GRPC_RERE = "0.0.0.0:9094"
 
@@ -204,6 +207,8 @@ def start_client_internal(
     max_retries: Optional[int] = None,
     max_wait_time: Optional[float] = None,
     flwr_path: Optional[Path] = None,
+    isolate: Optional[bool] = False,
+    supernode_address: Optional[str] = ADDRESS_CLIENTAPPIO_API_GRPC_RERE,
 ) -> None:
     """Start a Flower client node which connects to a Flower server.
 
@@ -251,6 +256,13 @@ class `flwr.client.Client` (default: None)
         If set to None, there is no limit to the total time.
     flwr_path: Optional[Path] (default: None)
         The fully resolved path containing installed Flower Apps.
+    isolate : Optional[bool] (default: False)
+        Whether to run `ClientApp` in a separate process. By default, this value is
+        `False`, and the `ClientApp` runs in the same process as the SuperNode. If
+        `True`, the `ClientApp` runs in an isolated process and communicates using
+        gRPC at the address `supernode_address`.
+    supernode_address : Optional[str] (default: `ADDRESS_CLIENTAPPIO_API_GRPC_RERE`)
+        The SuperNode gRPC server address.
     """
     if insecure is None:
         insecure = root_certificates is None
@@ -276,6 +288,11 @@ def _load_client_app(_1: str, _2: str) -> ClientApp:
 
         load_client_app_fn = _load_client_app
 
+    if isolate and supernode_address is not None:
+        _clientappio_grpc_server, clientappio_servicer = run_clientappio_api_grpc(
+            address=supernode_address
+        )
+
     # At this point, only `load_client_app_fn` should be used
     # Both `client` and `client_fn` must not be used directly
 
@@ -435,11 +452,43 @@ def _on_backoff(retry_state: RetryState) -> None:
 
                     # Handle app loading and task message
                     try:
-                        # Load ClientApp instance
-                        client_app: ClientApp = load_client_app_fn(fab_id, fab_version)
+                        if isolate and supernode_address is not None:
+                            # Generate SuperNode token
+                            token: int = generate_rand_int_from_bytes(RUN_ID_NUM_BYTES)
+
+                            # Share Message and Context with servicer
+                            clientappio_servicer.set_inputs(
+                                ClientAppIoInputs(
+                                    message=message,
+                                    context=context,
+                                    run=run,
+                                    token=token,
+                                )
+                            )
+                            # Run ClientApp
+                            command = [
+                                "flwr-clientapp",
+                                "--supernode",
+                                supernode_address,
+                                "--token",
+                                str(token),
+                            ]
+                            subprocess.run(
+                                command,
+                                stdout=None,
+                                stderr=None,
+                                check=True,
+                            )
+                            outputs = clientappio_servicer.get_outputs()
+                            reply_message, context = outputs.message, outputs.context
+                        else:
+                            # Load ClientApp instance
+                            client_app: ClientApp = load_client_app_fn(
+                                fab_id, fab_version
+                            )
 
-                        # Execute ClientApp
-                        reply_message = client_app(message=message, context=context)
+                            # Execute ClientApp
+                            reply_message = client_app(message=message, context=context)
                     except Exception as ex:  # pylint: disable=broad-exception-caught
 
                         # Legacy grpc-bidi
@@ -685,9 +734,7 @@ def signal_handler(sig, frame):  # type: ignore
         signal.signal(signal.SIGTERM, signal_handler)
 
 
-def run_clientappio_api_grpc(
-    address: str = ADDRESS_CLIENTAPPIO_API_GRPC_RERE,
-) -> Tuple[grpc.Server, ClientAppIoServicer]:
+def run_clientappio_api_grpc(address: str) -> Tuple[grpc.Server, ClientAppIoServicer]:
     """Run ClientAppIo API gRPC server."""
     clientappio_servicer: grpc.Server = ClientAppIoServicer()
     clientappio_add_servicer_to_server_fn = add_ClientAppIoServicer_to_server

src/py/flwr/client/process/clientappio_servicer.py

@@ -94,10 +94,6 @@ def PushClientAppOutputs(
     ) -> PushClientAppOutputsResponse:
         """Push Message and Context."""
         log(DEBUG, "ClientAppIo.PushClientAppOutputs")
-        if self.clientapp_output is None:
-            raise ValueError(
-                "ClientAppIoOutputs not set before calling `PushClientAppOutputs`."
-            )
         if self.clientapp_input is None:
             raise ValueError(
                 "ClientAppIoInputs not set before calling `PushClientAppOutputs`."
@@ -109,8 +105,11 @@ def PushClientAppOutputs(
             )
         try:
             # Update Message and Context
-            self.clientapp_output.message = message_from_proto(request.message)
-            self.clientapp_output.context = context_from_proto(request.context)
+            self.clientapp_output = ClientAppIoOutputs(
+                message=message_from_proto(request.message),
+                context=context_from_proto(request.context),
+            )
+
             # Set status
             code = typing.ClientAppOutputCode.SUCCESS
             status = typing.ClientAppOutputStatus(code=code, message="Success")

src/py/flwr/client/process/process.py

@@ -52,20 +52,20 @@ def on_channel_state_change(channel_connectivity: str) -> None:
 
 
 def run_clientapp(  # pylint: disable=R0914
-    address: str,
+    supernode: str,
     token: int,
 ) -> None:
     """Run Flower ClientApp process.
 
     Parameters
     ----------
-    address : str
+    supernode : str
         Address of SuperNode
     token : int
         Unique SuperNode token for ClientApp-SuperNode authentication
     """
     channel = create_channel(
-        server_address=address,
+        server_address=supernode,
         insecure=True,
     )
     channel.subscribe(on_channel_state_change)

src/py/flwr/client/supernode/app.py

@@ -73,6 +73,8 @@ def run_supernode() -> None:
         max_retries=args.max_retries,
         max_wait_time=args.max_wait_time,
         node_config=parse_config_args([args.node_config]),
+        isolate=args.isolate,
+        supernode_address=args.supernode_address,
     )
 
     # Graceful shutdown
@@ -121,7 +123,7 @@ def flwr_clientapp() -> None:
         description="Run a Flower ClientApp",
     )
     parser.add_argument(
-        "--address",
+        "--supernode",
         help="Address of SuperNode ClientAppIo gRPC servicer",
     )
     parser.add_argument(
@@ -133,10 +135,10 @@ def flwr_clientapp() -> None:
         DEBUG,
         "Staring isolated `ClientApp` connected to SuperNode ClientAppIo at %s "
         "with the token %s",
-        args.address,
+        args.supernode,
         args.token,
     )
-    run_clientapp(address=args.address, token=int(args.token))
+    run_clientapp(supernode=args.supernode, token=int(args.token))
 
 
 def _warn_deprecated_server_arg(args: argparse.Namespace) -> None:
@@ -223,6 +225,17 @@ def _parse_args_run_supernode() -> argparse.ArgumentParser:
         - `$HOME/.flwr/` in all other cases
     """,
     )
+    parser.add_argument(
+        "--isolate",
+        action="store_true",
+        help="Run `ClientApp` in an isolated subprocess. By default, `ClientApp` "
+        "runs in the same process that executes the SuperNode.",
+    )
+    parser.add_argument(
+        "--supernode-address",
+        default="0.0.0.0:9094",
+        help="Set the SuperNode gRPC server address. Defaults to `0.0.0.0:9094`.",
+    )
 
     return parser