5864 fixed thread pool limit insufficiency on do _execute_convergence…

…_plan cascade parallel calls

Signed-off-by: Юрий Сагитов <[email protected]>

compose/cli/command.py

@@ -85,6 +85,12 @@ def set_parallel_limit(environment):
             raise errors.UserError('COMPOSE_PARALLEL_LIMIT can not be less than 2')
         parallel.GlobalLimit.set_global_limit(parallel_limit)
 
+        # Each of parallel execution of multiple do() functions calls
+        # parallel execution of _execute_convergence_create.
+        # Global limit should be split between such cascade calls due to avoid deadlock
+        parallel.DoLimit.set_global_limit(parallel_limit // 2)
+        parallel.ExecLimit.set_global_limit(parallel_limit // 2)
+
 
 def get_config_from_options(base_dir, options, additional_options=None):
     additional_options = additional_options or {}

compose/parallel.py

@@ -39,6 +39,34 @@ def set_global_limit(cls, value):
         cls.global_limiter = Semaphore(value)
 
 
+class DoLimit:
+    """Simple class to hold a "do" function semaphore limiter for a project. This class
+    should be treated as a singleton that is instantiated when the project is.
+    """
+
+    limiter = Semaphore(PARALLEL_LIMIT // 2)
+
+    @classmethod
+    def set_global_limit(cls, value):
+        if value is None:
+            value = PARALLEL_LIMIT // 2
+        cls.limiter = Semaphore(value)
+
+
+class ExecLimit:
+    """Simple class to hold a "exec" function semaphore limiter for a project. This class
+    should be treated as a singleton that is instantiated when the project is.
+    """
+
+    limiter = Semaphore(PARALLEL_LIMIT // 2)
+
+    @classmethod
+    def set_global_limit(cls, value):
+        if value is None:
+            value = PARALLEL_LIMIT // 2
+        cls.limiter = Semaphore(value)
+
+
 def parallel_execute_watch(events, writer, errors, results, msg, get_name, fail_check):
     """ Watch events from a parallel execution, update status and fill errors and results.
         Returns exception to re-raise.
@@ -165,6 +193,8 @@ def parallel_execute_iter(objects, func, get_deps, limit):
 
     if limit is None:
         limiter = NoLimit()
+    elif limit in [DoLimit, ExecLimit]:
+        limiter = limit.limiter
     else:
         limiter = Semaphore(limit)
 

compose/project.py

@@ -24,6 +24,7 @@
 from .network import build_networks
 from .network import get_networks
 from .network import ProjectNetworks
+from .parallel import DoLimit
 from .progress_stream import read_status
 from .service import BuildAction
 from .service import ContainerIpcMode
@@ -646,6 +647,7 @@ def get_deps(service):
             operator.attrgetter('name'),
             None,
             get_deps,
+            limit=DoLimit,
         )
         if errors:
             raise ProjectError(

compose/service.py

@@ -48,6 +48,7 @@
 from .errors import HealthCheckFailed
 from .errors import NoHealthCheckConfigured
 from .errors import OperationFailedError
+from .parallel import ExecLimit
 from .parallel import parallel_execute
 from .progress_stream import stream_output
 from .progress_stream import StreamOutputError
@@ -475,7 +476,8 @@ def get_name(service_name):
             ],
             lambda service_name: create_and_start(self, service_name.number),
             get_name,
-            "Creating"
+            "Creating",
+            limit=ExecLimit,
         )
         for error in errors.values():
             raise OperationFailedError(error)
@@ -499,6 +501,7 @@ def recreate(container):
             recreate,
             lambda c: c.name,
             "Recreating",
+            limit=ExecLimit,
         )
         for error in errors.values():
             raise OperationFailedError(error)