We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
stop_fl.sh does not terminate client side worker process
The stop_fl.sh just does `touch shutdown.fl` , wait 5 seconds, then kill the process of pid.fl.


The worker process of client is not killed / terminated.


Steps to reproduce:
1. Launch server (start.sh)
2. Launch client (start.sh)
3. Using Admin: Upload app
4. Using Admin: Deploy app
5. Using Admin: Start app
6. Use stop_fl.sh

--- 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/app/client/worker_process.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 """Provides a command line interface for a federated client trainer."""
16 
17 import argparse
18 import os
19 import sys
20 import traceback
21 
22 from nvflare.apis.fl_constant import FLContextKey
23 from nvflare.apis.workspace import Workspace
24 from nvflare.fuel.sec.security_content_service import SecurityContentService
25 from nvflare.fuel.utils.argument_utils import parse_vars
26 from nvflare.private.defs import EngineConstant
27 from nvflare.private.fed.app.fl_conf import FLClientStarterConfiger
28 from nvflare.private.fed.client.client_json_config import ClientJsonConfigurator
29 from nvflare.private.fed.client.client_run_manager import ClientRunManager
30 from nvflare.private.fed.client.client_runner import ClientRunner
31 from nvflare.private.fed.client.client_status import ClientStatus
32 from nvflare.private.fed.client.command_agent import CommandAgent
33 
34 
35 def main():
36     """Worker_process start program."""
37     parser = argparse.ArgumentParser()
38     parser.add_argument("--workspace", "-m", type=str, help="WORKSPACE folder", required=True)
39     parser.add_argument("--startup", "-w", type=str, help="startup folder", required=True)
40 
41     parser.add_argument(
42         "--fed_client", "-s", type=str, help="an aggregation server specification json file", required=True
43     )
44 
45     parser.add_argument("--set", metavar="KEY=VALUE", nargs="*")
46 
47     parser.add_argument("--local_rank", type=int, default=0)
48 
49     args = parser.parse_args()
50     kv_list = parse_vars(args.set)
51 
52     args.train_config = os.path.join("config", "config_train.json")
53     config_folder = kv_list.get("config_folder", "")
54     secure_train = kv_list.get("secure_train", True)
55     if config_folder == "":
56         args.client_config = "config_fed_client.json"
57     else:
58         args.client_config = os.path.join(config_folder, "config_fed_client.json")
59     args.config_folder = config_folder
60     args.env = os.path.join("config", "environment.json")
61 
62     try:
63         remove_restart_file(args)
64     except BaseException:
65         print("Could not remove the restart.fl / shutdown.fl file.  Please check your system before starting FL.")
66         sys.exit(-1)
67 
68     restart_file = os.path.join(args.workspace, "restart.fl")
69     if os.path.exists(restart_file):
70         os.remove(restart_file)
71 
72     print("starting the client .....")
73 
74     deployer = None
75     command_agent = None
76 
77     startup = os.path.join(args.workspace, "startup")
78     SecurityContentService.initialize(content_folder=startup)
79 
80     try:
81         token_file = os.path.join(args.workspace, EngineConstant.CLIENT_TOKEN_FILE)
82         with open(token_file, "r") as f:
83             token = f.readline().strip()
84             run_number = f.readline().strip()
85             client_name = f.readline().strip()
86             listen_port = f.readline().strip()
87             print(
88                 "token is: {} run_number is: {} client_name: {} listen_port: {}".format(
89                     token, run_number, client_name, listen_port
90                 )
91             )
92 
93         startup = args.startup
94         app_root = os.path.join(args.workspace, "run_" + str(run_number), "app_" + client_name)
95 
96         app_log_config = os.path.join(app_root, config_folder, "log.config")
97         if os.path.exists(app_log_config):
98             args.log_config = app_log_config
99         else:
100             args.log_config = os.path.join(startup, "log.config")
101 
102         conf = FLClientStarterConfiger(
103             app_root=startup,
104             client_config_file_name=args.fed_client,
105             log_config_file_name=args.log_config,
106             kv_list=args.set,
107         )
108         conf.configure()
109 
110         deployer = conf.base_deployer
111         federated_client = deployer.create_fed_client()
112         federated_client.status = ClientStatus.STARTING
113 
114         federated_client.token = token
115         federated_client.client_name = client_name
116         federated_client.fl_ctx.set_prop(FLContextKey.CLIENT_NAME, client_name, private=False)
117         federated_client.fl_ctx.set_prop(EngineConstant.FL_TOKEN, token, private=False)
118         federated_client.fl_ctx.set_prop(FLContextKey.WORKSPACE_ROOT, args.workspace, private=True)
119 
120         client_config_file_name = os.path.join(app_root, args.client_config)
121         conf = ClientJsonConfigurator(
122             config_file_name=client_config_file_name,
123         )
124         conf.configure()
125 
126         workspace = Workspace(args.workspace, client_name, config_folder)
127         run_manager = ClientRunManager(
128             client_name=client_name,
129             run_num=int(run_number),
130             workspace=workspace,
131             client=federated_client,
132             components=conf.runner_config.components,
133             handlers=conf.runner_config.handlers,
134             conf=conf,
135         )
136         federated_client.run_manager = run_manager
137 
138         with run_manager.new_context() as fl_ctx:
139             fl_ctx.set_prop(FLContextKey.CLIENT_NAME, client_name, private=False)
140             fl_ctx.set_prop(EngineConstant.FL_TOKEN, token, private=False)
141             fl_ctx.set_prop(FLContextKey.WORKSPACE_ROOT, args.workspace, private=True)
142             fl_ctx.set_prop(FLContextKey.ARGS, args, sticky=True)
143             fl_ctx.set_prop(FLContextKey.APP_ROOT, app_root, private=True, sticky=True)
144             fl_ctx.set_prop(FLContextKey.WORKSPACE_OBJECT, workspace, private=True)
145             fl_ctx.set_prop(FLContextKey.SECURE_MODE, secure_train, private=True, sticky=True)
146 
147             client_runner = ClientRunner(config=conf.runner_config, run_num=int(run_number), engine=run_manager)
148             run_manager.add_handler(client_runner)
149             fl_ctx.set_prop(FLContextKey.RUNNER, client_runner, private=True)
150 
151             # # Start the thread for responding the inquire
152             # federated_client.stop_listen = False
153             # thread = threading.Thread(target=listen_command, args=[federated_client, int(listen_port), client_runner])
154             # thread.start()
155             # Start the command agent
156             command_agent = CommandAgent(federated_client, int(listen_port), client_runner)
157             command_agent.start(fl_ctx)
158 
159         federated_client.status = ClientStatus.STARTED
160         client_runner.run(app_root, args)
161 
162     except BaseException as e:
163         traceback.print_exc()
164         print("FL client execution exception: " + str(e))
165     finally:
166         # if federated_client:
167         #     federated_client.stop_listen = True
168         #     thread.join()
169         if command_agent:
170             command_agent.shutdown()
171         if deployer:
172             deployer.close()
173         # address = ('localhost', 6000)
174         # conn_client = Client(address, authkey='client process secret password'.encode())
175         # conn_client.send('bye')
176 
177 
178 def remove_restart_file(args):
179     """To remove the restart.fl file.
180 
181     Args:
182         args: command args
183 
184     """
185     restart_file = os.path.join(args.workspace, "restart.fl")
186     if os.path.exists(restart_file):
187         os.remove(restart_file)
188     restart_file = os.path.join(args.workspace, "shutdown.fl")
189     if os.path.exists(restart_file):
190         os.remove(restart_file)
191 
192 
193 if __name__ == "__main__":
194     """
195     This is the program when starting the child process for running the NVIDIA FLARE executor.
196     """
197 
198     main()
199
```
--- 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/app/client/worker_process.py b/nvflare/private/fed/app/client/worker_process.py
--- a/nvflare/private/fed/app/client/worker_process.py
+++ b/nvflare/private/fed/app/client/worker_process.py
@@ -17,8 +17,12 @@
 import argparse
 import os
 import sys
+import threading
+import time
 import traceback
 
+import psutil
+
 from nvflare.apis.fl_constant import FLContextKey
 from nvflare.apis.workspace import Workspace
 from nvflare.fuel.sec.security_content_service import SecurityContentService
@@ -32,23 +36,33 @@
 from nvflare.private.fed.client.command_agent import CommandAgent
 
 
+def check_parent_alive(parent_pid, stop_event: threading.Event):
+    while True:
+        if stop_event.is_set():
+            break
+        if not psutil.pid_exists(parent_pid):
+            # if parent is not alive, kill its worker process
+            os.killpg(os.getpgid(os.getpid()), 9)
+            break
+        time.sleep(1)
+
+
 def main():
-    """Worker_process start program."""
+    """Worker process start program."""
     parser = argparse.ArgumentParser()
     parser.add_argument("--workspace", "-m", type=str, help="WORKSPACE folder", required=True)
     parser.add_argument("--startup", "-w", type=str, help="startup folder", required=True)
-
     parser.add_argument(
         "--fed_client", "-s", type=str, help="an aggregation server specification json file", required=True
     )
-
     parser.add_argument("--set", metavar="KEY=VALUE", nargs="*")
-
     parser.add_argument("--local_rank", type=int, default=0)
-
     args = parser.parse_args()
     kv_list = parse_vars(args.set)
 
+    # get parent process id
+    parent_pid = os.getppid()
+
     args.train_config = os.path.join("config", "config_train.json")
     config_folder = kv_list.get("config_folder", "")
     secure_train = kv_list.get("secure_train", True)
@@ -68,7 +82,6 @@
     restart_file = os.path.join(args.workspace, "restart.fl")
     if os.path.exists(restart_file):
         os.remove(restart_file)
-
     print("starting the client .....")
 
     deployer = None
@@ -77,7 +90,13 @@
     startup = os.path.join(args.workspace, "startup")
     SecurityContentService.initialize(content_folder=startup)
 
+    federated_client = None
+    thread = None
+    stop_event = threading.Event()
     try:
+        # start parent process checking thread
+        thread = threading.Thread(target=check_parent_alive, args=(parent_pid, stop_event))
+        thread.start()
         token_file = os.path.join(args.workspace, EngineConstant.CLIENT_TOKEN_FILE)
         with open(token_file, "r") as f:
             token = f.readline().strip()
@@ -158,7 +177,6 @@
 
         federated_client.status = ClientStatus.STARTED
         client_runner.run(app_root, args)
-
     except BaseException as e:
         traceback.print_exc()
         print("FL client execution exception: " + str(e))
@@ -166,10 +184,16 @@
         # if federated_client:
         #     federated_client.stop_listen = True
         #     thread.join()
+        stop_event.set()
         if command_agent:
             command_agent.shutdown()
         if deployer:
             deployer.close()
+        if federated_client:
+            federated_client.close()
+        if thread and thread.is_alive():
+            thread.join()
+
         # address = ('localhost', 6000)
         # conn_client = Client(address, authkey='client process secret password'.encode())
         # conn_client.send('bye')
@@ -177,10 +201,8 @@
 
 def remove_restart_file(args):
     """To remove the restart.fl file.
-
     Args:
         args: command args
-
     """
     restart_file = os.path.join(args.workspace, "restart.fl")
     if os.path.exists(restart_file):

{"golden_diff": "diff --git a/nvflare/private/fed/app/client/worker_process.py b/nvflare/private/fed/app/client/worker_process.py\n--- a/nvflare/private/fed/app/client/worker_process.py\n+++ b/nvflare/private/fed/app/client/worker_process.py\n@@ -17,8 +17,12 @@\n import argparse\n import os\n import sys\n+import threading\n+import time\n import traceback\n \n+import psutil\n+\n from nvflare.apis.fl_constant import FLContextKey\n from nvflare.apis.workspace import Workspace\n from nvflare.fuel.sec.security_content_service import SecurityContentService\n@@ -32,23 +36,33 @@\n from nvflare.private.fed.client.command_agent import CommandAgent\n \n \n+def check_parent_alive(parent_pid, stop_event: threading.Event):\n+    while True:\n+        if stop_event.is_set():\n+            break\n+        if not psutil.pid_exists(parent_pid):\n+            # if parent is not alive, kill its worker process\n+            os.killpg(os.getpgid(os.getpid()), 9)\n+            break\n+        time.sleep(1)\n+\n+\n def main():\n-    \"\"\"Worker_process start program.\"\"\"\n+    \"\"\"Worker process start program.\"\"\"\n     parser = argparse.ArgumentParser()\n     parser.add_argument(\"--workspace\", \"-m\", type=str, help=\"WORKSPACE folder\", required=True)\n     parser.add_argument(\"--startup\", \"-w\", type=str, help=\"startup folder\", required=True)\n-\n     parser.add_argument(\n         \"--fed_client\", \"-s\", type=str, help=\"an aggregation server specification json file\", required=True\n     )\n-\n     parser.add_argument(\"--set\", metavar=\"KEY=VALUE\", nargs=\"*\")\n-\n     parser.add_argument(\"--local_rank\", type=int, default=0)\n-\n     args = parser.parse_args()\n     kv_list = parse_vars(args.set)\n \n+    # get parent process id\n+    parent_pid = os.getppid()\n+\n     args.train_config = os.path.join(\"config\", \"config_train.json\")\n     config_folder = kv_list.get(\"config_folder\", \"\")\n     secure_train = kv_list.get(\"secure_train\", True)\n@@ -68,7 +82,6 @@\n     restart_file = os.path.join(args.workspace, \"restart.fl\")\n     if os.path.exists(restart_file):\n         os.remove(restart_file)\n-\n     print(\"starting the client .....\")\n \n     deployer = None\n@@ -77,7 +90,13 @@\n     startup = os.path.join(args.workspace, \"startup\")\n     SecurityContentService.initialize(content_folder=startup)\n \n+    federated_client = None\n+    thread = None\n+    stop_event = threading.Event()\n     try:\n+        # start parent process checking thread\n+        thread = threading.Thread(target=check_parent_alive, args=(parent_pid, stop_event))\n+        thread.start()\n         token_file = os.path.join(args.workspace, EngineConstant.CLIENT_TOKEN_FILE)\n         with open(token_file, \"r\") as f:\n             token = f.readline().strip()\n@@ -158,7 +177,6 @@\n \n         federated_client.status = ClientStatus.STARTED\n         client_runner.run(app_root, args)\n-\n     except BaseException as e:\n         traceback.print_exc()\n         print(\"FL client execution exception: \" + str(e))\n@@ -166,10 +184,16 @@\n         # if federated_client:\n         #     federated_client.stop_listen = True\n         #     thread.join()\n+        stop_event.set()\n         if command_agent:\n             command_agent.shutdown()\n         if deployer:\n             deployer.close()\n+        if federated_client:\n+            federated_client.close()\n+        if thread and thread.is_alive():\n+            thread.join()\n+\n         # address = ('localhost', 6000)\n         # conn_client = Client(address, authkey='client process secret password'.encode())\n         # conn_client.send('bye')\n@@ -177,10 +201,8 @@\n \n def remove_restart_file(args):\n     \"\"\"To remove the restart.fl file.\n-\n     Args:\n         args: command args\n-\n     \"\"\"\n     restart_file = os.path.join(args.workspace, \"restart.fl\")\n     if os.path.exists(restart_file):\n", "issue": "stop_fl.sh does not terminate client side worker process\nThe stop_fl.sh just does `touch shutdown.fl` , wait 5 seconds, then kill the process of pid.fl.\r\n\r\n\r\nThe worker process of client is not killed / terminated.\r\n\r\n\r\nSteps to reproduce:\r\n1. Launch server (start.sh)\r\n2. Launch client (start.sh)\r\n3. Using Admin: Upload app\r\n4. Using Admin: Deploy app\r\n5. Using Admin: Start app\r\n6. Use stop_fl.sh\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\n\"\"\"Provides a command line interface for a federated client trainer.\"\"\"\n\nimport argparse\nimport os\nimport sys\nimport traceback\n\nfrom nvflare.apis.fl_constant import FLContextKey\nfrom nvflare.apis.workspace import Workspace\nfrom nvflare.fuel.sec.security_content_service import SecurityContentService\nfrom nvflare.fuel.utils.argument_utils import parse_vars\nfrom nvflare.private.defs import EngineConstant\nfrom nvflare.private.fed.app.fl_conf import FLClientStarterConfiger\nfrom nvflare.private.fed.client.client_json_config import ClientJsonConfigurator\nfrom nvflare.private.fed.client.client_run_manager import ClientRunManager\nfrom nvflare.private.fed.client.client_runner import ClientRunner\nfrom nvflare.private.fed.client.client_status import ClientStatus\nfrom nvflare.private.fed.client.command_agent import CommandAgent\n\n\ndef main():\n    \"\"\"Worker_process start program.\"\"\"\n    parser = argparse.ArgumentParser()\n    parser.add_argument(\"--workspace\", \"-m\", type=str, help=\"WORKSPACE folder\", required=True)\n    parser.add_argument(\"--startup\", \"-w\", type=str, help=\"startup folder\", required=True)\n\n    parser.add_argument(\n        \"--fed_client\", \"-s\", type=str, help=\"an aggregation server specification json file\", required=True\n    )\n\n    parser.add_argument(\"--set\", metavar=\"KEY=VALUE\", nargs=\"*\")\n\n    parser.add_argument(\"--local_rank\", type=int, default=0)\n\n    args = parser.parse_args()\n    kv_list = parse_vars(args.set)\n\n    args.train_config = os.path.join(\"config\", \"config_train.json\")\n    config_folder = kv_list.get(\"config_folder\", \"\")\n    secure_train = kv_list.get(\"secure_train\", True)\n    if config_folder == \"\":\n        args.client_config = \"config_fed_client.json\"\n    else:\n        args.client_config = os.path.join(config_folder, \"config_fed_client.json\")\n    args.config_folder = config_folder\n    args.env = os.path.join(\"config\", \"environment.json\")\n\n    try:\n        remove_restart_file(args)\n    except BaseException:\n        print(\"Could not remove the restart.fl / shutdown.fl file.  Please check your system before starting FL.\")\n        sys.exit(-1)\n\n    restart_file = os.path.join(args.workspace, \"restart.fl\")\n    if os.path.exists(restart_file):\n        os.remove(restart_file)\n\n    print(\"starting the client .....\")\n\n    deployer = None\n    command_agent = None\n\n    startup = os.path.join(args.workspace, \"startup\")\n    SecurityContentService.initialize(content_folder=startup)\n\n    try:\n        token_file = os.path.join(args.workspace, EngineConstant.CLIENT_TOKEN_FILE)\n        with open(token_file, \"r\") as f:\n            token = f.readline().strip()\n            run_number = f.readline().strip()\n            client_name = f.readline().strip()\n            listen_port = f.readline().strip()\n            print(\n                \"token is: {} run_number is: {} client_name: {} listen_port: {}\".format(\n                    token, run_number, client_name, listen_port\n                )\n            )\n\n        startup = args.startup\n        app_root = os.path.join(args.workspace, \"run_\" + str(run_number), \"app_\" + client_name)\n\n        app_log_config = os.path.join(app_root, config_folder, \"log.config\")\n        if os.path.exists(app_log_config):\n            args.log_config = app_log_config\n        else:\n            args.log_config = os.path.join(startup, \"log.config\")\n\n        conf = FLClientStarterConfiger(\n            app_root=startup,\n            client_config_file_name=args.fed_client,\n            log_config_file_name=args.log_config,\n            kv_list=args.set,\n        )\n        conf.configure()\n\n        deployer = conf.base_deployer\n        federated_client = deployer.create_fed_client()\n        federated_client.status = ClientStatus.STARTING\n\n        federated_client.token = token\n        federated_client.client_name = client_name\n        federated_client.fl_ctx.set_prop(FLContextKey.CLIENT_NAME, client_name, private=False)\n        federated_client.fl_ctx.set_prop(EngineConstant.FL_TOKEN, token, private=False)\n        federated_client.fl_ctx.set_prop(FLContextKey.WORKSPACE_ROOT, args.workspace, private=True)\n\n        client_config_file_name = os.path.join(app_root, args.client_config)\n        conf = ClientJsonConfigurator(\n            config_file_name=client_config_file_name,\n        )\n        conf.configure()\n\n        workspace = Workspace(args.workspace, client_name, config_folder)\n        run_manager = ClientRunManager(\n            client_name=client_name,\n            run_num=int(run_number),\n            workspace=workspace,\n            client=federated_client,\n            components=conf.runner_config.components,\n            handlers=conf.runner_config.handlers,\n            conf=conf,\n        )\n        federated_client.run_manager = run_manager\n\n        with run_manager.new_context() as fl_ctx:\n            fl_ctx.set_prop(FLContextKey.CLIENT_NAME, client_name, private=False)\n            fl_ctx.set_prop(EngineConstant.FL_TOKEN, token, private=False)\n            fl_ctx.set_prop(FLContextKey.WORKSPACE_ROOT, args.workspace, private=True)\n            fl_ctx.set_prop(FLContextKey.ARGS, args, sticky=True)\n            fl_ctx.set_prop(FLContextKey.APP_ROOT, app_root, private=True, sticky=True)\n            fl_ctx.set_prop(FLContextKey.WORKSPACE_OBJECT, workspace, private=True)\n            fl_ctx.set_prop(FLContextKey.SECURE_MODE, secure_train, private=True, sticky=True)\n\n            client_runner = ClientRunner(config=conf.runner_config, run_num=int(run_number), engine=run_manager)\n            run_manager.add_handler(client_runner)\n            fl_ctx.set_prop(FLContextKey.RUNNER, client_runner, private=True)\n\n            # # Start the thread for responding the inquire\n            # federated_client.stop_listen = False\n            # thread = threading.Thread(target=listen_command, args=[federated_client, int(listen_port), client_runner])\n            # thread.start()\n            # Start the command agent\n            command_agent = CommandAgent(federated_client, int(listen_port), client_runner)\n            command_agent.start(fl_ctx)\n\n        federated_client.status = ClientStatus.STARTED\n        client_runner.run(app_root, args)\n\n    except BaseException as e:\n        traceback.print_exc()\n        print(\"FL client execution exception: \" + str(e))\n    finally:\n        # if federated_client:\n        #     federated_client.stop_listen = True\n        #     thread.join()\n        if command_agent:\n            command_agent.shutdown()\n        if deployer:\n            deployer.close()\n        # address = ('localhost', 6000)\n        # conn_client = Client(address, authkey='client process secret password'.encode())\n        # conn_client.send('bye')\n\n\ndef remove_restart_file(args):\n    \"\"\"To remove the restart.fl file.\n\n    Args:\n        args: command args\n\n    \"\"\"\n    restart_file = os.path.join(args.workspace, \"restart.fl\")\n    if os.path.exists(restart_file):\n        os.remove(restart_file)\n    restart_file = os.path.join(args.workspace, \"shutdown.fl\")\n    if os.path.exists(restart_file):\n        os.remove(restart_file)\n\n\nif __name__ == \"__main__\":\n    \"\"\"\n    This is the program when starting the child process for running the NVIDIA FLARE executor.\n    \"\"\"\n\n    main()\n", "path": "nvflare/private/fed/app/client/worker_process.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\n\"\"\"Provides a command line interface for a federated client trainer.\"\"\"\n\nimport argparse\nimport os\nimport sys\nimport threading\nimport time\nimport traceback\n\nimport psutil\n\nfrom nvflare.apis.fl_constant import FLContextKey\nfrom nvflare.apis.workspace import Workspace\nfrom nvflare.fuel.sec.security_content_service import SecurityContentService\nfrom nvflare.fuel.utils.argument_utils import parse_vars\nfrom nvflare.private.defs import EngineConstant\nfrom nvflare.private.fed.app.fl_conf import FLClientStarterConfiger\nfrom nvflare.private.fed.client.client_json_config import ClientJsonConfigurator\nfrom nvflare.private.fed.client.client_run_manager import ClientRunManager\nfrom nvflare.private.fed.client.client_runner import ClientRunner\nfrom nvflare.private.fed.client.client_status import ClientStatus\nfrom nvflare.private.fed.client.command_agent import CommandAgent\n\n\ndef check_parent_alive(parent_pid, stop_event: threading.Event):\n    while True:\n        if stop_event.is_set():\n            break\n        if not psutil.pid_exists(parent_pid):\n            # if parent is not alive, kill its worker process\n            os.killpg(os.getpgid(os.getpid()), 9)\n            break\n        time.sleep(1)\n\n\ndef main():\n    \"\"\"Worker process start program.\"\"\"\n    parser = argparse.ArgumentParser()\n    parser.add_argument(\"--workspace\", \"-m\", type=str, help=\"WORKSPACE folder\", required=True)\n    parser.add_argument(\"--startup\", \"-w\", type=str, help=\"startup folder\", required=True)\n    parser.add_argument(\n        \"--fed_client\", \"-s\", type=str, help=\"an aggregation server specification json file\", required=True\n    )\n    parser.add_argument(\"--set\", metavar=\"KEY=VALUE\", nargs=\"*\")\n    parser.add_argument(\"--local_rank\", type=int, default=0)\n    args = parser.parse_args()\n    kv_list = parse_vars(args.set)\n\n    # get parent process id\n    parent_pid = os.getppid()\n\n    args.train_config = os.path.join(\"config\", \"config_train.json\")\n    config_folder = kv_list.get(\"config_folder\", \"\")\n    secure_train = kv_list.get(\"secure_train\", True)\n    if config_folder == \"\":\n        args.client_config = \"config_fed_client.json\"\n    else:\n        args.client_config = os.path.join(config_folder, \"config_fed_client.json\")\n    args.config_folder = config_folder\n    args.env = os.path.join(\"config\", \"environment.json\")\n\n    try:\n        remove_restart_file(args)\n    except BaseException:\n        print(\"Could not remove the restart.fl / shutdown.fl file.  Please check your system before starting FL.\")\n        sys.exit(-1)\n\n    restart_file = os.path.join(args.workspace, \"restart.fl\")\n    if os.path.exists(restart_file):\n        os.remove(restart_file)\n    print(\"starting the client .....\")\n\n    deployer = None\n    command_agent = None\n\n    startup = os.path.join(args.workspace, \"startup\")\n    SecurityContentService.initialize(content_folder=startup)\n\n    federated_client = None\n    thread = None\n    stop_event = threading.Event()\n    try:\n        # start parent process checking thread\n        thread = threading.Thread(target=check_parent_alive, args=(parent_pid, stop_event))\n        thread.start()\n        token_file = os.path.join(args.workspace, EngineConstant.CLIENT_TOKEN_FILE)\n        with open(token_file, \"r\") as f:\n            token = f.readline().strip()\n            run_number = f.readline().strip()\n            client_name = f.readline().strip()\n            listen_port = f.readline().strip()\n            print(\n                \"token is: {} run_number is: {} client_name: {} listen_port: {}\".format(\n                    token, run_number, client_name, listen_port\n                )\n            )\n\n        startup = args.startup\n        app_root = os.path.join(args.workspace, \"run_\" + str(run_number), \"app_\" + client_name)\n\n        app_log_config = os.path.join(app_root, config_folder, \"log.config\")\n        if os.path.exists(app_log_config):\n            args.log_config = app_log_config\n        else:\n            args.log_config = os.path.join(startup, \"log.config\")\n\n        conf = FLClientStarterConfiger(\n            app_root=startup,\n            client_config_file_name=args.fed_client,\n            log_config_file_name=args.log_config,\n            kv_list=args.set,\n        )\n        conf.configure()\n\n        deployer = conf.base_deployer\n        federated_client = deployer.create_fed_client()\n        federated_client.status = ClientStatus.STARTING\n\n        federated_client.token = token\n        federated_client.client_name = client_name\n        federated_client.fl_ctx.set_prop(FLContextKey.CLIENT_NAME, client_name, private=False)\n        federated_client.fl_ctx.set_prop(EngineConstant.FL_TOKEN, token, private=False)\n        federated_client.fl_ctx.set_prop(FLContextKey.WORKSPACE_ROOT, args.workspace, private=True)\n\n        client_config_file_name = os.path.join(app_root, args.client_config)\n        conf = ClientJsonConfigurator(\n            config_file_name=client_config_file_name,\n        )\n        conf.configure()\n\n        workspace = Workspace(args.workspace, client_name, config_folder)\n        run_manager = ClientRunManager(\n            client_name=client_name,\n            run_num=int(run_number),\n            workspace=workspace,\n            client=federated_client,\n            components=conf.runner_config.components,\n            handlers=conf.runner_config.handlers,\n            conf=conf,\n        )\n        federated_client.run_manager = run_manager\n\n        with run_manager.new_context() as fl_ctx:\n            fl_ctx.set_prop(FLContextKey.CLIENT_NAME, client_name, private=False)\n            fl_ctx.set_prop(EngineConstant.FL_TOKEN, token, private=False)\n            fl_ctx.set_prop(FLContextKey.WORKSPACE_ROOT, args.workspace, private=True)\n            fl_ctx.set_prop(FLContextKey.ARGS, args, sticky=True)\n            fl_ctx.set_prop(FLContextKey.APP_ROOT, app_root, private=True, sticky=True)\n            fl_ctx.set_prop(FLContextKey.WORKSPACE_OBJECT, workspace, private=True)\n            fl_ctx.set_prop(FLContextKey.SECURE_MODE, secure_train, private=True, sticky=True)\n\n            client_runner = ClientRunner(config=conf.runner_config, run_num=int(run_number), engine=run_manager)\n            run_manager.add_handler(client_runner)\n            fl_ctx.set_prop(FLContextKey.RUNNER, client_runner, private=True)\n\n            # # Start the thread for responding the inquire\n            # federated_client.stop_listen = False\n            # thread = threading.Thread(target=listen_command, args=[federated_client, int(listen_port), client_runner])\n            # thread.start()\n            # Start the command agent\n            command_agent = CommandAgent(federated_client, int(listen_port), client_runner)\n            command_agent.start(fl_ctx)\n\n        federated_client.status = ClientStatus.STARTED\n        client_runner.run(app_root, args)\n    except BaseException as e:\n        traceback.print_exc()\n        print(\"FL client execution exception: \" + str(e))\n    finally:\n        # if federated_client:\n        #     federated_client.stop_listen = True\n        #     thread.join()\n        stop_event.set()\n        if command_agent:\n            command_agent.shutdown()\n        if deployer:\n            deployer.close()\n        if federated_client:\n            federated_client.close()\n        if thread and thread.is_alive():\n            thread.join()\n\n        # address = ('localhost', 6000)\n        # conn_client = Client(address, authkey='client process secret password'.encode())\n        # conn_client.send('bye')\n\n\ndef remove_restart_file(args):\n    \"\"\"To remove the restart.fl file.\n    Args:\n        args: command args\n    \"\"\"\n    restart_file = os.path.join(args.workspace, \"restart.fl\")\n    if os.path.exists(restart_file):\n        os.remove(restart_file)\n    restart_file = os.path.join(args.workspace, \"shutdown.fl\")\n    if os.path.exists(restart_file):\n        os.remove(restart_file)\n\n\nif __name__ == \"__main__\":\n    \"\"\"\n    This is the program when starting the child process for running the NVIDIA FLARE executor.\n    \"\"\"\n\n    main()\n", "path": "nvflare/private/fed/app/client/worker_process.py"}]}