🐛addresses blocking Thread/Process pool executors when shutting down

packages/service-library/src/servicelib/archiving_utils.py

@@ -1,10 +1,11 @@
 import asyncio
 import logging
 import zipfile
-from concurrent.futures import ProcessPoolExecutor
 from pathlib import Path
 from typing import Iterator, List, Set
 
+from servicelib.pools import non_blocking_process_pool_executor
+
 MAX_UNARCHIVING_WORKER_COUNT = 2
 
 log = logging.getLogger(__name__)
@@ -81,7 +82,7 @@ async def unarchive_dir(
     all tree leafs, which might include files or empty folders
     """
     with zipfile.ZipFile(archive_to_extract, mode="r") as zip_file_handler:
-        with ProcessPoolExecutor(max_workers=max_workers) as pool:
+        with non_blocking_process_pool_executor(max_workers=max_workers) as pool:
             loop = asyncio.get_event_loop()
 
             # running in process poll is not ideal for concurrency issues
@@ -141,7 +142,7 @@ async def archive_dir(
     dir_to_compress: Path, destination: Path, compress: bool, store_relative_path: bool
 ) -> bool:
     """Returns True if successuly archived"""
-    with ProcessPoolExecutor(max_workers=1) as pool:
+    with non_blocking_process_pool_executor(max_workers=1) as pool:
         return await asyncio.get_event_loop().run_in_executor(
             pool,
             _serial_add_to_archive,

packages/service-library/src/servicelib/pools.py

@@ -0,0 +1,15 @@
+from concurrent.futures import ProcessPoolExecutor
+from contextlib import contextmanager
+
+
+@contextmanager
+def non_blocking_process_pool_executor(**kwargs) -> ProcessPoolExecutor:
+    """
+    Avoids default context manger behavior which calls
+    shutdown with wait=True an blocks.
+    """
+    pool = ProcessPoolExecutor(**kwargs)  # pylint: disable=consider-using-with
+    try:
+        yield pool
+    finally:
+        pool.shutdown(wait=False)

packages/service-library/src/servicelib/utils.py

@@ -7,6 +7,7 @@
 import asyncio
 import logging
 import os
+
 from pathlib import Path
 from typing import Any, Awaitable, Coroutine, List, Optional, Union
 

services/catalog/src/simcore_service_catalog/api/routes/services.py

@@ -3,7 +3,6 @@
 import asyncio
 import logging
 import urllib.parse
-from concurrent.futures.process import ProcessPoolExecutor
 from typing import Any, Dict, List, Optional, Set, Tuple
 
 from fastapi import APIRouter, Depends, Header, HTTPException, status
@@ -27,6 +26,7 @@
     list_frontend_services,
 )
 from ...utils.requests_decorators import cancellable_request
+from ...utils.pools import non_blocking_process_pool_executor
 from ..dependencies.database import get_repository
 from ..dependencies.director import DirectorApi, get_director_api
 
@@ -155,7 +155,7 @@ async def list_services(
     # 3. then we compose the final service using as a base the registry service, overriding with the same
     #    service from the database, adding also the access rights and the owner as email address instead of gid
     # NOTE: this final step runs in a process pool so that it runs asynchronously and does not block in any way
-    with ProcessPoolExecutor(max_workers=2) as pool:
+    with non_blocking_process_pool_executor(max_workers=2) as pool:
         services_details = await asyncio.gather(
             *[
                 asyncio.get_event_loop().run_in_executor(

services/catalog/src/simcore_service_catalog/utils/pools.py

@@ -0,0 +1,17 @@
+from contextlib import contextmanager
+from concurrent.futures import ProcessPoolExecutor
+
+
+# because there is no shared fastapi library, this is a
+# duplicate of servicelib.pools.non_blocking_process_pool_executor
+@contextmanager
+def non_blocking_process_pool_executor(**kwargs) -> ProcessPoolExecutor:
+    """
+    Avoids default context manger behavior which calls
+    shutdown with wait=True an blocks.
+    """
+    pool = ProcessPoolExecutor(**kwargs)  # pylint: disable=consider-using-with
+    try:
+        yield pool
+    finally:
+        pool.shutdown(wait=False)

services/web/server/src/simcore_service_webserver/diagnostics_monitoring.py

@@ -1,7 +1,6 @@
 """ Enables monitoring of some quantities needed for diagnostics
 
 """
-import concurrent.futures
 import logging
 import time
 from typing import Callable, Coroutine
@@ -32,14 +31,13 @@ def get_collector_registry(app: web.Application) -> CollectorRegistry:
 async def metrics_handler(request: web.Request):
     registry = get_collector_registry(request.app)
 
-    with concurrent.futures.ThreadPoolExecutor() as pool:
-        # NOTE: Cannot use ProcessPoolExecutor because registry is not pickable
-        result = await request.loop.run_in_executor(
-            pool, prometheus_client.generate_latest, registry
-        )
-        response = web.Response(body=result)
-        response.content_type = CONTENT_TYPE_LATEST
-        return response
+    # NOTE: Cannot use ProcessPoolExecutor because registry is not pickable
+    result = await request.loop.run_in_executor(
+        None, prometheus_client.generate_latest, registry
+    )
+    response = web.Response(body=result)
+    response.content_type = CONTENT_TYPE_LATEST
+    return response
 
 
 def middleware_factory(app_name: str) -> Coroutine:

services/web/server/src/simcore_service_webserver/exporter/formatters/formatter_v2.py

@@ -2,14 +2,14 @@
 import logging
 from pathlib import Path
 from typing import Optional
-from concurrent.futures import ProcessPoolExecutor
 from collections import deque
 
 from aiohttp import web
 from aiopg.sa.result import ResultProxy, RowProxy
 from aiopg.sa.engine import SAConnection
 
 from simcore_postgres_database.models.scicrunch_resources import scicrunch_resources
+from servicelib.pools import non_blocking_process_pool_executor
 
 from ..exceptions import ExporterException
 from .formatter_v1 import FormatterV1
@@ -168,7 +168,7 @@ async def _write_sds_content(
     )
 
     # writing SDS structure with process pool to avoid blocking
-    with ProcessPoolExecutor(max_workers=1) as pool:
+    with non_blocking_process_pool_executor(max_workers=1) as pool:
         return await asyncio.get_event_loop().run_in_executor(
             pool,
             write_sds_directory_content,

services/web/server/src/simcore_service_webserver/projects/projects_db.py

@@ -5,7 +5,6 @@
 
 """
 import asyncio
-import concurrent.futures
 import logging
 import textwrap
 import uuid as uuidlib
@@ -397,10 +396,9 @@ async def __load_projects(
                 continue
             try:
 
-                with concurrent.futures.ThreadPoolExecutor() as pool:
-                    await asyncio.get_event_loop().run_in_executor(
-                        pool, ProjectAtDB.from_orm, row
-                    )
+                await asyncio.get_event_loop().run_in_executor(
+                    None, ProjectAtDB.from_orm, row
+                )
             except ValidationError as exc:
                 log.warning(
                     "project in db with uuid [%s] failed validation, please check. error: %s",