[Task Runner API] [Flower Interoperability] Enable option to run ClientApp as a separate process in SGX (securefederatedai#1580)

* add clientapp isolation

Signed-off-by: kta-intel <kevin.ta@intel.com>

* enforce the sequence for spawning and terminating clientapp relative to supernode

Signed-off-by: kta-intel <kevin.ta@intel.com>

* move is_port_open() check to inside conditional, add a small delay after port is confirmed to be open

Signed-off-by: kta-intel <kevin.ta@intel.com>

* refactor patch > SGX_enabled

Signed-off-by: kta-intel <kevin.ta@intel.com>

* refactor local_grpc_server > interop_server

Signed-off-by: kta-intel <kevin.ta@intel.com>

* format

Signed-off-by: kta-intel <kevin.ta@intel.com>

* change SGX_enabled to lowercase sgx_enabled for visual consistency

Signed-off-by: kta-intel <kevin.ta@intel.com>

---------

Signed-off-by: kta-intel <kevin.ta@intel.com>
Co-authored-by: teoparvanov <teodor.parvanov@intel.com>

Author: kminhta

openfl-workspace/flower-app-pytorch/plan/plan.yaml

@@ -22,7 +22,7 @@ connector :
     flwr_run_params :
       flwr_app_name : "app-pytorch"
       federation_name : "local-poc"
-      patch: True
+      sgx_enabled: False
 
 collaborator :
   defaults : plan/defaults/collaborator.yaml
@@ -39,7 +39,7 @@ task_runner :
   template : src.runner.FlowerTaskRunner
   settings :
     flwr_app_name: app-pytorch
-    patch: True
+    sgx_enabled: False
 
 network :
   defaults : plan/defaults/network.yaml

openfl-workspace/flower-app-pytorch/src/connector_flower.py

@@ -156,7 +156,7 @@ def _build_flwr_run_command(self) -> list[str]:
 
         os.environ["TMPDIR"] = os.environ["FLWR_HOME"]
 
-        if self.flwr_run_params.get("patch"):
+        if self.flwr_run_params.get("sgx_enabled"):
             command = ["python", "src/patch/flwr_run_patch.py", "run", f"./src/{flwr_app_name}"]
         else:
             command = ["flwr", "run", f"./src/{flwr_app_name}"]

openfl-workspace/flower-app-pytorch/src/runner.py

@@ -34,10 +34,10 @@ def __init__(self, **kwargs):
         """
         super().__init__(**kwargs)
 
-        self.patch = kwargs.get('patch')
+        self.sgx_enabled = kwargs.get('sgx_enabled')
         if self.data_loader is None:
             flwr_app_name = kwargs.get('flwr_app_name')
-            if self.patch:
+            if self.sgx_enabled:
                 install_flower_FAB(flwr_app_name)
             return
 
@@ -52,12 +52,12 @@ def __init__(self, **kwargs):
 
         self.shutdown_requested = False  # Flag to signal shutdown
 
-    def start_client_adapter(self, local_grpc_server, **kwargs):
+    def start_client_adapter(self, interop_server, **kwargs):
         """
         Start the local gRPC server and the Flower SuperNode.
 
         Args:
-            local_grpc_server: The local gRPC server instance.
+            interop_server: The local gRPC server instance.
             **kwargs: Additional parameters, including 'local_server_port'.
         """
         local_server_port = kwargs.get('local_server_port')
@@ -66,10 +66,10 @@ def message_callback():
             self.shutdown_requested = True
 
         # Set the callback for ending the experiment
-        local_grpc_server.set_end_experiment_callback(message_callback)
-        local_grpc_server.start_server(local_server_port)
+        interop_server.set_end_experiment_callback(message_callback)
+        interop_server.start_server(local_server_port)
 
-        local_server_port = local_grpc_server.get_port()
+        local_server_port = interop_server.get_port()
 
         command = [
             "flower-supernode",
@@ -80,20 +80,43 @@ def message_callback():
             "--node-config", f"data-path='{self.data_path}'"
         ]
 
+        if self.sgx_enabled:
+            command += ["--isolation", "process"]
+            flwr_clientapp_command = [
+                "flwr-clientapp",
+                "--insecure",
+                "--clientappio-api-address", f"127.0.0.1:{self.client_port}",
+            ]
+
+        self.logger.info("Starting Flower SuperNode process...")
         supernode_process = subprocess.Popen(command, shell=False)
-        local_grpc_server.handle_signals(supernode_process)
+        interop_server.handle_signals(supernode_process)
+
+        if self.sgx_enabled:
+            # Check if port is open before starting the client app
+            while not is_port_open('127.0.0.1', local_server_port):
+                time.sleep(0.5)
+
+            time.sleep(1) # Add a small delay after confirming the port is open
+
+            self.logger.info("Starting Flower ClientApp process...")
+            flwr_clientapp_process = subprocess.Popen(flwr_clientapp_command, shell=False)
+            interop_server.handle_signals(flwr_clientapp_process)
 
         self.logger.info("Press CTRL+C to stop the server and SuperNode process.")
 
-        try:
-            while not local_grpc_server.termination_event.is_set():
-                if self.shutdown_requested:
-                    local_grpc_server.terminate_supernode_process(supernode_process)
-                    local_grpc_server.stop_server()
-                time.sleep(0.1)
-        except KeyboardInterrupt:
-            local_grpc_server.terminate_supernode_process(supernode_process)
-            local_grpc_server.stop_server()
+        while not interop_server.termination_event.is_set():
+            if self.shutdown_requested:
+                if self.sgx_enabled:
+                    self.logger.info("Terminating Flower ClientApp process...")
+                    interop_server.terminate_supernode_process(flwr_clientapp_process)
+                    flwr_clientapp_process.wait()
+
+                self.logger.info("Shutting down the server and SuperNode process...")
+                interop_server.terminate_supernode_process(supernode_process)
+                interop_server.stop_server()
+            time.sleep(0.1)
+
 
     def set_tensor_dict(self, tensor_dict, with_opt_vars=False):
         """
@@ -186,3 +209,10 @@ def get_dynamic_port():
         # Get the assigned port number
         port = s.getsockname()[1]
     return port
+
+def is_port_open(host, port):
+    """Check if a port is open on the given host."""
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.settimeout(1)
+        result = sock.connect_ex((host, port))
+        return result == 0