diff --git a/services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/base_scheduler.py b/services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/base_scheduler.py index 523d2443183..3ba703a78b7 100644 --- a/services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/base_scheduler.py +++ b/services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/base_scheduler.py @@ -30,7 +30,7 @@ from pydantic import PositiveInt from servicelib.common_headers import UNDEFINED_DEFAULT_SIMCORE_USER_AGENT_VALUE from servicelib.rabbitmq import RabbitMQClient, RabbitMQRPCClient -from servicelib.utils import logged_gather +from servicelib.utils import limited_gather from ...constants import UNDEFINED_STR_METADATA from ...core.errors import ( @@ -220,7 +220,7 @@ async def stop_pipeline( async def schedule_all_pipelines(self) -> None: self.wake_up_event.clear() # if one of the task throws, the other are NOT cancelled which is what we want - await logged_gather( + await limited_gather( *( self._schedule_pipeline( user_id=user_id, @@ -234,8 +234,10 @@ async def schedule_all_pipelines(self) -> None: iteration, ), pipeline_params in self.scheduled_pipelines.items() ), + reraise=False, log=_logger, - max_concurrency=40, + limit=40, + tasks_group_prefix="computational-scheduled-pipeline", ) async def _get_pipeline_dag(self, project_id: ProjectID) -> nx.DiGraph: