feat(framework) Add ClientApp process function

src/py/flwr/client/app.py

@@ -22,6 +22,7 @@
 from pathlib import Path
 from typing import Callable, ContextManager, Dict, Optional, Tuple, Type, Union
 
+import grpc
 from cryptography.hazmat.primitives.asymmetric import ec
 from grpc import RpcError
 
@@ -43,13 +44,18 @@
 from flwr.common.message import Error
 from flwr.common.retry_invoker import RetryInvoker, RetryState, exponential
 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 .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
+
+ADDRESS_CLIENTAPPIO_API_GRPC_RERE = "0.0.0.0:9094"
 
 
 def _check_actionable_client(
@@ -667,3 +673,22 @@ def signal_handler(sig, frame):  # type: ignore
 
         signal.signal(signal.SIGINT, signal_handler)
         signal.signal(signal.SIGTERM, signal_handler)
+
+
+def run_clientappio_api_grpc(
+    address: str = ADDRESS_CLIENTAPPIO_API_GRPC_RERE,
+) -> Tuple[grpc.Server, grpc.Server]:
+    """Run ClientAppIo API gRPC server."""
+    clientappio_servicer: grpc.Server = ClientAppIoServicer()
+    clientappio_add_servicer_to_server_fn = add_ClientAppIoServicer_to_server
+    clientappio_grpc_server = generic_create_grpc_server(
+        servicer_and_add_fn=(
+            clientappio_servicer,
+            clientappio_add_servicer_to_server_fn,
+        ),
+        server_address=address,
+        max_message_length=GRPC_MAX_MESSAGE_LENGTH,
+    )
+    log(INFO, "Starting Flower ClientAppIo gRPC server on %s", address)
+    clientappio_grpc_server.start()
+    return clientappio_servicer, clientappio_grpc_server

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

@@ -0,0 +1,122 @@
+# Copyright 2024 Flower Labs GmbH. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==============================================================================
+"""Flower background ClientApp."""
+
+from logging import DEBUG, ERROR, INFO
+from typing import Tuple
+
+import grpc
+
+from flwr.client.client_app import ClientApp
+from flwr.common import Context, Message
+from flwr.common.grpc import create_channel
+from flwr.common.logger import log
+from flwr.common.serde import (
+    context_from_proto,
+    context_to_proto,
+    message_from_proto,
+    message_to_proto,
+    run_from_proto,
+)
+from flwr.common.typing import Run
+
+# pylint: disable=E0611
+from flwr.proto.clientappio_pb2 import (
+    PullClientAppInputsRequest,
+    PullClientAppInputsResponse,
+    PushClientAppOutputsRequest,
+    PushClientAppOutputsResponse,
+)
+from flwr.proto.clientappio_pb2_grpc import ClientAppIoStub
+
+from .utils import _get_load_client_app_fn
+
+
+def _run_background_client(  # pylint: disable=R0914
+    address: str,
+    token: int,
+) -> None:
+    """Run background Flower ClientApp process.
+
+    Parameters
+    ----------
+    address : str
+        Address of SuperNode
+    token : int
+        Unique SuperNode token for ClientApp-SuperNode authentication
+    """
+
+    def on_channel_state_change(channel_connectivity: str) -> None:
+        """Log channel connectivity."""
+        log(DEBUG, channel_connectivity)
+
+    channel = create_channel(
+        server_address=address,
+        insecure=True,
+    )
+    channel.subscribe(on_channel_state_change)
+
+    try:
+        stub = ClientAppIoStub(channel)
+
+        # Pull Message, Context, and Run from SuperNode
+        message, context, run = pull_message(stub=stub, token=token)
+
+        load_client_app_fn = _get_load_client_app_fn(
+            default_app_ref="",
+            app_path=None,
+            multi_app=True,
+            flwr_dir=None,
+        )
+
+        # Load ClientApp
+        client_app: ClientApp = load_client_app_fn(run.fab_id, run.fab_version)
+
+        # Execute ClientApp
+        reply_message = client_app(message=message, context=context)
+
+        # Push Message and Context to SuperNode
+        _ = push_message(token=token, message=reply_message, context=context, stub=stub)
+    except KeyboardInterrupt:
+        log(INFO, "Closing connection")
+    except grpc.RpcError as e:
+        log(ERROR, "GRPC error occurred: %s", str(e))
+    finally:
+        channel.close()
+
+
+def pull_message(stub: grpc.Channel, token: int) -> Tuple[Message, Context, Run]:
+    """Pull message from SuperNode to ClientApp."""
+    res: PullClientAppInputsResponse = stub.PullClientAppInputs(
+        PullClientAppInputsRequest(token=token)
+    )
+    message = message_from_proto(res.message)
+    context = context_from_proto(res.context)
+    run = run_from_proto(res.run)
+    return message, context, run
+
+
+def push_message(
+    stub: grpc.Channel, token: int, message: Message, context: Context
+) -> PushClientAppOutputsResponse:
+    """Push message to SuperNode from ClientApp."""
+    proto_message = message_to_proto(message)
+    proto_context = context_to_proto(context)
+    res: PushClientAppOutputsResponse = stub.PushClientAppOutputs(
+        PushClientAppOutputsRequest(
+            token=token, message=proto_message, context=proto_context
+        )
+    )
+    return res

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

@@ -0,0 +1,108 @@
+# Copyright 2024 Flower Labs GmbH. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==============================================================================
+"""Flower ClientApp loading utils."""
+
+from logging import DEBUG
+from pathlib import Path
+from typing import Callable, Optional
+
+from flwr.client.client_app import ClientApp, LoadClientAppError
+from flwr.common.config import (
+    get_flwr_dir,
+    get_metadata_from_config,
+    get_project_config,
+    get_project_dir,
+)
+from flwr.common.logger import log
+from flwr.common.object_ref import load_app, validate
+
+
+def _get_load_client_app_fn(
+    default_app_ref: str,
+    app_path: Optional[str],
+    multi_app: bool,
+    flwr_dir: Optional[str] = None,
+) -> Callable[[str, str], ClientApp]:
+    """Get the load_client_app_fn function.
+
+    If `multi_app` is True, this function loads the specified ClientApp
+    based on `fab_id` and `fab_version`. If `fab_id` is empty, a default
+    ClientApp will be loaded.
+
+    If `multi_app` is False, it ignores `fab_id` and `fab_version` and
+    loads a default ClientApp.
+    """
+    if not multi_app:
+        log(
+            DEBUG,
+            "Flower SuperNode will load and validate ClientApp `%s`",
+            default_app_ref,
+        )
+
+        valid, error_msg = validate(default_app_ref, project_dir=app_path)
+        if not valid and error_msg:
+            raise LoadClientAppError(error_msg) from None
+
+    def _load(fab_id: str, fab_version: str) -> ClientApp:
+        runtime_app_dir = Path(app_path if app_path else "").absolute()
+        # If multi-app feature is disabled
+        if not multi_app:
+            # Set app reference
+            client_app_ref = default_app_ref
+        # If multi-app feature is enabled but app directory is provided
+        elif app_path is not None:
+            config = get_project_config(runtime_app_dir)
+            this_fab_version, this_fab_id = get_metadata_from_config(config)
+
+            if this_fab_version != fab_version or this_fab_id != fab_id:
+                raise LoadClientAppError(
+                    f"FAB ID or version mismatch: Expected FAB ID '{this_fab_id}' and "
+                    f"FAB version '{this_fab_version}', but received FAB ID '{fab_id}' "
+                    f"and FAB version '{fab_version}'.",
+                ) from None
+
+            # log(WARN, "FAB ID is not provided; the default ClientApp will be loaded.")
+
+            # Set app reference
+            client_app_ref = config["tool"]["flwr"]["app"]["components"]["clientapp"]
+        # If multi-app feature is enabled
+        else:
+            try:
+                runtime_app_dir = get_project_dir(
+                    fab_id, fab_version, get_flwr_dir(flwr_dir)
+                )
+                config = get_project_config(runtime_app_dir)
+            except Exception as e:
+                raise LoadClientAppError("Failed to load ClientApp") from e
+
+            # Set app reference
+            client_app_ref = config["tool"]["flwr"]["app"]["components"]["clientapp"]
+
+        # Load ClientApp
+        log(
+            DEBUG,
+            "Loading ClientApp `%s`",
+            client_app_ref,
+        )
+        client_app = load_app(client_app_ref, LoadClientAppError, runtime_app_dir)
+
+        if not isinstance(client_app, ClientApp):
+            raise LoadClientAppError(
+                f"Attribute {client_app_ref} is not of type {ClientApp}",
+            ) from None
+
+        return client_app
+
+    return _load