Actually wait for tasks when cancelling

`await asyncio.sleep(0)` is almost always a bad idea, as we have the ability to actually await completion of something (or multiple things) instead of assuming timeouts. I would have used `asyncio.gather` if the `get_result` parameter didn't exist. I also inlined `_cancel` function that is only used by `cancel` and not descriptive of what it does.

Author: FichteFoll

asyncio_pool/base_pool.py

@@ -248,10 +248,19 @@ async def itermap(self, fn, iterable, cb=None, ctx=None, *, flat=True,
  '''
  raise NotImplementedError('Use one of mixins')
 
- def _cancel(self, *futures):
+ async def cancel(self, *futures, get_result=getres.flat):
+ '''Cancels spawned or waiting tasks, found by their `futures`. If no
+ `futures` are passed -- cancels all spwaned and waiting tasks.
+
+ Cancelling futures, returned by pool methods, usually won't help you
+ to cancel executing tasks, so you have to use this method.
+
+ Returns tuple of (`cancelled` count of cancelled tasks, `results`
+ collected from futures of cancelled tasks).
+ '''
  tasks, _futures = [], []
 
- if not len(futures): # meaning cancel all
+ if not futures: # meaning cancel all
  tasks.extend(self._waiting.values())
  tasks.extend(self._active.values())
  _futures.extend(self._waiting.keys())
@@ -263,21 +272,7 @@ def _cancel(self, *futures):
  tasks.append(task)
  _futures.append(fut)
 
- cancelled = sum([1 for fut in tasks if fut.cancel()])
- return cancelled, _futures
-
- async def cancel(self, *futures, get_result=getres.flat):
- '''Cancels spawned or waiting tasks, found by their `futures`. If no
- `futures` are passed -- cancels all spwaned and waiting tasks.
-
- Cancelling futures, returned by pool methods, usually won't help you
- to cancel executing tasks, so you have to use this method.
-
- Returns tuple of (`cancelled` count of cancelled tasks, `results`
- collected from futures of cancelled tasks).
- '''
- cancelled, _futures = self._cancel(*futures)
- await aio.sleep(0) # let them actually cancel
+ cancelled = sum(1 for task in tasks if task.cancel())
+ await aio.wait(tasks) # let them actually cancel
  # need to collect them anyway, to supress warnings
- results = [get_result(fut) for fut in _futures]
- return cancelled, results
+ return cancelled, [get_result(fut) for fut in _futures]