Merge pull request #2150 from FedML-AI/raphael/fix-multi-subfd

[Deploy] Recursively find the model serving package folder

Author: Raphael-Jin

python/fedml/computing/scheduler/comm_utils/file_utils.py

@@ -0,0 +1,13 @@
+import os
+
+
+def find_file_inside_folder(folder_path, file_name):
+    """
+    Recursively search for a file inside a folder and its sub-folders.
+    return the full path of the file if found, otherwise return None.
+    """
+    for root, dirs, files in os.walk(folder_path):
+        if file_name in files:
+            return os.path.join(root, file_name)
+
+    return None

python/fedml/computing/scheduler/model_scheduler/worker_job_runner.py

@@ -10,6 +10,7 @@
 import yaml
 from fedml.computing.scheduler.comm_utils.job_utils import JobRunnerUtils
 from fedml.core.mlops import MLOpsRuntimeLog
+from fedml.computing.scheduler.comm_utils import file_utils
 from .device_client_constants import ClientConstants
 from .device_model_cache import FedMLModelCache
 from ..scheduler_core.general_constants import GeneralConstants
@@ -205,7 +206,7 @@ def run_impl(self, run_extend_queue_list, sender_message_center,
         # Check if the package is already downloaded
         unzip_package_path = ""
         if os.path.exists(os.path.join(models_root_dir, parent_fd)):
-            unzip_package_path = self.find_previous_downloaded_pkg(os.path.join(models_root_dir, parent_fd), model_name)
+            unzip_package_path = self.find_previous_downloaded_pkg(os.path.join(models_root_dir, parent_fd))
 
         # Download the package if not found
         if unzip_package_path == "":
@@ -510,30 +511,13 @@ def build_dynamic_constrain_variables(self, run_id, run_config):
         pass
 
     @staticmethod
-    def find_previous_downloaded_pkg(parent_dir: str, model_name: str) -> str:
-        unzip_fd = ""
-        res = ""
-
-        for folder in os.listdir(parent_dir):
-            if folder.startswith("unzip_fedml_run"):
-                unzip_fd = os.path.join(parent_dir, folder)
-                break
-
-        exact_matched = False
-
-        if unzip_fd == "":
-            return res
-
-        for folder in os.listdir(unzip_fd):
-            if folder == model_name:
-                res = os.path.join(unzip_fd, folder)
-                exact_matched = True
-                break
-
-        if not exact_matched:
-            # Use the first folder found
-            for folder in os.listdir(unzip_fd):
-                res = os.path.join(unzip_fd, folder)
-                break
-
-        return res
+    def find_previous_downloaded_pkg(parent_dir: str) -> str:
+        """
+        Find a folder inside parent_dir that contains the fedml_model_config.yaml file.
+        """
+        res = file_utils.find_file_inside_folder(parent_dir, ClientConstants.MODEL_REQUIRED_MODEL_CONFIG_FILE)
+        if res is not None:
+            # return the parent folder of res
+            return os.path.dirname(res)
+        else:
+            return ""

python/fedml/computing/scheduler/scheduler_core/scheduler_base_job_runner.py

@@ -138,10 +138,13 @@ def get_client_id_list(self, server_edge_id_list):
 
     @staticmethod
     def unzip_file(zip_file, unzip_file_path) -> str:
+        unzipped_file_name = ""
         if zipfile.is_zipfile(zip_file):
-            with zipfile.ZipFile(zip_file, "r") as zipf:
+            with (zipfile.ZipFile(zip_file, "r") as zipf):
                 zipf.extractall(unzip_file_path)
-                unzipped_file_name = zipf.namelist()[0]
+                # Make sure the unzipped file is a directory.
+                if zipf.namelist()[0].endswith("/"):
+                    unzipped_file_name = zipf.namelist()[0]
         else:
             raise Exception("Invalid zip file {}".format(zip_file))
 
@@ -156,7 +159,7 @@ def package_download_progress(self, count, blksize, filesize):
         progress_int = int(progress)
         downloaded_kb = format(downloaded / 1024, '.2f')
 
-        # since this hook funtion is stateless, we need a state to avoid print progress repeatly
+        # Since this hook function is stateless, we need a state to avoid print progress repeatedly.
         if count == 0:
             self.prev_download_progress = 0
         if progress_int != self.prev_download_progress and progress_int % 5 == 0: