[CoreEngine] make the handshaking protocol work.

Author: fedml-alex

python/fedml/computing/scheduler/slave/base_slave_protocol_manager.py

@@ -48,6 +48,7 @@ def __init__(self, args, agent_config=None):
         self.topic_report_status = None
         self.topic_ota_msg = None
         self.topic_request_device_info = None
+        self.topic_request_device_info_from_mlops = None
         self.topic_client_logout = None
         self.topic_response_job_status = None
         self.topic_report_device_status_in_job = None
@@ -87,6 +88,9 @@ def generate_topics(self):
         # The topic for requesting device info from the client.
         self.topic_request_device_info = "server/client/request_device_info/" + str(self.edge_id)
 
+        # The topic for requesting device info from mlops.
+        self.topic_request_device_info_from_mlops = f"deploy/mlops/slave_agent/request_device_info/{self.edge_id}"
+
         # The topic for requesting device info from MLOps.
         self.topic_client_logout = "mlops/client/logout/" + str(self.edge_id)
 
@@ -114,6 +118,7 @@ def generate_topics(self):
         self.add_subscribe_topic(self.topic_report_status)
         self.add_subscribe_topic(self.topic_ota_msg)
         self.add_subscribe_topic(self.topic_request_device_info)
+        self.add_subscribe_topic(self.topic_request_device_info_from_mlops)
         self.add_subscribe_topic(self.topic_client_logout)
         self.add_subscribe_topic(self.topic_response_job_status)
         self.add_subscribe_topic(self.topic_report_device_status_in_job)
@@ -132,6 +137,7 @@ def add_protocol_handler(self):
         self.add_message_listener(self.topic_ota_msg, FedMLBaseSlaveProtocolManager.callback_client_ota_msg)
         self.add_message_listener(self.topic_report_status, self.callback_report_current_status)
         self.add_message_listener(self.topic_request_device_info, self.callback_report_device_info)
+        self.add_message_listener(self.topic_request_device_info_from_mlops, self.callback_request_device_info_from_mlops)
         self.add_message_listener(self.topic_client_logout, self.callback_client_logout)
         self.add_message_listener(self.topic_response_job_status, self.callback_response_job_status)
         self.add_message_listener(self.topic_report_device_status_in_job, self.callback_response_device_status_in_job)
@@ -402,6 +408,15 @@ def callback_report_device_info(self, topic, payload):
                 response_payload["context"] = context
             self.message_center.send_message(response_topic, json.dumps(response_payload), run_id=run_id)
 
+    def callback_request_device_info_from_mlops(self, topic, payload):
+        self.response_device_info_to_mlops(topic, payload)
+
+    def response_device_info_to_mlops(self, topic, payload):
+        response_topic = f"deploy/slave_agent/mlops/response_device_info"
+        response_payload = {"run_id": self.run_id, "slave_agent_device_id": self.edge_id,
+                            "fedml_version": fedml.__version__}
+        self.message_center.send_message(response_topic, json.dumps(response_payload))
+
     def callback_client_logout(self, topic, payload):
         payload_json = json.loads(payload)
         secret = payload_json.get("auth", None)

python/fedml/computing/scheduler/slave/slave_protocol_manager.py

@@ -13,19 +13,11 @@ class FedMLLaunchSlaveProtocolManager(FedMLBaseSlaveProtocolManager):
 
     def __init__(self, args, agent_config=None):
         FedMLBaseSlaveProtocolManager.__init__(self, args, agent_config=agent_config)
-        self.topic_request_deploy_slave_device_info_from_mlops = None
-        self.topic_request_deploy_master_device_info_from_mlops = None
-        self.topic_request_edge_device_info_from_mlops = None
 
     # Override
     def generate_topics(self):
         super().generate_topics()
 
-        # The topic for requesting device info from mlops.
-        self.topic_request_edge_device_info_from_mlops = f"deploy/mlops/slave_agent/request_device_info/{self.edge_id}"
-
-        self.add_subscribe_topic(self.topic_request_edge_device_info_from_mlops)
-
     # Override
     def add_protocol_handler(self):
         super().add_protocol_handler()
@@ -106,54 +98,8 @@ def _init_extra_items(self):
         os.environ["FEDML_DEPLOY_MASTER_ID"] = str(self.model_device_server_id)
         os.environ["FEDML_DEPLOY_WORKER_IDS"] = str(self.model_device_client_edge_id_list)
 
-        # Subscribe handshaking messages from MLOps.
-        self.subscribe_handshaking_messages_from_mlops()
-
         # Start the monitor process
         self.args = copy.deepcopy(in_args)
         self.mlops_metrics.stop_device_realtime_perf()
         self.mlops_metrics.report_device_realtime_perf(self.args, self.args.agent_config["mqtt_config"])
-        pass
-
-    def callback_response_device_info_to_mlops(self, topic, payload):
-        payload_json = json.loads(payload)
-        server_id = payload_json.get("server_id", 0)
-        run_id = payload_json.get("run_id", 0)
-        listen_edge_id = str(topic).split("/")[-1]
-        context = payload_json.get("context", None)
-        response_topic = f"deploy/slave_agent/mlops/response_device_info"
-        if self.mlops_metrics is not None and self.model_device_client_edge_id_list is not None and \
-                self.model_device_server_id is not None:
-            device_info_json = {
-                "edge_id": listen_edge_id,
-                "fedml_version": fedml.__version__,
-                "user_id": self.args.user
-            }
-            salve_device_ids = list()
-            for model_client_edge_id in self.model_device_client_edge_id_list:
-                salve_device_ids.append(model_client_edge_id)
-            response_payload = {"slave_device_id": self.model_device_client_edge_id_list[0],
-                                "slave_device_id_list": salve_device_ids,
-                                "master_device_id": self.model_device_server_id,
-                                "run_id": run_id, "edge_id": listen_edge_id,
-                                "edge_info": device_info_json}
-            if context is not None:
-                response_payload["context"] = context
-            self.message_center.send_message(response_topic, json.dumps(response_payload), run_id=run_id)
-
-    def subscribe_handshaking_messages_from_mlops(self):
-        # The topic for requesting deployment master device info from mlops.
-        self.topic_request_deploy_master_device_info_from_mlops = f"deploy/mlops/master_agent/request_device_info/{self.model_device_server_id}"
-
-        # The topic for requesting deployment slave device info from mlops.
-        self.topic_request_deploy_slave_device_info_from_mlops = f"deploy/mlops/slave_agent/request_device_info/{self.model_device_client_edge_id_list[0]}"
-
-        self.add_subscribe_topic(self.topic_request_deploy_master_device_info_from_mlops)
-        self.add_subscribe_topic(self.topic_request_deploy_slave_device_info_from_mlops)
-
-        self.add_message_listener(
-            self.topic_request_edge_device_info_from_mlops, self.callback_response_device_info_to_mlops)
-        self.add_message_listener(
-            self.topic_request_deploy_master_device_info_from_mlops, self.callback_response_device_info_to_mlops)
-        self.add_message_listener(
-            self.topic_request_deploy_slave_device_info_from_mlops, self.callback_response_device_info_to_mlops)
+        pass