fix: normalized how exceptions are handled

Author: Michael Kryukov

README.md

@@ -13,6 +13,9 @@ Tasks will only be submited when there are available workers (specified by
 tasks than amount of available workers (so workers won't have to wait for
 task creation). Refer to `tests/test_queue.py` for details.
 
+Exceptions inside of the tasks craeted with executors are ignored. Refer
+to `tests/test_exceptions.py` for details.
+
 ## Example / Showcase
 
 ```py

bounded_pool/__init__.py

@@ -4,7 +4,7 @@
 from abc import ABC
 from asyncio import Semaphore, Task, create_task
 from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor
-from typing import List, Optional
+from typing import Optional
 
 
 class BoundedExecutor(ABC):
@@ -31,27 +31,24 @@ def _get_default_max_workers():
         return min(32, (os.cpu_count() or 1) + 4)
 
     async def __aenter__(self):
-        self._tasks: List[Task] = []
         self._semaphore = Semaphore(self._semaphore_size)
         return self
 
     async def __aexit__(self, exc_type, exc_value, traceback):
-        while self._tasks:
-            await self._tasks.pop()
+        # make sure no tasks being executed
+        for _ in range(self._semaphore_size):
+            await self._semaphore.acquire()
 
     async def _acquire(self):
         await self._semaphore.acquire()
 
     def _release(self, fut):
-        if fut in self._tasks:
-            self._tasks.remove(fut)
         self._semaphore.release()
 
     async def submit(self, coro, *args, **kwargs) -> Task:
         await self._acquire()
         task = create_task(coro(*args, **kwargs))
         task.add_done_callback(self._release)
-        self._tasks.append(task)
         return task
 
 

tests/test_basic.py

@@ -21,16 +21,12 @@ async def test_asyncio_executor_sleeps():
                 for _ in range(2):
                     await pool.submit(asyncio.sleep, 1)
 
-        assert not pool._tasks
-
     async with aassert_takes(more=3.5):
         async with BoundedAsyncioPoolExecutor(1) as pool:
             async with aassert_takes(more=3):
                 for _ in range(4):
                     await pool.submit(asyncio.sleep, 1)
 
-        assert not pool._tasks
-
 
 @pytest.mark.parametrize('cls', [BoundedThreadPoolExecutor, BoundedProcessPoolExecutor])
 def test_sync_executor_sleeps(cls):

tests/test_exceptions.py

@@ -0,0 +1,43 @@
+import pytest
+
+from bounded_pool import (
+    BoundedAsyncioPoolExecutor,
+    BoundedProcessPoolExecutor,
+    BoundedThreadPoolExecutor,
+)
+
+
+def raiser():
+    raise ValueError()
+
+
+async def araiser():
+    raise ValueError()
+
+
+@pytest.mark.anyio
+async def test_asyncio_executor_ignores_exceptions():
+    async with BoundedAsyncioPoolExecutor() as pool:
+        task = await pool.submit(araiser)
+
+        with pytest.raises(ValueError):
+            await task
+
+        task = await pool.submit(araiser)
+
+    with pytest.raises(ValueError):
+        await task
+
+
+@pytest.mark.parametrize('cls', [BoundedThreadPoolExecutor, BoundedProcessPoolExecutor])
+def test_sync_executor_ignores_exceptions(cls):
+    with cls() as pool:
+        task = pool.submit(raiser)
+
+        with pytest.raises(ValueError):
+            task.result()
+
+        task = pool.submit(raiser)
+
+    with pytest.raises(ValueError):
+        task.result()

tests/test_queue.py

@@ -23,7 +23,7 @@ async def test_asyncio_executor_queues():
 
 
 @pytest.mark.parametrize('cls', [BoundedThreadPoolExecutor, BoundedProcessPoolExecutor])
-def test_sync_executor_sleeps(cls):
+def test_sync_executor_queues(cls):
     with cls(1, 1) as pool:
         with assert_takes(more=1, less=2):
             with assert_takes(less=0.5):