itermap for python3.5

Author: gistart

README.md

@@ -8,39 +8,39 @@ AioPool makes sure _no more_ and _no less_ (if possible) than `size` spawned cor
 
 Read [code doctrings](../master/asyncio_pool/base_pool.py) for details.
 
-##### AioPool(size=4, *, loop=None)
+> `AioPool(size=4, *, loop=None)`
 
 Creates pool of `size` concurrent tasks. Supports async context manager interface.
 
-##### spawn(coro, cb=None, ctx=None)
+> `spawn(coro, cb=None, ctx=None)`
 
 Waits for pool space, then creates task for `coro` coroutine, returning future for it's result. Can spawn coroutine, created by `cb` with result of `coro` as first argument. `ctx` context is passed to callback as third positinal argument.
 
-##### exec(coro, cb=None, ctx=None)
+> `exec(coro, cb=None, ctx=None)`
 
 Waits for pool space, then creates task for `coro`, then waits for it to finish, then returns result of `coro` if no callback is provided, otherwise creates task for callback, waits for it and returns result of callback.
 
-##### spawn_n(coro, cb=None, ctx=None)
+> `spawn_n(coro, cb=None, ctx=None)`
 
 Creates waiting task for `coro`, returns future without waiting for pool space. Task is executed "in pool" when pool space is available.
 
-##### join()
+> `join()`
 
 Waits for all spawned (active and waiting) tasks to finish. Joining pool from coroutine, spawned by the same pool leads to *deadlock*.
 
-##### cancel(*futures)
+> `cancel(*futures)`
 
 Cancels spawned tasks (active and waiting), finding them by provided `futures`. If no futures provided -- cancels all spawned tasks.
 
-##### map(fn, iterable, cb=None, ctx=None, *, exc_as_result=True)
+> `map(fn, iterable, cb=None, ctx=None, *, exc_as_result=True)`
 
 Spawns coroutines created by `fn` function for each item in `iterable` with `spawn`, waits for all of them to finish (including callbacks), returns results maintaining order of `iterable`.
 
-##### map_n(fn, iterable, cb=None, ctx=None, *, exc_as_result=True)
+> `map_n(fn, iterable, cb=None, ctx=None, *, exc_as_result=True)`
 
 Spawns coroutines created by `fn` function for each item in `iterable` with `spawn_n`, returns futures for task results maintaining order of `iterable`.
 
-##### itermap(fn, iterable, cb=None, ctx=None, *, flat=True, exc_as_result=True, timeout=None, yield_when=asyncio.ALL_COMPLETED)
+> `itermap(fn, iterable, cb=None, ctx=None, *, flat=True, exc_as_result=True, timeout=None, yield_when=asyncio.ALL_COMPLETED)`
 
 Spawns tasks with `map_n(fn, iterable, cb, ctx)`, then waits for results with `asyncio.wait` function, yielding ready results one by one if `flat` == True, otherwise yielding list of ready results.
 

asyncio_pool/__init__.py

@@ -8,7 +8,9 @@
 
 if sys.version_info < (3, 6):  # this means 3.5  # TODO test 3.4?
 
-    class AioPool(BaseAioPool): pass
+    from .mx_asynciter import MxAsyncIterPool, iterwait
+
+    class AioPool(MxAsyncIterPool, BaseAioPool): pass
 
 else:
     from .mx_asyncgen import MxAsyncGenPool, iterwait

asyncio_pool/base_pool.py

@@ -7,7 +7,6 @@
 
 
 class BaseAioPool(object):
-    # python3.5 friendly
 
     def __init__(self, size=1024, *, loop=None):
         '''Pool of asyncio coroutines with familiar interface.
@@ -231,11 +230,13 @@ async def map(self, fn, iterable, cb=None, ctx=None, *, exc_as_result=True):
         await aio.wait(futures)
         return [result_noraise(fut, exc_as_result) for fut in futures]
 
-    async def iterwait(self, *arg, **kw):  # TODO there's a way to support 3.5?
-        raise NotImplementedError('python3.6+ required')
-
-    async def itermap(self, *arg, **kw):  # TODO there's a way to support 3.5?
-        raise NotImplementedError('python3.6+ required')
+    async def itermap(self, fn, iterable, cb=None, ctx=None, *, flat=True,
+            exc_as_result=True, timeout=None, yield_when=aio.ALL_COMPLETED):
+        '''Spawns coroutines created with `fn` for each item in `iterable`, then
+        waits for results with `iterwait` (implementation specific). See docs
+        for `map_n` and `iterwait` (in mixins for py3.5 and py3.6+).
+        '''
+        raise NotImplementedError('Use one of mixins')
 
     def _cancel(self, *futures):
         tasks, _futures = [], []

asyncio_pool/mx_asyncgen.py

@@ -31,9 +31,8 @@ async def iterwait(futures, *, flat=True, exc_as_result=True,
 class MxAsyncGenPool(object):
     # Asynchronous generator wrapper for asyncio.wait.
 
-    async def itermap(self, fn, iterable, cb=None, ctx=None, *,
-            flat=True, exc_as_result=True, timeout=None,
-            yield_when=aio.ALL_COMPLETED):
+    async def itermap(self, fn, iterable, cb=None, ctx=None, *, flat=True,
+            exc_as_result=True, timeout=None, yield_when=aio.ALL_COMPLETED):
         '''Spawns coroutines created with `fn` for each item in `iterable`, then
         waits for results with `iterwait`. See docs for `map_n` and `iterwait`.
         '''

asyncio_pool/mx_asynciter.py

@@ -0,0 +1,65 @@
+# coding: utf8
+'''Mixin for BaseAioPool with async generator _simulation_ for python 3.5'''
+
+import asyncio as aio
+from collections import deque
+from functools import partial
+from .utils import result_noraise
+
+
+class iterwait:
+
+    def __init__(self, futures, *, flat=True, exc_as_result=True,
+            timeout=None, yield_when=aio.ALL_COMPLETED, loop=None):
+
+        self.results = deque()
+        self.flat = flat
+        self._futures = futures
+        self._extract = partial(result_noraise, exc_as_result=exc_as_result)
+        self._wait = partial(aio.wait, timeout=timeout, loop=loop,
+                             return_when=yield_when)
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if not (self._futures or self.results):
+            raise StopAsyncIteration()
+        while not self.results:
+            await self._wait_next()
+        return self.results.popleft()
+
+    async def _wait_next(self):
+        while True:
+            done, self._futures = await self._wait(self._futures)
+            if done:
+                batch = [self._extract(fut) for fut in done]
+                if self.flat:
+                    self.results.extend(batch)
+                else:
+                    self.results.append(batch)
+                break
+
+
+class MxAsyncIterPool(object):
+
+    def itermap(self, fn, iterable, cb=None, ctx=None, *, flat=True,
+            exc_as_result=True, timeout=None, yield_when=aio.ALL_COMPLETED):
+        '''Spawns coroutines created with `fn` for each item in `iterable`, then
+        waits for results with `iterwait`. See docs for `map_n` and `iterwait`.
+        '''
+        mk_map = partial(self.map_n, fn, iterable, cb=cb, ctx=ctx)
+        mk_waiter = partial(iterwait, flat=flat, loop=self.loop,
+                            exc_as_result=exc_as_result, timeout=timeout,
+                            yield_when=yield_when)
+
+        class _itermap:
+            def __aiter__(_self):
+                return _self
+
+            async def __anext__(_self):
+                if not hasattr(_self, 'waiter'):
+                    _self.waiter = mk_waiter(await mk_map())
+                return await _self.waiter.__anext__()
+
+        return _itermap()

asyncio_pool/utils.py

@@ -1,10 +1,10 @@
 # coding: utf8
 
-import asyncio as aio
-
 
 def result_noraise(future, exc_as_result=True):
     try:
-        return future.result()
+        res = future.result()
+        return res if exc_as_result else (res, None)
     except Exception as exc:
-        return exc if exc_as_result else None
+        # TODO traceback ??
+        return exc if exc_as_result else (None, exc)

docs/_readme_template.md

@@ -8,39 +8,39 @@ AioPool makes sure _no more_ and _no less_ (if possible) than `size` spawned cor
 
 Read [code doctrings](../master/asyncio_pool/base_pool.py) for details.
 
-##### AioPool(size=4, *, loop=None)
+> `AioPool(size=4, *, loop=None)`
 
 Creates pool of `size` concurrent tasks. Supports async context manager interface.
 
-##### spawn(coro, cb=None, ctx=None)
+> `spawn(coro, cb=None, ctx=None)`
 
 Waits for pool space, then creates task for `coro` coroutine, returning future for it's result. Can spawn coroutine, created by `cb` with result of `coro` as first argument. `ctx` context is passed to callback as third positinal argument.
 
-##### exec(coro, cb=None, ctx=None)
+> `exec(coro, cb=None, ctx=None)`
 
 Waits for pool space, then creates task for `coro`, then waits for it to finish, then returns result of `coro` if no callback is provided, otherwise creates task for callback, waits for it and returns result of callback.
 
-##### spawn_n(coro, cb=None, ctx=None)
+> `spawn_n(coro, cb=None, ctx=None)`
 
 Creates waiting task for `coro`, returns future without waiting for pool space. Task is executed "in pool" when pool space is available.
 
-##### join()
+> `join()`
 
 Waits for all spawned (active and waiting) tasks to finish. Joining pool from coroutine, spawned by the same pool leads to *deadlock*.
 
-##### cancel(*futures)
+> `cancel(*futures)`
 
 Cancels spawned tasks (active and waiting), finding them by provided `futures`. If no futures provided -- cancels all spawned tasks.
 
-##### map(fn, iterable, cb=None, ctx=None, *, exc_as_result=True)
+> `map(fn, iterable, cb=None, ctx=None, *, exc_as_result=True)`
 
 Spawns coroutines created by `fn` function for each item in `iterable` with `spawn`, waits for all of them to finish (including callbacks), returns results maintaining order of `iterable`.
 
-##### map_n(fn, iterable, cb=None, ctx=None, *, exc_as_result=True)
+> `map_n(fn, iterable, cb=None, ctx=None, *, exc_as_result=True)`
 
 Spawns coroutines created by `fn` function for each item in `iterable` with `spawn_n`, returns futures for task results maintaining order of `iterable`.
 
-##### itermap(fn, iterable, cb=None, ctx=None, *, flat=True, exc_as_result=True, timeout=None, yield_when=asyncio.ALL_COMPLETED)
+> `itermap(fn, iterable, cb=None, ctx=None, *, flat=True, exc_as_result=True, timeout=None, yield_when=asyncio.ALL_COMPLETED)`
 
 Spawns tasks with `map_n(fn, iterable, cb, ctx)`, then waits for results with `asyncio.wait` function, yielding ready results one by one if `flat` == True, otherwise yielding list of ready results.
 

examples/_usage.py

@@ -66,7 +66,6 @@ async def map_usage(todo=range(100)):
 
 
 async def itermap_usage(todo=range(1,11)):
-    # Python 3.6+
     result = 0
     async with AioPool(size=10) as pool:
         # Combines spawn_n and iterwait, which is a wrapper for asyncio.wait,

setup.py

@@ -3,8 +3,14 @@
 import setuptools
 
 
+github_url = 'https://github.com/gistart/asyncio-pool'
+
+readme_lines = []
 with open('README.md') as fd:
-    readme = fd.read()
+    readme_lines = filter(None, fd.read().splitlines())
+readme_lines = list(readme_lines)[:3]
+readme_lines.append('Read more at [github page](%s).' % github_url)
+readme = '\n\n'.join(readme_lines)
 
 
 setuptools.setup(
@@ -15,7 +21,7 @@
     description='Pool of asyncio coroutines with familiar interface',
     long_description=readme,
     long_description_content_type='text/markdown',
-    url='https://github.com/gistart/asyncio-pool',
+    url=github_url,
     license='MIT',
     packages=['asyncio_pool'],
     # install_requires=['asyncio'],  # where " openstack/deb-python-trollius asyncio" comes from???

tests/loadtest.py

@@ -53,19 +53,15 @@ async def wrk(i):
 
 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
+    overhead_perc = ((exec_time / ideal) - 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'{exec_time:15.5f}s -- total executing time')
+    print(f'{overhead:15.5f}s -- total overhead')
     print(f'{per_task:15.5f}s -- overhead per task')
-    print(f'{per_task_perc:15.5f}% -- overhead per task percent')
+    print(f'{overhead_perc:13.3f}%   -- overhead total percent')
 
 
 if __name__ == "__main__":