map uses spawn under the hood, instead of spawn_n; loadtest added

Author: gistart

README.md

@@ -4,7 +4,14 @@
 
 
 ## Usage
-(more in `tests/` and `examples/` dirs): # TODO
+
+`spawn` and `map` methods is probably what you should use in 99% of cases. Their overhead is minimal (~3% execution time), and even in worst cases memory usage is insignificant.
+
+`spawn_n`, `map_n` and `itermap` methods give you more control and flexibily, but they come with a price of higher overhead. They spawn all tasks that you want, and most of the tasks wait their turn "in background". If you spawn too much (10**6+ tasks) -- you'll use most of the memory you have in system, also you'll lose a lot of time on "concurrency management" of all the tasks spawned.
+
+Play with `python tests/loadtest.py -h` to understand what you want to use.
+
+#### Usage examples (more in `tests/` and `examples/` dirs):
 
 ```python
 async def worker(n): # dummy worker
@@ -176,7 +183,7 @@ async def details(todo=range(1,11)):
 - [x] setup
 - [x] cancelled, timeouts
 - [x] tests
-- [ ] loadtests
+- [x] loadtests
 - [x] usage
 - [ ] docs
 - [ ] examples

asyncio_pool/base_pool.py

@@ -128,7 +128,11 @@ async def map_n(self, fn, iterable):
  return futures
 
  async def map(self, fn, iterable, exc_as_result=True):
- futures = await self.map_n(fn, iterable)
+ futures = []
+ for it in iterable:
+ fut = await self.spawn(fn(it))
+ futures.append(fut)
+
  await aio.wait(futures)
  return [result_noraise(fut, exc_as_result) for fut in futures]
 

tests/loadtest.py

@@ -0,0 +1,95 @@
+# coding: utf8
+
+import os
+import sys
+curr_dir = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.split(curr_dir)[0])
+
+import time
+import argparse
+import asyncio as aio
+from asyncio_pool import AioPool, result_noraise
+
+
+async def loadtest_spawn(tasks, pool_size, duration):
+ futures = []
+ async with AioPool(size=pool_size) as pool:
+ for i in range(tasks):
+ fut = await pool.spawn(aio.sleep(duration))
+ futures.append(fut)
+
+ return [result_noraise(f) for f in futures]
+
+
+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))
+ futures.append(fut)
+
+ return [result_noraise(f) for f in futures]
+
+
+async def loadtest_map(tasks, pool_size, duration):
+ async def wrk(i):
+ await aio.sleep(duration)
+
+ async with AioPool(size=pool_size) as pool:
+ return await pool.map(wrk, range(tasks))
+
+
+async def loadtest_itermap(tasks, pool_size, duration):
+ async def wrk(i):
+ await aio.sleep(duration)
+
+ results = []
+ async with AioPool(size=pool_size) as pool:
+ async for res in pool.itermap(wrk, range(tasks)):
+ results.append(res)
+
+ return results
+
+
+def print_stats(args, exec_time):
+ ideal = args.task_duration * (args.tasks / args.pool_size)
+
+ overhead = exec_time - ideal
+ overhead_perc = ((exec_time / ideal) - 1) * 100
+
+ per_task = overhead / args.tasks
+ per_task_perc = (((args.task_duration + per_task) / args.task_duration) - 1) * 100
+
+ print(f'{ideal:15.5f}s -- ideal result')
+ print(f'{exec_time:15.5f}s -- were executing')
+ print(f'{overhead:15.5f}s -- overhead total')
+ print(f'{overhead_perc:15.5f}% -- overhead total percent')
+ print(f'{per_task:15.5f}s -- overhead per task')
+ print(f'{per_task_perc:15.5f}% -- overhead per task percent')
+
+
+if __name__ == "__main__":
+ methods = {
+ 'spawn': loadtest_spawn,
+ 'spawn_n': loadtest_spawn_n,
+ 'map': loadtest_map,
+ 'itermap': loadtest_itermap,
+ }
+
+ p = argparse.ArgumentParser()
+ p.add_argument('method', choices=methods.keys())
+ p.add_argument('--tasks', '-t', type=int, default=10**5)
+ p.add_argument('--task-duration', '-d', type=float, default=0.2)
+ p.add_argument('--pool-size', '-p', type=int, default=10**3)
+ args = p.parse_args()
+
+ print('>>> Running %d tasks in pool of size=%s, each task takes %.3f sec.' %
+ (args.tasks, args.pool_size, args.task_duration))
+ print('>>> This will run more than %.5f seconds' %
+ (args.task_duration * (args.tasks / args.pool_size)))
+
+ ts_start = time.perf_counter()
+ m = methods.get(args.method)(args.tasks, args.pool_size, args.task_duration)
+ aio.get_event_loop().run_until_complete(m)
+ exec_time = time.perf_counter() - ts_start
+ print_stats(args, exec_time)

tests/test_base.py

@@ -1,5 +1,6 @@
 # coding: utf8
 
+import sys
 import pytest
 import warnings
 import asyncio as aio
@@ -70,17 +71,17 @@ async def outer(n, pool):
  joined = [loop.create_task(outer(j, pool)) for j in to_release]
  await pool.join()
 
- assert len(released) < len(to_release)
+ assert len(released) <= len(to_release)
  await aio.wait(joined)
  assert len(todo) == len(done) and len(released) == len(to_release)
 
 
-@pytest.mark.filterwarnings('ignore:coroutines') # doesn't work??
+# @pytest.mark.filterwarnings('ignore:coroutines') # doesn't work??
+@pytest.mark.skipif((3,6) <= sys.version_info < (3,7) , reason='no 3.6')
 @pytest.mark.asyncio
 async def test_cancel():
  # seems to break other tests in different files in py36 only
  # TODO investigate
- return True
 
  async def wrk(*arg, **kw):
  await aio.sleep(0.5)