Fix usages of spawn_n and map_n in tests and readme

Author: FichteFoll

README.md

@@ -72,7 +72,7 @@ async def spawn_n_usage(todo=[range(1,51), range(51,101), range(101,200)]):
  # Returns quickly for all tasks, does not wait for pool space.
  # Workers are not spawned, they wait for pool space in their
  # own background tasks.
- fut = await pool.spawn_n(worker(i))
+ fut = pool.spawn_n(worker(i))
  futures.append(fut)
  # At this point not a single worker should start.
 
@@ -147,7 +147,7 @@ async def callbacks_usage():
 
  async with AioPool(size=2) as pool:
  for i in todo:
- fut = await pool.spawn_n(wrk(i), cb, (pool, i))
+ fut = pool.spawn_n(wrk(i), cb, (pool, i))
  futures.append(fut)
 
  results = []
@@ -173,7 +173,7 @@ async def callbacks_usage():
 
 async def exec_usage(todo=range(1,11)):
  async with AioPool(size=4) as pool:
- futures = await pool.map_n(worker, todo)
+ futures = pool.map_n(worker, todo)
 
  # While other workers are waiting or active, you can "synchronously"
  # execute one task. It does not interrupt others, just waits for pool
@@ -195,9 +195,9 @@ async def cancel_usage():
 
  pool = AioPool(size=2)
 
- f_quick = await pool.spawn_n(aio.sleep(0.1))
- f12 = await pool.spawn(wrk()), await pool.spawn_n(wrk())
- f35 = await pool.map_n(wrk, range(3))
+ f_quick = pool.spawn_n(aio.sleep(0.1))
+ f12 = await pool.spawn(wrk()), pool.spawn_n(wrk())
+ f35 = pool.map_n(wrk, range(3))
 
  # At this point, if you cancel futures, returned by pool methods,
  # you just won't be able to retrieve spawned task results, task
@@ -232,9 +232,9 @@ async def details(todo=range(1,11)):
  # This code:
  f1 = []
  for i in todo:
- f1.append(await pool.spawn_n(worker(i)))
+ f1.append(pool.spawn_n(worker(i)))
  # is equivalent to one call of `map_n`:
- f2 = await pool.map_n(worker, todo)
+ f2 = pool.map_n(worker, todo)
 
  # Afterwards you can await for any given future:
  try:

asyncio_pool/mx_asyncgen.py

@@ -37,7 +37,7 @@ async def itermap(self, fn, iterable, cb=None, ctx=None, *, flat=True,
  '''Spawns coroutines created with `fn` for each item in `iterable`, then
  waits for results with `iterwait`. See docs for `map_n` and `iterwait`.
  '''
- futures = await self.map_n(fn, iterable, cb, ctx)
+ futures = self.map_n(fn, iterable, cb, ctx)
  generator = iterwait(futures, flat=flat, timeout=timeout,
  get_result=get_result, yield_when=yield_when, loop=self.loop)
  async for batch in generator:

asyncio_pool/mx_asynciter.py

@@ -60,7 +60,7 @@ def __aiter__(_self):
 
  async def __anext__(_self):
  if not hasattr(_self, 'waiter'):
- _self.waiter = mk_waiter(await mk_map())
+ _self.waiter = mk_waiter(mk_map())
  return await _self.waiter.__anext__()
 
  return _itermap()

examples/_usage.py

@@ -24,7 +24,7 @@ async def spawn_n_usage(todo=[range(1,51), range(51,101), range(101,200)]):
  # Returns quickly for all tasks, does not wait for pool space.
  # Workers are not spawned, they wait for pool space in their
  # own background tasks.
- fut = await pool.spawn_n(worker(i))
+ fut = pool.spawn_n(worker(i))
  futures.append(fut)
  # At this point not a single worker should start.
 
@@ -99,7 +99,7 @@ async def cb(res, err, ctx): # callback
 
  async with AioPool(size=2) as pool:
  for i in todo:
- fut = await pool.spawn_n(wrk(i), cb, (pool, i))
+ fut = pool.spawn_n(wrk(i), cb, (pool, i))
  futures.append(fut)
 
  results = []
@@ -125,7 +125,7 @@ async def cb(res, err, ctx): # callback
 
 async def exec_usage(todo=range(1,11)):
  async with AioPool(size=4) as pool:
- futures = await pool.map_n(worker, todo)
+ futures = pool.map_n(worker, todo)
 
  # While other workers are waiting or active, you can "synchronously"
  # execute one task. It does not interrupt others, just waits for pool
@@ -147,9 +147,9 @@ async def wrk(*arg, **kw):
 
  pool = AioPool(size=2)
 
- f_quick = await pool.spawn_n(aio.sleep(0.1))
- f12 = await pool.spawn(wrk()), await pool.spawn_n(wrk())
- f35 = await pool.map_n(wrk, range(3))
+ f_quick = pool.spawn_n(aio.sleep(0.1))
+ f12 = await pool.spawn(wrk()), pool.spawn_n(wrk())
+ f35 = pool.map_n(wrk, range(3))
 
  # At this point, if you cancel futures, returned by pool methods,
  # you just won't be able to retrieve spawned task results, task
@@ -184,15 +184,15 @@ async def details(todo=range(1,11)):
  # This code:
  f1 = []
  for i in todo:
- f1.append(await pool.spawn_n(worker(i)))
+ f1.append(pool.spawn_n(worker(i)))
  # is equivalent to one call of `map_n`:
- f2 = await pool.map_n(worker, todo)
+ f2 = pool.map_n(worker, todo)
 
  # Afterwards you can await for any given future:
  try:
  assert 3 == await f1[2] # result of spawn_n(worker(3))
- except Exception as e:
- # exception happened in worker (or CancelledError) will be re-raised
+ except BaseException:
+ # exception happened in worker (including CancelledError) will be re-raised
  pass
 
  # Or use `asyncio.wait` to handle results in batches (see `iterwait` also):

tests/loadtest.py

@@ -25,7 +25,7 @@ async def loadtest_spawn_n(tasks, pool_size, duration):
  futures = []
  async with AioPool(size=pool_size) as pool:
  for i in range(tasks):
- fut = await pool.spawn_n(aio.sleep(duration))
+ fut = pool.spawn_n(aio.sleep(duration))
  futures.append(fut)
 
  return [getres.flat(f) for f in futures]

tests/test_base.py

@@ -70,7 +70,7 @@ async def outer(n, pool):
  loop = aio.get_event_loop()
  pool = AioPool(size=100)
 
- tasks = pool.map_n(inner, todo)
+ pool.map_n(inner, todo)
  joined = [loop.create_task(outer(j, pool)) for j in to_release]
  await pool.join()
 

tests/test_callbacks.py

@@ -31,7 +31,7 @@ async def test_spawn_n():
  async with AioPool(size=2) as pool:
  for i in todo:
  ctx = (pool, i)
- fut = await pool.spawn_n(wrk(i), cb, ctx)
+ fut = pool.spawn_n(wrk(i), cb, ctx)
  futures.append(fut)
 
  results = [getres.flat(f) for f in futures]
@@ -52,7 +52,7 @@ async def test_map():
 async def test_map_n():
  todo = range(2,11)
  async with AioPool(size=3) as pool:
- futures = await pool.map_n(wrk, todo, cb)
+ futures = pool.map_n(wrk, todo, cb)
 
  results = [getres.flat(f) for f in futures]
  assert 2 * sum(todo) == sum(results)

tests/test_spawn.py

@@ -27,7 +27,7 @@ async def wrk(n):
 
  async with AioPool(size=2) as pool:
  for i in range(1,6):
- await pool.spawn_n(wrk(i)) # does not wait for pool, just spawns waiting coros
+ pool.spawn_n(wrk(i)) # does not wait for pool, just spawns waiting coros
  assert len(started) == 0 # so atm no worker should be able to start