Multiple bug fixes from 2.4 (#2518)

* [2.4] Support client custom code in simulator (#2447)

* Support client custom code in simulator

* Fix client custom code

* Remove cancel_futures args (#2457)

* Fix sub_worker_process shutdown (#2458)

* Set GRPC_ENABLE_FORK_SUPPORT to False (#2474)

nvflare/apis/utils/reliable_message.py

@@ -357,7 +357,7 @@ def shutdown(cls):
         """
         if not cls._shutdown_asked:
             cls._shutdown_asked = True
-            cls._executor.shutdown(cancel_futures=True, wait=False)
+            cls._executor.shutdown(wait=False)
             cls._logger.info("ReliableMessage is shutdown")
 
     @classmethod

nvflare/fuel/f3/drivers/aio_grpc_driver.py

@@ -11,7 +11,9 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+
 import asyncio
+import os
 import random
 import threading
 import time
@@ -278,6 +280,9 @@ class AioGrpcDriver(BaseDriver):
 
     def __init__(self):
         super().__init__()
+        # GRPC with fork issue: https://github.com/grpc/grpc/issues/28557
+        os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "False"
+
         self.server = None
         self.options = GRPC_DEFAULT_OPTIONS
         self.logger = get_logger(self)

nvflare/fuel/f3/drivers/grpc_driver.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
 import threading
 from concurrent import futures
 from typing import Any, Dict, List, Union
@@ -202,6 +203,8 @@ def shutdown(self):
 class GrpcDriver(BaseDriver):
     def __init__(self):
         BaseDriver.__init__(self)
+        # GRPC with fork issue: https://github.com/grpc/grpc/issues/28557
+        os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "False"
         self.server = None
         self.closing = False
         self.max_workers = 100

nvflare/private/fed/app/client/sub_worker_process.py

@@ -294,8 +294,6 @@ def _handle_event(self, data):
 
     def _close(self, data):
         self.done = True
-        self.cell.stop()
-        # mpm.stop()
 
     def run(self):
         self.logger.info("SubWorkerExecutor process started.")

nvflare/private/fed/app/simulator/simulator_worker.py

@@ -235,6 +235,8 @@ def main(args):
     log_file = os.path.join(args.workspace, WorkspaceConstants.LOG_FILE_NAME)
     add_logfile_handler(log_file)
 
+    app_custom_folder = os.path.join(args.workspace, "custom")
+    sys.path.append(app_custom_folder)
     os.chdir(args.workspace)
     fobs_initialize()
     AuthorizationService.initialize(EmptyAuthorizer())