We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Remove self.server_meta in fed_client_base.py
This is not used: https://github.com/NVIDIA/NVFlare/blob/main/nvflare/private/fed/client/fed_client_base.py#L285

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `nvflare/private/fed/client/fed_client_base.py`
Content:
```
1 # Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import logging
16 import threading
17 import time
18 from functools import partial
19 from multiprocessing.dummy import Pool as ThreadPool
20 from typing import List, Optional
21 
22 from nvflare.apis.filter import Filter
23 from nvflare.apis.fl_component import FLComponent
24 from nvflare.apis.fl_constant import FLContextKey
25 from nvflare.apis.fl_context import FLContext
26 from nvflare.apis.fl_exception import FLCommunicationError
27 from nvflare.apis.overseer_spec import SP
28 from nvflare.apis.shareable import Shareable
29 from nvflare.apis.signal import Signal
30 from nvflare.fuel.utils.argument_utils import parse_vars
31 from nvflare.private.defs import EngineConstant
32 from nvflare.security.logging import secure_format_exception
33 
34 from .client_status import ClientStatus
35 from .communicator import Communicator
36 
37 
38 class FederatedClientBase:
39     """The client-side base implementation of federated learning.
40 
41     This class provide the tools function which will be used in both FedClient and FedClientLite.
42     """
43 
44     def __init__(
45         self,
46         client_name,
47         client_args,
48         secure_train,
49         server_args=None,
50         retry_timeout=30,
51         client_state_processors: Optional[List[Filter]] = None,
52         handlers: Optional[List[FLComponent]] = None,
53         compression=None,
54         overseer_agent=None,
55         args=None,
56         components=None,
57     ):
58         """To init FederatedClientBase.
59 
60         Args:
61             client_name: client name
62             client_args: client config args
63             secure_train: True/False to indicate secure train
64             server_args: server config args
65             retry_timeout: retry timeout
66             client_state_processors: client state processor filters
67             handlers: handlers
68             compression: communication compression algorithm
69         """
70         self.logger = logging.getLogger(self.__class__.__name__)
71 
72         self.client_name = client_name
73         self.token = None
74         self.ssid = None
75         self.client_args = client_args
76         self.servers = server_args
77 
78         self.communicator = Communicator(
79             ssl_args=client_args,
80             secure_train=secure_train,
81             retry_timeout=retry_timeout,
82             client_state_processors=client_state_processors,
83             compression=compression,
84         )
85 
86         self.secure_train = secure_train
87         self.handlers = handlers
88         self.components = components
89 
90         self.heartbeat_done = False
91         self.fl_ctx = FLContext()
92         self.platform = None
93         self.abort_signal = Signal()
94         self.engine = None
95 
96         self.status = ClientStatus.NOT_STARTED
97         self.remote_tasks = None
98 
99         self.sp_established = False
100         self.overseer_agent = overseer_agent
101 
102         self.overseer_agent = self._init_agent(args)
103 
104         if secure_train:
105             if self.overseer_agent:
106                 self.overseer_agent.set_secure_context(
107                     ca_path=client_args["ssl_root_cert"],
108                     cert_path=client_args["ssl_cert"],
109                     prv_key_path=client_args["ssl_private_key"],
110                 )
111 
112         if self.overseer_agent:
113             self.overseer_agent.start(self.overseer_callback)
114 
115     def _init_agent(self, args=None):
116         kv_list = parse_vars(args.set)
117         sp = kv_list.get("sp")
118 
119         if sp:
120             fl_ctx = FLContext()
121             fl_ctx.set_prop(FLContextKey.SP_END_POINT, sp)
122             self.overseer_agent.initialize(fl_ctx)
123 
124         return self.overseer_agent
125 
126     def overseer_callback(self, overseer_agent):
127         if overseer_agent.is_shutdown():
128             self.engine.shutdown()
129             return
130 
131         sp = overseer_agent.get_primary_sp()
132         self.set_primary_sp(sp)
133 
134     def set_sp(self, project_name, sp: SP):
135         if sp and sp.primary is True:
136             server = self.servers[project_name].get("target")
137             location = sp.name + ":" + sp.fl_port
138             if server != location:
139                 self.servers[project_name]["target"] = location
140                 self.sp_established = True
141                 self.logger.info(f"Got the new primary SP: {location}")
142 
143             if self.ssid and self.ssid != sp.service_session_id:
144                 self.ssid = sp.service_session_id
145                 thread = threading.Thread(target=self._switch_ssid)
146                 thread.start()
147 
148     def _switch_ssid(self):
149         if self.engine:
150             for job_id in self.engine.get_all_job_ids():
151                 self.engine.abort_task(job_id)
152         # self.register()
153         self.logger.info(f"Primary SP switched to new SSID: {self.ssid}")
154 
155     def client_register(self, project_name):
156         """Register the client to the FL server.
157 
158         Args:
159             project_name: FL study project name.
160         """
161         if not self.token:
162             try:
163                 self.token, self.ssid = self.communicator.client_registration(
164                     self.client_name, self.servers, project_name
165                 )
166                 if self.token is not None:
167                     self.fl_ctx.set_prop(FLContextKey.CLIENT_NAME, self.client_name, private=False)
168                     self.fl_ctx.set_prop(EngineConstant.FL_TOKEN, self.token, private=False)
169                     self.logger.info(
170                         "Successfully registered client:{} for project {}. Token:{} SSID:{}".format(
171                             self.client_name, project_name, self.token, self.ssid
172                         )
173                     )
174 
175             except FLCommunicationError:
176                 self.communicator.heartbeat_done = True
177 
178     def fetch_execute_task(self, project_name, fl_ctx: FLContext):
179         """Fetch a task from the server.
180 
181         Args:
182             project_name: FL study project name
183             fl_ctx: FLContext
184 
185         Returns:
186             A CurrentTask message from server
187         """
188         try:
189             self.logger.debug("Starting to fetch execute task.")
190             task = self.communicator.getTask(self.servers, project_name, self.token, self.ssid, fl_ctx)
191 
192             return task
193         except FLCommunicationError as e:
194             self.logger.info(secure_format_exception(e))
195 
196     def push_execute_result(self, project_name, shareable: Shareable, fl_ctx: FLContext):
197         """Submit execution results of a task to server.
198 
199         Args:
200             project_name: FL study project name
201             shareable: Shareable object
202             fl_ctx: FLContext
203 
204         Returns:
205             A FederatedSummary message from the server.
206         """
207         try:
208             self.logger.info("Starting to push execute result.")
209             execute_task_name = fl_ctx.get_prop(FLContextKey.TASK_NAME)
210             message = self.communicator.submitUpdate(
211                 self.servers,
212                 project_name,
213                 self.token,
214                 self.ssid,
215                 fl_ctx,
216                 self.client_name,
217                 shareable,
218                 execute_task_name,
219             )
220 
221             return message
222         except FLCommunicationError as e:
223             self.logger.info(secure_format_exception(e))
224 
225     def send_aux_message(self, project_name, topic: str, shareable: Shareable, timeout: float, fl_ctx: FLContext):
226         """Send auxiliary message to the server.
227 
228         Args:
229             project_name: FL study project name
230             topic: aux topic name
231             shareable: Shareable object
232             timeout: communication timeout
233             fl_ctx: FLContext
234 
235         Returns:
236             A reply message
237         """
238         try:
239             self.logger.debug("Starting to send aux message.")
240             message = self.communicator.auxCommunicate(
241                 self.servers, project_name, self.token, self.ssid, fl_ctx, self.client_name, shareable, topic, timeout
242             )
243 
244             return message
245         except FLCommunicationError as e:
246             self.logger.info(secure_format_exception(e))
247 
248     def send_heartbeat(self, project_name):
249         try:
250             if self.token:
251                 while not self.engine:
252                     time.sleep(1.0)
253                 self.communicator.send_heartbeat(
254                     self.servers, project_name, self.token, self.ssid, self.client_name, self.engine
255                 )
256         except FLCommunicationError:
257             self.communicator.heartbeat_done = True
258 
259     def heartbeat(self):
260         """Sends a heartbeat from the client to the server."""
261         pool = None
262         try:
263             pool = ThreadPool(len(self.servers))
264             return pool.map(self.send_heartbeat, tuple(self.servers))
265         finally:
266             if pool:
267                 pool.terminate()
268 
269     def pull_task(self, fl_ctx: FLContext):
270         """Fetch remote models and update the local client's session."""
271         pool = None
272         try:
273             pool = ThreadPool(len(self.servers))
274             self.remote_tasks = pool.map(partial(self.fetch_execute_task, fl_ctx=fl_ctx), tuple(self.servers))
275             pull_success, task_name = self.check_progress(self.remote_tasks)
276             # # Update app_ctx's current round info
277             # if self.app_context and self.remote_models[0] is not None:
278             #     self.app_context.global_round = self.remote_models[0].meta.current_round
279             # TODO: if some of the servers failed
280             # return self.model_manager.assign_current_model(self.remote_models)
281             return pull_success, task_name, self.remote_tasks
282         finally:
283             if pool:
284                 pool.terminate()
285 
286     def push_results(self, shareable: Shareable, fl_ctx: FLContext):
287         """Push the local model to multiple servers."""
288         pool = None
289         try:
290             pool = ThreadPool(len(self.servers))
291             return pool.map(partial(self.push_execute_result, shareable=shareable, fl_ctx=fl_ctx), tuple(self.servers))
292         finally:
293             if pool:
294                 pool.terminate()
295 
296     def aux_send(self, topic, shareable: Shareable, timeout: float, fl_ctx: FLContext):
297         """Push the local model to multiple servers."""
298         pool = None
299         try:
300             pool = ThreadPool(len(self.servers))
301             messages = pool.map(
302                 partial(self.send_aux_message, topic=topic, shareable=shareable, timeout=timeout, fl_ctx=fl_ctx),
303                 tuple(self.servers),
304             )
305             if messages is not None and messages[0] is not None:
306                 # Only handle single server communication for now.
307                 return messages
308             else:
309                 return None
310         finally:
311             if pool:
312                 pool.terminate()
313 
314     def register(self):
315         """Push the local model to multiple servers."""
316         pool = None
317         try:
318             pool = ThreadPool(len(self.servers))
319             return pool.map(self.client_register, tuple(self.servers))
320         finally:
321             if pool:
322                 pool.terminate()
323 
324     def set_primary_sp(self, sp):
325         pool = None
326         try:
327             pool = ThreadPool(len(self.servers))
328             return pool.map(partial(self.set_sp, sp=sp), tuple(self.servers))
329         finally:
330             if pool:
331                 pool.terminate()
332 
333     def run_heartbeat(self):
334         """Periodically runs the heartbeat."""
335         self.heartbeat()
336 
337     def start_heartbeat(self):
338         heartbeat_thread = threading.Thread(target=self.run_heartbeat)
339         heartbeat_thread.start()
340 
341     def quit_remote(self, task_name, fl_ctx: FLContext):
342         """Sending the last message to the server before leaving.
343 
344         Args:
345             task_name: task name
346             fl_ctx: FLContext
347 
348         Returns: N/A
349 
350         """
351         return self.communicator.quit_remote(self.servers, task_name, self.token, self.ssid, fl_ctx)
352 
353     def set_client_engine(self, engine):
354         self.engine = engine
355 
356     def close(self):
357         """Quit the remote federated server, close the local session."""
358         self.logger.info(f"Shutting down client: {self.client_name}")
359         if self.overseer_agent:
360             self.overseer_agent.end()
361 
362         return 0
363 
364     def check_progress(self, remote_tasks):
365         if remote_tasks[0] is not None:
366             self.server_meta = remote_tasks[0].meta
367             return True, remote_tasks[0].task_name
368         else:
369             return False, None
370
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/nvflare/private/fed/client/fed_client_base.py b/nvflare/private/fed/client/fed_client_base.py
--- a/nvflare/private/fed/client/fed_client_base.py
+++ b/nvflare/private/fed/client/fed_client_base.py
@@ -35,6 +35,13 @@
 from .communicator import Communicator
 
 
+def _check_progress(remote_tasks):
+    if remote_tasks[0] is not None:
+        return True, remote_tasks[0].task_name
+    else:
+        return False, None
+
+
 class FederatedClientBase:
     """The client-side base implementation of federated learning.
 
@@ -272,12 +279,8 @@
         try:
             pool = ThreadPool(len(self.servers))
             self.remote_tasks = pool.map(partial(self.fetch_execute_task, fl_ctx=fl_ctx), tuple(self.servers))
-            pull_success, task_name = self.check_progress(self.remote_tasks)
-            # # Update app_ctx's current round info
-            # if self.app_context and self.remote_models[0] is not None:
-            #     self.app_context.global_round = self.remote_models[0].meta.current_round
+            pull_success, task_name = _check_progress(self.remote_tasks)
             # TODO: if some of the servers failed
-            # return self.model_manager.assign_current_model(self.remote_models)
             return pull_success, task_name, self.remote_tasks
         finally:
             if pool:
@@ -360,10 +363,3 @@
             self.overseer_agent.end()
 
         return 0
-
-    def check_progress(self, remote_tasks):
-        if remote_tasks[0] is not None:
-            self.server_meta = remote_tasks[0].meta
-            return True, remote_tasks[0].task_name
-        else:
-            return False, None

{"golden_diff": "diff --git a/nvflare/private/fed/client/fed_client_base.py b/nvflare/private/fed/client/fed_client_base.py\n--- a/nvflare/private/fed/client/fed_client_base.py\n+++ b/nvflare/private/fed/client/fed_client_base.py\n@@ -35,6 +35,13 @@\n from .communicator import Communicator\n \n \n+def _check_progress(remote_tasks):\n+    if remote_tasks[0] is not None:\n+        return True, remote_tasks[0].task_name\n+    else:\n+        return False, None\n+\n+\n class FederatedClientBase:\n     \"\"\"The client-side base implementation of federated learning.\n \n@@ -272,12 +279,8 @@\n         try:\n             pool = ThreadPool(len(self.servers))\n             self.remote_tasks = pool.map(partial(self.fetch_execute_task, fl_ctx=fl_ctx), tuple(self.servers))\n-            pull_success, task_name = self.check_progress(self.remote_tasks)\n-            # # Update app_ctx's current round info\n-            # if self.app_context and self.remote_models[0] is not None:\n-            #     self.app_context.global_round = self.remote_models[0].meta.current_round\n+            pull_success, task_name = _check_progress(self.remote_tasks)\n             # TODO: if some of the servers failed\n-            # return self.model_manager.assign_current_model(self.remote_models)\n             return pull_success, task_name, self.remote_tasks\n         finally:\n             if pool:\n@@ -360,10 +363,3 @@\n             self.overseer_agent.end()\n \n         return 0\n-\n-    def check_progress(self, remote_tasks):\n-        if remote_tasks[0] is not None:\n-            self.server_meta = remote_tasks[0].meta\n-            return True, remote_tasks[0].task_name\n-        else:\n-            return False, None\n", "issue": "Remove self.server_meta in fed_client_base.py\nThis is not used: https://github.com/NVIDIA/NVFlare/blob/main/nvflare/private/fed/client/fed_client_base.py#L285\n", "before_files": [{"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport logging\nimport threading\nimport time\nfrom functools import partial\nfrom multiprocessing.dummy import Pool as ThreadPool\nfrom typing import List, Optional\n\nfrom nvflare.apis.filter import Filter\nfrom nvflare.apis.fl_component import FLComponent\nfrom nvflare.apis.fl_constant import FLContextKey\nfrom nvflare.apis.fl_context import FLContext\nfrom nvflare.apis.fl_exception import FLCommunicationError\nfrom nvflare.apis.overseer_spec import SP\nfrom nvflare.apis.shareable import Shareable\nfrom nvflare.apis.signal import Signal\nfrom nvflare.fuel.utils.argument_utils import parse_vars\nfrom nvflare.private.defs import EngineConstant\nfrom nvflare.security.logging import secure_format_exception\n\nfrom .client_status import ClientStatus\nfrom .communicator import Communicator\n\n\nclass FederatedClientBase:\n    \"\"\"The client-side base implementation of federated learning.\n\n    This class provide the tools function which will be used in both FedClient and FedClientLite.\n    \"\"\"\n\n    def __init__(\n        self,\n        client_name,\n        client_args,\n        secure_train,\n        server_args=None,\n        retry_timeout=30,\n        client_state_processors: Optional[List[Filter]] = None,\n        handlers: Optional[List[FLComponent]] = None,\n        compression=None,\n        overseer_agent=None,\n        args=None,\n        components=None,\n    ):\n        \"\"\"To init FederatedClientBase.\n\n        Args:\n            client_name: client name\n            client_args: client config args\n            secure_train: True/False to indicate secure train\n            server_args: server config args\n            retry_timeout: retry timeout\n            client_state_processors: client state processor filters\n            handlers: handlers\n            compression: communication compression algorithm\n        \"\"\"\n        self.logger = logging.getLogger(self.__class__.__name__)\n\n        self.client_name = client_name\n        self.token = None\n        self.ssid = None\n        self.client_args = client_args\n        self.servers = server_args\n\n        self.communicator = Communicator(\n            ssl_args=client_args,\n            secure_train=secure_train,\n            retry_timeout=retry_timeout,\n            client_state_processors=client_state_processors,\n            compression=compression,\n        )\n\n        self.secure_train = secure_train\n        self.handlers = handlers\n        self.components = components\n\n        self.heartbeat_done = False\n        self.fl_ctx = FLContext()\n        self.platform = None\n        self.abort_signal = Signal()\n        self.engine = None\n\n        self.status = ClientStatus.NOT_STARTED\n        self.remote_tasks = None\n\n        self.sp_established = False\n        self.overseer_agent = overseer_agent\n\n        self.overseer_agent = self._init_agent(args)\n\n        if secure_train:\n            if self.overseer_agent:\n                self.overseer_agent.set_secure_context(\n                    ca_path=client_args[\"ssl_root_cert\"],\n                    cert_path=client_args[\"ssl_cert\"],\n                    prv_key_path=client_args[\"ssl_private_key\"],\n                )\n\n        if self.overseer_agent:\n            self.overseer_agent.start(self.overseer_callback)\n\n    def _init_agent(self, args=None):\n        kv_list = parse_vars(args.set)\n        sp = kv_list.get(\"sp\")\n\n        if sp:\n            fl_ctx = FLContext()\n            fl_ctx.set_prop(FLContextKey.SP_END_POINT, sp)\n            self.overseer_agent.initialize(fl_ctx)\n\n        return self.overseer_agent\n\n    def overseer_callback(self, overseer_agent):\n        if overseer_agent.is_shutdown():\n            self.engine.shutdown()\n            return\n\n        sp = overseer_agent.get_primary_sp()\n        self.set_primary_sp(sp)\n\n    def set_sp(self, project_name, sp: SP):\n        if sp and sp.primary is True:\n            server = self.servers[project_name].get(\"target\")\n            location = sp.name + \":\" + sp.fl_port\n            if server != location:\n                self.servers[project_name][\"target\"] = location\n                self.sp_established = True\n                self.logger.info(f\"Got the new primary SP: {location}\")\n\n            if self.ssid and self.ssid != sp.service_session_id:\n                self.ssid = sp.service_session_id\n                thread = threading.Thread(target=self._switch_ssid)\n                thread.start()\n\n    def _switch_ssid(self):\n        if self.engine:\n            for job_id in self.engine.get_all_job_ids():\n                self.engine.abort_task(job_id)\n        # self.register()\n        self.logger.info(f\"Primary SP switched to new SSID: {self.ssid}\")\n\n    def client_register(self, project_name):\n        \"\"\"Register the client to the FL server.\n\n        Args:\n            project_name: FL study project name.\n        \"\"\"\n        if not self.token:\n            try:\n                self.token, self.ssid = self.communicator.client_registration(\n                    self.client_name, self.servers, project_name\n                )\n                if self.token is not None:\n                    self.fl_ctx.set_prop(FLContextKey.CLIENT_NAME, self.client_name, private=False)\n                    self.fl_ctx.set_prop(EngineConstant.FL_TOKEN, self.token, private=False)\n                    self.logger.info(\n                        \"Successfully registered client:{} for project {}. Token:{} SSID:{}\".format(\n                            self.client_name, project_name, self.token, self.ssid\n                        )\n                    )\n\n            except FLCommunicationError:\n                self.communicator.heartbeat_done = True\n\n    def fetch_execute_task(self, project_name, fl_ctx: FLContext):\n        \"\"\"Fetch a task from the server.\n\n        Args:\n            project_name: FL study project name\n            fl_ctx: FLContext\n\n        Returns:\n            A CurrentTask message from server\n        \"\"\"\n        try:\n            self.logger.debug(\"Starting to fetch execute task.\")\n            task = self.communicator.getTask(self.servers, project_name, self.token, self.ssid, fl_ctx)\n\n            return task\n        except FLCommunicationError as e:\n            self.logger.info(secure_format_exception(e))\n\n    def push_execute_result(self, project_name, shareable: Shareable, fl_ctx: FLContext):\n        \"\"\"Submit execution results of a task to server.\n\n        Args:\n            project_name: FL study project name\n            shareable: Shareable object\n            fl_ctx: FLContext\n\n        Returns:\n            A FederatedSummary message from the server.\n        \"\"\"\n        try:\n            self.logger.info(\"Starting to push execute result.\")\n            execute_task_name = fl_ctx.get_prop(FLContextKey.TASK_NAME)\n            message = self.communicator.submitUpdate(\n                self.servers,\n                project_name,\n                self.token,\n                self.ssid,\n                fl_ctx,\n                self.client_name,\n                shareable,\n                execute_task_name,\n            )\n\n            return message\n        except FLCommunicationError as e:\n            self.logger.info(secure_format_exception(e))\n\n    def send_aux_message(self, project_name, topic: str, shareable: Shareable, timeout: float, fl_ctx: FLContext):\n        \"\"\"Send auxiliary message to the server.\n\n        Args:\n            project_name: FL study project name\n            topic: aux topic name\n            shareable: Shareable object\n            timeout: communication timeout\n            fl_ctx: FLContext\n\n        Returns:\n            A reply message\n        \"\"\"\n        try:\n            self.logger.debug(\"Starting to send aux message.\")\n            message = self.communicator.auxCommunicate(\n                self.servers, project_name, self.token, self.ssid, fl_ctx, self.client_name, shareable, topic, timeout\n            )\n\n            return message\n        except FLCommunicationError as e:\n            self.logger.info(secure_format_exception(e))\n\n    def send_heartbeat(self, project_name):\n        try:\n            if self.token:\n                while not self.engine:\n                    time.sleep(1.0)\n                self.communicator.send_heartbeat(\n                    self.servers, project_name, self.token, self.ssid, self.client_name, self.engine\n                )\n        except FLCommunicationError:\n            self.communicator.heartbeat_done = True\n\n    def heartbeat(self):\n        \"\"\"Sends a heartbeat from the client to the server.\"\"\"\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            return pool.map(self.send_heartbeat, tuple(self.servers))\n        finally:\n            if pool:\n                pool.terminate()\n\n    def pull_task(self, fl_ctx: FLContext):\n        \"\"\"Fetch remote models and update the local client's session.\"\"\"\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            self.remote_tasks = pool.map(partial(self.fetch_execute_task, fl_ctx=fl_ctx), tuple(self.servers))\n            pull_success, task_name = self.check_progress(self.remote_tasks)\n            # # Update app_ctx's current round info\n            # if self.app_context and self.remote_models[0] is not None:\n            #     self.app_context.global_round = self.remote_models[0].meta.current_round\n            # TODO: if some of the servers failed\n            # return self.model_manager.assign_current_model(self.remote_models)\n            return pull_success, task_name, self.remote_tasks\n        finally:\n            if pool:\n                pool.terminate()\n\n    def push_results(self, shareable: Shareable, fl_ctx: FLContext):\n        \"\"\"Push the local model to multiple servers.\"\"\"\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            return pool.map(partial(self.push_execute_result, shareable=shareable, fl_ctx=fl_ctx), tuple(self.servers))\n        finally:\n            if pool:\n                pool.terminate()\n\n    def aux_send(self, topic, shareable: Shareable, timeout: float, fl_ctx: FLContext):\n        \"\"\"Push the local model to multiple servers.\"\"\"\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            messages = pool.map(\n                partial(self.send_aux_message, topic=topic, shareable=shareable, timeout=timeout, fl_ctx=fl_ctx),\n                tuple(self.servers),\n            )\n            if messages is not None and messages[0] is not None:\n                # Only handle single server communication for now.\n                return messages\n            else:\n                return None\n        finally:\n            if pool:\n                pool.terminate()\n\n    def register(self):\n        \"\"\"Push the local model to multiple servers.\"\"\"\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            return pool.map(self.client_register, tuple(self.servers))\n        finally:\n            if pool:\n                pool.terminate()\n\n    def set_primary_sp(self, sp):\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            return pool.map(partial(self.set_sp, sp=sp), tuple(self.servers))\n        finally:\n            if pool:\n                pool.terminate()\n\n    def run_heartbeat(self):\n        \"\"\"Periodically runs the heartbeat.\"\"\"\n        self.heartbeat()\n\n    def start_heartbeat(self):\n        heartbeat_thread = threading.Thread(target=self.run_heartbeat)\n        heartbeat_thread.start()\n\n    def quit_remote(self, task_name, fl_ctx: FLContext):\n        \"\"\"Sending the last message to the server before leaving.\n\n        Args:\n            task_name: task name\n            fl_ctx: FLContext\n\n        Returns: N/A\n\n        \"\"\"\n        return self.communicator.quit_remote(self.servers, task_name, self.token, self.ssid, fl_ctx)\n\n    def set_client_engine(self, engine):\n        self.engine = engine\n\n    def close(self):\n        \"\"\"Quit the remote federated server, close the local session.\"\"\"\n        self.logger.info(f\"Shutting down client: {self.client_name}\")\n        if self.overseer_agent:\n            self.overseer_agent.end()\n\n        return 0\n\n    def check_progress(self, remote_tasks):\n        if remote_tasks[0] is not None:\n            self.server_meta = remote_tasks[0].meta\n            return True, remote_tasks[0].task_name\n        else:\n            return False, None\n", "path": "nvflare/private/fed/client/fed_client_base.py"}], "after_files": [{"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport logging\nimport threading\nimport time\nfrom functools import partial\nfrom multiprocessing.dummy import Pool as ThreadPool\nfrom typing import List, Optional\n\nfrom nvflare.apis.filter import Filter\nfrom nvflare.apis.fl_component import FLComponent\nfrom nvflare.apis.fl_constant import FLContextKey\nfrom nvflare.apis.fl_context import FLContext\nfrom nvflare.apis.fl_exception import FLCommunicationError\nfrom nvflare.apis.overseer_spec import SP\nfrom nvflare.apis.shareable import Shareable\nfrom nvflare.apis.signal import Signal\nfrom nvflare.fuel.utils.argument_utils import parse_vars\nfrom nvflare.private.defs import EngineConstant\nfrom nvflare.security.logging import secure_format_exception\n\nfrom .client_status import ClientStatus\nfrom .communicator import Communicator\n\n\ndef _check_progress(remote_tasks):\n    if remote_tasks[0] is not None:\n        return True, remote_tasks[0].task_name\n    else:\n        return False, None\n\n\nclass FederatedClientBase:\n    \"\"\"The client-side base implementation of federated learning.\n\n    This class provide the tools function which will be used in both FedClient and FedClientLite.\n    \"\"\"\n\n    def __init__(\n        self,\n        client_name,\n        client_args,\n        secure_train,\n        server_args=None,\n        retry_timeout=30,\n        client_state_processors: Optional[List[Filter]] = None,\n        handlers: Optional[List[FLComponent]] = None,\n        compression=None,\n        overseer_agent=None,\n        args=None,\n        components=None,\n    ):\n        \"\"\"To init FederatedClientBase.\n\n        Args:\n            client_name: client name\n            client_args: client config args\n            secure_train: True/False to indicate secure train\n            server_args: server config args\n            retry_timeout: retry timeout\n            client_state_processors: client state processor filters\n            handlers: handlers\n            compression: communication compression algorithm\n        \"\"\"\n        self.logger = logging.getLogger(self.__class__.__name__)\n\n        self.client_name = client_name\n        self.token = None\n        self.ssid = None\n        self.client_args = client_args\n        self.servers = server_args\n\n        self.communicator = Communicator(\n            ssl_args=client_args,\n            secure_train=secure_train,\n            retry_timeout=retry_timeout,\n            client_state_processors=client_state_processors,\n            compression=compression,\n        )\n\n        self.secure_train = secure_train\n        self.handlers = handlers\n        self.components = components\n\n        self.heartbeat_done = False\n        self.fl_ctx = FLContext()\n        self.platform = None\n        self.abort_signal = Signal()\n        self.engine = None\n\n        self.status = ClientStatus.NOT_STARTED\n        self.remote_tasks = None\n\n        self.sp_established = False\n        self.overseer_agent = overseer_agent\n\n        self.overseer_agent = self._init_agent(args)\n\n        if secure_train:\n            if self.overseer_agent:\n                self.overseer_agent.set_secure_context(\n                    ca_path=client_args[\"ssl_root_cert\"],\n                    cert_path=client_args[\"ssl_cert\"],\n                    prv_key_path=client_args[\"ssl_private_key\"],\n                )\n\n        if self.overseer_agent:\n            self.overseer_agent.start(self.overseer_callback)\n\n    def _init_agent(self, args=None):\n        kv_list = parse_vars(args.set)\n        sp = kv_list.get(\"sp\")\n\n        if sp:\n            fl_ctx = FLContext()\n            fl_ctx.set_prop(FLContextKey.SP_END_POINT, sp)\n            self.overseer_agent.initialize(fl_ctx)\n\n        return self.overseer_agent\n\n    def overseer_callback(self, overseer_agent):\n        if overseer_agent.is_shutdown():\n            self.engine.shutdown()\n            return\n\n        sp = overseer_agent.get_primary_sp()\n        self.set_primary_sp(sp)\n\n    def set_sp(self, project_name, sp: SP):\n        if sp and sp.primary is True:\n            server = self.servers[project_name].get(\"target\")\n            location = sp.name + \":\" + sp.fl_port\n            if server != location:\n                self.servers[project_name][\"target\"] = location\n                self.sp_established = True\n                self.logger.info(f\"Got the new primary SP: {location}\")\n\n            if self.ssid and self.ssid != sp.service_session_id:\n                self.ssid = sp.service_session_id\n                thread = threading.Thread(target=self._switch_ssid)\n                thread.start()\n\n    def _switch_ssid(self):\n        if self.engine:\n            for job_id in self.engine.get_all_job_ids():\n                self.engine.abort_task(job_id)\n        # self.register()\n        self.logger.info(f\"Primary SP switched to new SSID: {self.ssid}\")\n\n    def client_register(self, project_name):\n        \"\"\"Register the client to the FL server.\n\n        Args:\n            project_name: FL study project name.\n        \"\"\"\n        if not self.token:\n            try:\n                self.token, self.ssid = self.communicator.client_registration(\n                    self.client_name, self.servers, project_name\n                )\n                if self.token is not None:\n                    self.fl_ctx.set_prop(FLContextKey.CLIENT_NAME, self.client_name, private=False)\n                    self.fl_ctx.set_prop(EngineConstant.FL_TOKEN, self.token, private=False)\n                    self.logger.info(\n                        \"Successfully registered client:{} for project {}. Token:{} SSID:{}\".format(\n                            self.client_name, project_name, self.token, self.ssid\n                        )\n                    )\n\n            except FLCommunicationError:\n                self.communicator.heartbeat_done = True\n\n    def fetch_execute_task(self, project_name, fl_ctx: FLContext):\n        \"\"\"Fetch a task from the server.\n\n        Args:\n            project_name: FL study project name\n            fl_ctx: FLContext\n\n        Returns:\n            A CurrentTask message from server\n        \"\"\"\n        try:\n            self.logger.debug(\"Starting to fetch execute task.\")\n            task = self.communicator.getTask(self.servers, project_name, self.token, self.ssid, fl_ctx)\n\n            return task\n        except FLCommunicationError as e:\n            self.logger.info(secure_format_exception(e))\n\n    def push_execute_result(self, project_name, shareable: Shareable, fl_ctx: FLContext):\n        \"\"\"Submit execution results of a task to server.\n\n        Args:\n            project_name: FL study project name\n            shareable: Shareable object\n            fl_ctx: FLContext\n\n        Returns:\n            A FederatedSummary message from the server.\n        \"\"\"\n        try:\n            self.logger.info(\"Starting to push execute result.\")\n            execute_task_name = fl_ctx.get_prop(FLContextKey.TASK_NAME)\n            message = self.communicator.submitUpdate(\n                self.servers,\n                project_name,\n                self.token,\n                self.ssid,\n                fl_ctx,\n                self.client_name,\n                shareable,\n                execute_task_name,\n            )\n\n            return message\n        except FLCommunicationError as e:\n            self.logger.info(secure_format_exception(e))\n\n    def send_aux_message(self, project_name, topic: str, shareable: Shareable, timeout: float, fl_ctx: FLContext):\n        \"\"\"Send auxiliary message to the server.\n\n        Args:\n            project_name: FL study project name\n            topic: aux topic name\n            shareable: Shareable object\n            timeout: communication timeout\n            fl_ctx: FLContext\n\n        Returns:\n            A reply message\n        \"\"\"\n        try:\n            self.logger.debug(\"Starting to send aux message.\")\n            message = self.communicator.auxCommunicate(\n                self.servers, project_name, self.token, self.ssid, fl_ctx, self.client_name, shareable, topic, timeout\n            )\n\n            return message\n        except FLCommunicationError as e:\n            self.logger.info(secure_format_exception(e))\n\n    def send_heartbeat(self, project_name):\n        try:\n            if self.token:\n                while not self.engine:\n                    time.sleep(1.0)\n                self.communicator.send_heartbeat(\n                    self.servers, project_name, self.token, self.ssid, self.client_name, self.engine\n                )\n        except FLCommunicationError:\n            self.communicator.heartbeat_done = True\n\n    def heartbeat(self):\n        \"\"\"Sends a heartbeat from the client to the server.\"\"\"\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            return pool.map(self.send_heartbeat, tuple(self.servers))\n        finally:\n            if pool:\n                pool.terminate()\n\n    def pull_task(self, fl_ctx: FLContext):\n        \"\"\"Fetch remote models and update the local client's session.\"\"\"\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            self.remote_tasks = pool.map(partial(self.fetch_execute_task, fl_ctx=fl_ctx), tuple(self.servers))\n            pull_success, task_name = _check_progress(self.remote_tasks)\n            # TODO: if some of the servers failed\n            return pull_success, task_name, self.remote_tasks\n        finally:\n            if pool:\n                pool.terminate()\n\n    def push_results(self, shareable: Shareable, fl_ctx: FLContext):\n        \"\"\"Push the local model to multiple servers.\"\"\"\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            return pool.map(partial(self.push_execute_result, shareable=shareable, fl_ctx=fl_ctx), tuple(self.servers))\n        finally:\n            if pool:\n                pool.terminate()\n\n    def aux_send(self, topic, shareable: Shareable, timeout: float, fl_ctx: FLContext):\n        \"\"\"Push the local model to multiple servers.\"\"\"\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            messages = pool.map(\n                partial(self.send_aux_message, topic=topic, shareable=shareable, timeout=timeout, fl_ctx=fl_ctx),\n                tuple(self.servers),\n            )\n            if messages is not None and messages[0] is not None:\n                # Only handle single server communication for now.\n                return messages\n            else:\n                return None\n        finally:\n            if pool:\n                pool.terminate()\n\n    def register(self):\n        \"\"\"Push the local model to multiple servers.\"\"\"\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            return pool.map(self.client_register, tuple(self.servers))\n        finally:\n            if pool:\n                pool.terminate()\n\n    def set_primary_sp(self, sp):\n        pool = None\n        try:\n            pool = ThreadPool(len(self.servers))\n            return pool.map(partial(self.set_sp, sp=sp), tuple(self.servers))\n        finally:\n            if pool:\n                pool.terminate()\n\n    def run_heartbeat(self):\n        \"\"\"Periodically runs the heartbeat.\"\"\"\n        self.heartbeat()\n\n    def start_heartbeat(self):\n        heartbeat_thread = threading.Thread(target=self.run_heartbeat)\n        heartbeat_thread.start()\n\n    def quit_remote(self, task_name, fl_ctx: FLContext):\n        \"\"\"Sending the last message to the server before leaving.\n\n        Args:\n            task_name: task name\n            fl_ctx: FLContext\n\n        Returns: N/A\n\n        \"\"\"\n        return self.communicator.quit_remote(self.servers, task_name, self.token, self.ssid, fl_ctx)\n\n    def set_client_engine(self, engine):\n        self.engine = engine\n\n    def close(self):\n        \"\"\"Quit the remote federated server, close the local session.\"\"\"\n        self.logger.info(f\"Shutting down client: {self.client_name}\")\n        if self.overseer_agent:\n            self.overseer_agent.end()\n\n        return 0\n", "path": "nvflare/private/fed/client/fed_client_base.py"}]}