reworked the _map method to make use of an arguments queue; added docstrings to the TaskPool methods

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = asyncio-taskpool
-version = 0.0.2
+version = 0.1.1
 author = Daniil Fajnberg
 author_email = mail@daniil.fajnberg.de
 description = Dynamically manage pools of asyncio tasks

src/asyncio_taskpool/exceptions.py

@@ -24,3 +24,7 @@ class InvalidTaskID(PoolException):
 
 class PoolStillOpen(PoolException):
     pass
+
+
+class NotCoroutine(PoolException):
+    pass

src/asyncio_taskpool/helpers.py

@@ -0,0 +1,24 @@
+from asyncio.coroutines import iscoroutinefunction
+from typing import Any, Optional
+
+from .types import T, AnyCallableT, ArgsT, KwArgsT
+
+
+async def execute_optional(function: AnyCallableT, args: ArgsT = (), kwargs: KwArgsT = None) -> Optional[T]:
+    if not callable(function):
+        return
+    if kwargs is None:
+        kwargs = {}
+    if iscoroutinefunction(function):
+        return await function(*args, **kwargs)
+    return function(*args, **kwargs)
+
+
+def star_function(function: AnyCallableT, arg: Any, arg_stars: int = 0) -> T:
+    if arg_stars == 0:
+        return function(arg)
+    if arg_stars == 1:
+        return function(*arg)
+    if arg_stars == 2:
+        return function(**arg)
+    raise ValueError(f"Invalid argument arg_stars={arg_stars}; must be 0, 1, or 2.")

src/asyncio_taskpool/pool.py

@@ -1,13 +1,16 @@
 import logging
 from asyncio import gather
-from asyncio.coroutines import iscoroutinefunction
+from asyncio.coroutines import iscoroutine, iscoroutinefunction
 from asyncio.exceptions import CancelledError
 from asyncio.locks import Event, Semaphore
+from asyncio.queues import Queue, QueueEmpty
 from asyncio.tasks import Task, create_task
+from functools import partial
 from math import inf
-from typing import Any, Awaitable, Callable, Dict, Iterable, Iterator, List, Optional, Tuple
+from typing import Any, Awaitable, Dict, Iterable, Iterator, List
 
 from . import exceptions
+from .helpers import execute_optional, star_function
 from .types import ArgsT, KwArgsT, CoroutineFunc, EndCallbackT, CancelCallbackT
 
 
@@ -37,8 +40,7 @@ class BaseTaskPool:
         self._num_ended: int = 0
         self._idx: int = self._add_pool(self)
         self._name: str = name
-        self._all_tasks_known_flag: Event = Event()
+        self._before_gathering: List[Awaitable] = []
-        self._all_tasks_known_flag.set()
         self._interrupt_flag: Event = Event()
         log.debug("%s initialized", str(self))
 
@@ -135,7 +137,7 @@ class BaseTaskPool:
         self._cancelled[task_id] = self._running.pop(task_id)
         self._num_cancelled += 1
         log.debug("Cancelled %s", self._task_name(task_id))
-        await _execute_function(custom_callback, args=(task_id, ))
+        await execute_optional(custom_callback, args=(task_id,))
 
     async def _task_ending(self, task_id: int, custom_callback: EndCallbackT = None) -> None:
         """
@@ -157,7 +159,7 @@ class BaseTaskPool:
         self._num_ended += 1
         self._enough_room.release()
         log.info("Ended %s", self._task_name(task_id))
-        await _execute_function(custom_callback, args=(task_id, ))
+        await execute_optional(custom_callback, args=(task_id,))
 
     async def _task_wrapper(self, awaitable: Awaitable, task_id: int, end_callback: EndCallbackT = None,
                             cancel_callback: CancelCallbackT = None) -> Any:
@@ -199,14 +201,17 @@ class BaseTaskPool:
                 If `True`, even if the pool is closed, the task will still be started.
             end_callback (optional):
                 A callback to execute after the task has ended.
-                It is run with the `task_id` as its only positional argument.
+                It is run with the task's ID as its only positional argument.
             cancel_callback (optional):
                 A callback to execute after cancellation of the task.
-                It is run with the `task_id` as its only positional argument.
+                It is run with the task's ID as its only positional argument.
 
         Raises:
+            `asyncio_taskpool.exceptions.NotCoroutine` if `awaitable` is not a coroutine.
             `asyncio_taskpool.exceptions.PoolIsClosed` if the pool has been closed and `ignore_closed` is `False`.
         """
+        if not iscoroutine(awaitable):
+            raise exceptions.NotCoroutine(f"Not awaitable: {awaitable}")
         if not (self.is_open or ignore_closed):
             raise exceptions.PoolIsClosed("Cannot start new tasks")
         await self._enough_room.acquire()
@@ -325,7 +330,7 @@ class BaseTaskPool:
         """
         if self._open:
             raise exceptions.PoolStillOpen("Pool must be closed, before tasks can be gathered")
-        await self._all_tasks_known_flag.wait()
+        await gather(*self._before_gathering)
         results = await gather(*self._ended.values(), *self._cancelled.values(), *self._running.values(),
                                return_exceptions=return_exceptions)
         self._ended = self._cancelled = self._running = {}
@@ -335,66 +340,270 @@ class BaseTaskPool:
 
 
 class TaskPool(BaseTaskPool):
+    """
+    General task pool class.
+    Attempts to somewhat emulate part of the interface of `multiprocessing.pool.Pool` from the stdlib.
+
+    A `TaskPool` instance can manage an arbitrary number of concurrent tasks from any coroutine function.
+    Tasks in the pool can all belong to the same coroutine function,
+    but they can also come from any number of different and unrelated coroutine functions.
+
+    As long as there is room in the pool, more tasks can be added. (By default, there is no pool size limit.)
+    Each task started in the pool receives a unique ID, which can be used to cancel specific tasks at any moment.
+
+    Adding tasks blocks **only if** the pool is full at that moment.
+    """
+
     async def _apply_one(self, func: CoroutineFunc, args: ArgsT = (), kwargs: KwArgsT = None,
                          end_callback: EndCallbackT = None, cancel_callback: CancelCallbackT = None) -> int:
+        """
+        Creates a coroutine with the supplied arguments and runs it as a new task in the pool.
+
+        This method blocks, **only if** the pool is full.
+
+        Args:
+            func:
+                The coroutine function to be run as a task within the task pool.
+            args (optional):
+                The positional arguments to pass into the function call.
+            kwargs (optional):
+                The keyword-arguments to pass into the function call.
+            end_callback (optional):
+                A callback to execute after the task has ended.
+                It is run with the task's ID as its only positional argument.
+            cancel_callback (optional):
+                A callback to execute after cancellation of the task.
+                It is run with the task's ID as its only positional argument.
+
+        Returns:
+            The newly spawned task's ID within the pool.
+        """
         if kwargs is None:
             kwargs = {}
         return await self._start_task(func(*args, **kwargs), end_callback=end_callback, cancel_callback=cancel_callback)
 
     async def apply(self, func: CoroutineFunc, args: ArgsT = (), kwargs: KwArgsT = None, num: int = 1,
-                    end_callback: EndCallbackT = None, cancel_callback: CancelCallbackT = None) -> Tuple[int]:
+                    end_callback: EndCallbackT = None, cancel_callback: CancelCallbackT = None) -> List[int]:
-        return tuple(await self._apply_one(func, args, kwargs, end_callback, cancel_callback) for _ in range(num))
+        """
+        Creates an arbitrary number of coroutines with the supplied arguments and runs them as new tasks in the pool.
+        Each coroutine looks like `func(*args, **kwargs)`.
 
-    @staticmethod
+        This method blocks, **only if** there is not enough room in the pool for the desired number of new tasks.
-    def _get_next_coroutine(func: CoroutineFunc, args_iter: Iterator[Any], arg_stars: int = 0) -> Optional[Awaitable]:
+
+        Args:
+            func:
+                The coroutine function to use for spawning the new tasks within the task pool.
+            args (optional):
+                The positional arguments to pass into each function call.
+            kwargs (optional):
+                The keyword-arguments to pass into each function call.
+            num (optional):
+                The number of tasks to spawn with the specified parameters.
+            end_callback (optional):
+                A callback to execute after a task has ended.
+                It is run with the task's ID as its only positional argument.
+            cancel_callback (optional):
+                A callback to execute after cancellation of a task.
+                It is run with the task's ID as its only positional argument.
+
+        Returns:
+            The newly spawned tasks' IDs within the pool as a list of integers.
+
+        Raises:
+            `NotCoroutine` if `func` is not a coroutine function.
+            `PoolIsClosed` if the pool has been closed already.
+        """
+        ids = await gather(*(self._apply_one(func, args, kwargs, end_callback, cancel_callback) for _ in range(num)))
+        # TODO: for some reason PyCharm wrongly claims that `gather` returns a tuple of exceptions
+        assert isinstance(ids, list)
+        return ids
+
+    async def _queue_producer(self, q: Queue, args_iter: Iterator[Any]) -> None:
+        """
+        Keeps the arguments queue from `_map()` full as long as the iterator has elements.
+        If the `_interrupt_flag` gets set, the loop ends prematurely.
+
+        Args:
+            q:
+                The queue of function arguments to consume for starting the next task.
+            args_iter:
+                The iterator of function arguments to put into the queue.
+        """
+        for arg in args_iter:
+            if self._interrupt_flag.is_set():
+                break
+            await q.put(arg)  # This blocks as long as the queue is full.
+
+    async def _queue_consumer(self, q: Queue, func: CoroutineFunc, arg_stars: int = 0,
+                              end_callback: EndCallbackT = None, cancel_callback: CancelCallbackT = None) -> None:
+        """
+        Wrapper around the `_start_task()` taking the next element from the arguments queue set up in `_map()`.
+        Partially constructs the `_queue_callback` function with the same arguments.
+
+        Args:
+            q:
+                The queue of function arguments to consume for starting the next task.
+            func:
+                The coroutine function to use for spawning the tasks within the task pool.
+            arg_stars (optional):
+                Whether or not to unpack an element from `q` using stars; must be 0, 1, or 2.
+            end_callback (optional):
+                The actual callback specified to execute after the task (and the next one) has ended.
+                It is run with the task's ID as its only positional argument.
+            cancel_callback (optional):
+                The callback that was specified to execute after cancellation of the task (and the next one).
+                It is run with the task's ID as its only positional argument.
+        """
         try:
-            arg = next(args_iter)
+            arg = q.get_nowait()
-        except StopIteration:
+        except QueueEmpty:
             return
-        if arg_stars == 0:
+        try:
-            return func(arg)
+            await self._start_task(
-        if arg_stars == 1:
+                star_function(func, arg, arg_stars=arg_stars),
-            return func(*arg)
+                ignore_closed=True,
-        if arg_stars == 2:
+                end_callback=partial(TaskPool._queue_callback, self, q=q, func=func, arg_stars=arg_stars,
-            return func(**arg)
+                                     end_callback=end_callback, cancel_callback=cancel_callback),
-        raise ValueError
+                cancel_callback=cancel_callback
+            )
+        finally:
+            q.task_done()
+
+    async def _queue_callback(self, task_id: int, q: Queue, func: CoroutineFunc, arg_stars: int = 0,
+                              end_callback: EndCallbackT = None, cancel_callback: CancelCallbackT = None) -> None:
+        """
+        Wrapper around an end callback function passed into the `_map()` method.
+        Triggers the next `_queue_consumer` with the same arguments.
+
+        Args:
+            task_id:
+                The ID of the ending task.
+            q:
+                The queue of function arguments to consume for starting the next task.
+            func:
+                The coroutine function to use for spawning the tasks within the task pool.
+            arg_stars (optional):
+                Whether or not to unpack an element from `q` using stars; must be 0, 1, or 2.
+            end_callback (optional):
+                The actual callback specified to execute after the task (and the next one) has ended.
+                It is run with the `task_id` as its only positional argument.
+            cancel_callback (optional):
+                The callback that was specified to execute after cancellation of the task (and the next one).
+                It is run with the `task_id` as its only positional argument.
+        """
+        await self._queue_consumer(q, func, arg_stars, end_callback=end_callback, cancel_callback=cancel_callback)
+        await execute_optional(end_callback, args=(task_id,))
 
     async def _map(self, func: CoroutineFunc, args_iter: ArgsT, arg_stars: int = 0, num_tasks: int = 1,
                    end_callback: EndCallbackT = None, cancel_callback: CancelCallbackT = None) -> None:
+        """
+        Creates coroutines with arguments from a supplied iterable and runs them as new tasks in the pool in batches.
+        Each coroutine looks like `func(arg)`, `func(*arg)`, or `func(**arg)`, `arg` being an element from the iterable.
 
-        if self._all_tasks_known_flag.is_set():
+        This method blocks, **only if** there is not enough room in the pool for the first batch of new tasks.
-            self._all_tasks_known_flag.clear()
+
+        It sets up an internal queue which is filled while consuming the arguments iterable.
+        The queue's `join()` method is added to the pool's `_before_gathering` list.
+
+        Args:
+            func:
+                The coroutine function to use for spawning the new tasks within the task pool.
+            args_iter:
+                The iterable of arguments; each element is to be passed into a `func` call when spawning a new task.
+            arg_stars (optional):
+                Whether or not to unpack an element from `args_iter` using stars; must be 0, 1, or 2.
+            num_tasks (optional):
+                The maximum number of the new tasks to run concurrently.
+            end_callback (optional):
+                A callback to execute after a task has ended.
+                It is run with the task's ID as its only positional argument.
+            cancel_callback (optional):
+                A callback to execute after cancellation of a task.
+                It is run with the task's ID as its only positional argument.
+
+        Raises:
+            `asyncio_taskpool.exceptions.PoolIsClosed` if the pool has been closed.
+        """
+        if not self.is_open:
+            raise exceptions.PoolIsClosed("Cannot start new tasks")
+        args_queue = Queue(maxsize=num_tasks)
+        self._before_gathering.append(args_queue.join())
         args_iter = iter(args_iter)
-
+        try:
-        async def _start_next_coroutine() -> bool:
+            # Here we guarantee that the queue will contain as many arguments as needed for starting the first batch of
-            cor = self._get_next_coroutine(func, args_iter, arg_stars)
+            # tasks, which will be at most `num_tasks` (meaning the queue will be full).
-            if cor is None or self._interrupt_flag.is_set():
+            for i in range(num_tasks):
-                self._all_tasks_known_flag.set()
+                args_queue.put_nowait(next(args_iter))
-                return True
+        except StopIteration:
-            await self._start_task(cor, ignore_closed=True, end_callback=_start_next, cancel_callback=cancel_callback)
+            # If we get here, this means that the number of elements in the arguments iterator was less than the
-            return False
+            # specified `num_tasks`. Thus, the number of tasks to start immediately will be the size of the queue.
-
+            # The `_queue_producer` won't be necessary, since we already put all the elements in the queue.
-        async def _start_next(task_id: int) -> None:
+            num_tasks = args_queue.qsize()
-            await _start_next_coroutine()
+        else:
-            await _execute_function(end_callback, args=(task_id, ))
+            # There may be more elements in the arguments iterator, so we need the `_queue_producer`.
-
+            # It will have exclusive access to the `args_iter` from now on.
+            # If the queue is full already, it will wait until one of the tasks in the first batch ends, before putting
+            # the next item in it.
+            create_task(self._queue_producer(args_queue, args_iter))
         for _ in range(num_tasks):
-            reached_end = await _start_next_coroutine()
+            # This is where blocking can occur, if the pool is full.
-            if reached_end:
+            await self._queue_consumer(args_queue, func,
-                break
+                                       arg_stars=arg_stars, end_callback=end_callback, cancel_callback=cancel_callback)
 
-    async def map(self, func: CoroutineFunc, args_iter: ArgsT, num_tasks: int = 1,
+    async def map(self, func: CoroutineFunc, arg_iter: ArgsT, num_tasks: int = 1,
                   end_callback: EndCallbackT = None, cancel_callback: CancelCallbackT = None) -> None:
-        await self._map(func, args_iter, arg_stars=0, num_tasks=num_tasks,
+        """
+        An asyncio-task-based equivalent of the `multiprocessing.pool.Pool.map` method.
+
+        Creates coroutines with arguments from a supplied iterable and runs them as new tasks in the pool in batches.
+        Each coroutine looks like `func(arg)`, `arg` being an element from the iterable.
+
+        Once the first batch of tasks has started to run, this method returns.
+        As soon as on of them finishes, it triggers the start of a new task (assuming there is room in the pool)
+        consuming the next element from the arguments iterable.
+        If the size of the pool never imposes a limit, this ensures that there is almost continuously the desired number
+        of tasks from this call concurrently running within the pool.
+
+        This method blocks, **only if** there is not enough room in the pool for the first batch of new tasks.
+
+        Args:
+            func:
+                The coroutine function to use for spawning the new tasks within the task pool.
+            arg_iter:
+                The iterable of arguments; each argument is to be passed into a `func` call when spawning a new task.
+            num_tasks (optional):
+                The maximum number of the new tasks to run concurrently.
+            end_callback (optional):
+                A callback to execute after a task has ended.
+                It is run with the task's ID as its only positional argument.
+            cancel_callback (optional):
+                A callback to execute after cancellation of a task.
+                It is run with the task's ID as its only positional argument.
+
+        Raises:
+            `PoolIsClosed` if the pool has been closed.
+            `NotCoroutine` if `func` is not a coroutine function.
+        """
+        await self._map(func, arg_iter, arg_stars=0, num_tasks=num_tasks,
                         end_callback=end_callback, cancel_callback=cancel_callback)
 
     async def starmap(self, func: CoroutineFunc, args_iter: Iterable[ArgsT], num_tasks: int = 1,
                       end_callback: EndCallbackT = None, cancel_callback: CancelCallbackT = None) -> None:
+        """
+        Like `map()` except that the elements of `args_iter` are expected to be iterables themselves to be unpacked as
+        positional arguments to the function.
+        Each coroutine then looks like `func(*arg)`, `arg` being an element from `args_iter`.
+        """
         await self._map(func, args_iter, arg_stars=1, num_tasks=num_tasks,
                         end_callback=end_callback, cancel_callback=cancel_callback)
 
     async def doublestarmap(self, func: CoroutineFunc, kwargs_iter: Iterable[KwArgsT], num_tasks: int = 1,
                             end_callback: EndCallbackT = None, cancel_callback: CancelCallbackT = None) -> None:
+        """
+        Like `map()` except that the elements of `kwargs_iter` are expected to be iterables themselves to be unpacked as
+        keyword-arguments to the function.
+        Each coroutine then looks like `func(**arg)`, `arg` being an element from `kwargs_iter`.
+        """
         await self._map(func, kwargs_iter, arg_stars=2, num_tasks=num_tasks,
                         end_callback=end_callback, cancel_callback=cancel_callback)
 
@@ -403,6 +612,8 @@ class SimpleTaskPool(BaseTaskPool):
     def __init__(self, func: CoroutineFunc, args: ArgsT = (), kwargs: KwArgsT = None,
                  end_callback: EndCallbackT = None, cancel_callback: CancelCallbackT = None,
                  name: str = None) -> None:
+        if not iscoroutinefunction(func):
+            raise exceptions.NotCoroutine(f"Not a coroutine function: {func}")
         self._func: CoroutineFunc = func
         self._args: ArgsT = args
         self._kwargs: KwArgsT = kwargs if kwargs is not None else {}
@@ -437,13 +648,3 @@ class SimpleTaskPool(BaseTaskPool):
 
     def stop_all(self) -> List[int]:
         return self.stop(self.size)
-
-
-async def _execute_function(func: Callable, args: ArgsT = (), kwargs: KwArgsT = None) -> None:
-    if kwargs is None:
-        kwargs = {}
-    if callable(func):
-        if iscoroutinefunction(func):
-            await func(*args, **kwargs)
-        else:
-            func(*args, **kwargs)

src/asyncio_taskpool/types.py

@@ -1,10 +1,15 @@
 from asyncio.streams import StreamReader, StreamWriter
-from typing import Any, Awaitable, Callable, Iterable, Mapping, Tuple, Union
+from typing import Any, Awaitable, Callable, Iterable, Mapping, Tuple, TypeVar, Union
 
 
+T = TypeVar('T')
+
 ArgsT = Iterable[Any]
 KwArgsT = Mapping[str, Any]
+
+AnyCallableT = Callable[[...], Union[Awaitable[T], T]]
 CoroutineFunc = Callable[[...], Awaitable[Any]]
+
 EndCallbackT = Callable
 CancelCallbackT = Callable
 

tests/test_pool.py

@@ -123,9 +123,9 @@ class BaseTaskPoolTestCase(IsolatedAsyncioTestCase):
         i = 123
         self.assertEqual(f'{self.mock_str}_Task-{i}', self.task_pool._task_name(i))
 
-    @patch.object(pool, '_execute_function')
+    @patch.object(pool, 'execute_optional')
     @patch.object(pool.BaseTaskPool, '_task_name', return_value=FOO)
-    async def test__task_cancellation(self, mock__task_name: MagicMock, mock__execute_function: AsyncMock):
+    async def test__task_cancellation(self, mock__task_name: MagicMock, mock_execute_optional: AsyncMock):
         task_id, mock_task, mock_callback = 1, MagicMock(), MagicMock()
         self.task_pool._num_cancelled = cancelled = 3
         self.task_pool._running[task_id] = mock_task
@@ -134,11 +134,11 @@ class BaseTaskPoolTestCase(IsolatedAsyncioTestCase):
         self.assertEqual(mock_task, self.task_pool._cancelled[task_id])
         self.assertEqual(cancelled + 1, self.task_pool._num_cancelled)
         mock__task_name.assert_called_with(task_id)
-        mock__execute_function.assert_awaited_once_with(mock_callback, args=(task_id, ))
+        mock_execute_optional.assert_awaited_once_with(mock_callback, args=(task_id, ))
 
-    @patch.object(pool, '_execute_function')
+    @patch.object(pool, 'execute_optional')
     @patch.object(pool.BaseTaskPool, '_task_name', return_value=FOO)
-    async def test__task_ending(self, mock__task_name: MagicMock, mock__execute_function: AsyncMock):
+    async def test__task_ending(self, mock__task_name: MagicMock, mock_execute_optional: AsyncMock):
         task_id, mock_task, mock_callback = 1, MagicMock(), MagicMock()
         self.task_pool._num_ended = ended = 3
         self.task_pool._enough_room._value = room = 123
@@ -151,9 +151,9 @@ class BaseTaskPoolTestCase(IsolatedAsyncioTestCase):
         self.assertEqual(ended + 1, self.task_pool._num_ended)
         self.assertEqual(room + 1, self.task_pool._enough_room._value)
         mock__task_name.assert_called_with(task_id)
-        mock__execute_function.assert_awaited_once_with(mock_callback, args=(task_id, ))
+        mock_execute_optional.assert_awaited_once_with(mock_callback, args=(task_id, ))
         mock__task_name.reset_mock()
-        mock__execute_function.reset_mock()
+        mock_execute_optional.reset_mock()
 
         # End cancelled task:
         self.task_pool._cancelled[task_id] = self.task_pool._ended.pop(task_id)
@@ -163,7 +163,7 @@ class BaseTaskPoolTestCase(IsolatedAsyncioTestCase):
         self.assertEqual(ended + 2, self.task_pool._num_ended)
         self.assertEqual(room + 2, self.task_pool._enough_room._value)
         mock__task_name.assert_called_with(task_id)
-        mock__execute_function.assert_awaited_once_with(mock_callback, args=(task_id, ))
+        mock_execute_optional.assert_awaited_once_with(mock_callback, args=(task_id, ))
 
     @patch.object(pool.BaseTaskPool, '_task_ending')
     @patch.object(pool.BaseTaskPool, '_task_cancellation')
@@ -213,14 +213,27 @@ class BaseTaskPoolTestCase(IsolatedAsyncioTestCase):
 
         mock_create_task.return_value = mock_task = MagicMock()
         mock__task_wrapper.return_value = mock_wrapped = MagicMock()
-        mock_awaitable, mock_cancel_cb, mock_end_cb = MagicMock(), MagicMock(), MagicMock()
+        mock_coroutine, mock_cancel_cb, mock_end_cb = AsyncMock(), MagicMock(), MagicMock()
         self.task_pool._counter = count = 123
         self.task_pool._enough_room._value = room = 123
 
+        with self.assertRaises(exceptions.NotCoroutine):
+            await self.task_pool._start_task(MagicMock(), end_callback=mock_end_cb, cancel_callback=mock_cancel_cb)
+        self.assertEqual(count, self.task_pool._counter)
+        self.assertNotIn(count, self.task_pool._running)
+        self.assertEqual(room, self.task_pool._enough_room._value)
+        mock_is_open.assert_not_called()
+        mock__task_name.assert_not_called()
+        mock__task_wrapper.assert_not_called()
+        mock_create_task.assert_not_called()
+        reset_mocks()
+
         mock_is_open.return_value = ignore_closed = False
+        mock_awaitable = mock_coroutine()
         with self.assertRaises(exceptions.PoolIsClosed):
             await self.task_pool._start_task(mock_awaitable, ignore_closed,
                                              end_callback=mock_end_cb, cancel_callback=mock_cancel_cb)
+        await mock_awaitable
         self.assertEqual(count, self.task_pool._counter)
         self.assertNotIn(count, self.task_pool._running)
         self.assertEqual(room, self.task_pool._enough_room._value)
@@ -231,8 +244,10 @@ class BaseTaskPoolTestCase(IsolatedAsyncioTestCase):
         reset_mocks()
 
         ignore_closed = True
+        mock_awaitable = mock_coroutine()
         output = await self.task_pool._start_task(mock_awaitable, ignore_closed,
                                                   end_callback=mock_end_cb, cancel_callback=mock_cancel_cb)
+        await mock_awaitable
         self.assertEqual(count, output)
         self.assertEqual(count + 1, self.task_pool._counter)
         self.assertEqual(mock_task, self.task_pool._running[count])
@@ -246,11 +261,13 @@ class BaseTaskPoolTestCase(IsolatedAsyncioTestCase):
         self.task_pool._enough_room._value = room
         del self.task_pool._running[count]
 
+        mock_awaitable = mock_coroutine()
         mock_create_task.side_effect = test_exception = TestException()
         with self.assertRaises(TestException) as e:
             await self.task_pool._start_task(mock_awaitable, ignore_closed,
                                              end_callback=mock_end_cb, cancel_callback=mock_cancel_cb)
             self.assertEqual(test_exception, e)
+        await mock_awaitable
         self.assertEqual(count + 1, self.task_pool._counter)
         self.assertNotIn(count, self.task_pool._running)
         self.assertEqual(room, self.task_pool._enough_room._value)