Improving orchestration of I/O bound processes with wait_process_release.

Author: igorborgest

awswrangler/s3.py

@@ -6,7 +6,7 @@
 import s3fs
 import tenacity
 
-from awswrangler.utils import calculate_bounders
+from awswrangler.utils import calculate_bounders, wait_process_release
 
 logger = logging.getLogger(__name__)
 
@@ -92,13 +92,8 @@ def delete_objects(self, path):
                 proc.daemon = False
                 proc.start()
                 procs.append(proc)
-                while len(procs) >= self._session.procs_io_bound:
-                    logger.debug(
-                        f"len(procs) ({len(procs)}) >= self._session.procs_io_bound ({self._session.procs_io_bound})"
-                    )
-                    procs[0].join()
-                    del procs[0]
-                    logger.debug(f"Processes deleted from list.")
+                if len(procs) == self._session.procs_io_bound:
+                    wait_process_release(procs)
             else:
                 logger.debug(f"Starting last delete call...")
                 self.delete_objects_batch(self._session.primitives, bucket,
@@ -166,13 +161,8 @@ def delete_not_listed_objects(self, objects_paths, procs_io_bound=None):
             proc.daemon = False
             proc.start()
             procs.append(proc)
-            while len(procs) >= procs_io_bound:
-                logger.debug(
-                    f"len(procs) ({len(procs)}) >= procs_io_bound ({procs_io_bound})"
-                )
-                procs[0].join()
-                del procs[0]
-                logger.debug(f"Processes deleted from list.")
+            if len(procs) == self._session.procs_io_bound:
+                wait_process_release(procs)
         logger.debug(f"Waiting final processes...")
         for proc in procs:
             proc.join()
@@ -280,8 +270,8 @@ def get_objects_sizes(self, objects_paths, procs_io_bound=None):
         logger.debug(f"len(procs): {len(bounders)}")
         for i in range(len(procs)):
             logger.debug(f"Waiting pipe number: {i}")
-            receved = receive_pipes[i].recv()
-            objects_sizes.update(receved)
+            received = receive_pipes[i].recv()
+            objects_sizes.update(received)
             logger.debug(f"Waiting proc number: {i}")
             procs[i].join()
             logger.debug(f"Closing proc number: {i}")

awswrangler/utils.py

@@ -1,5 +1,5 @@
-from math import ceil
-from math import gcd
+from math import ceil, gcd
+from time import sleep
 import logging
 
 from awswrangler.exceptions import InvalidArguments
@@ -29,6 +29,24 @@ def calculate_bounders(num_items, num_groups=None, max_size=None):
         raise InvalidArguments("You must give num_groups or max_size!")
 
 
+def wait_process_release(processes):
+    """
+    Wait one of the processes releases
+    :param processes: List of processes
+    :return: None
+    """
+    n = len(processes)
+    i = 0
+    while True:
+        if not processes[i].is_alive():
+            del processes[i]
+            return None
+        i += 1
+        if i == n:
+            i = 0
+        sleep(0.1)
+
+
 def lcm(a, b):
     """
     Least Common Multiple