jc211
/
tractor
forked from goodboy/tractor
1
0
Fork 0
Code Pull Requests Activity

More comments

eg_worker_poolz
Tyler Goodlet 2021-01-17 15:41:35 -05:00
parent 9b07e9ad7c
commit 57a24cdcf8
1 changed files with 9 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
examples/parallelism/concurrent_actors_primes.py
View File

@ -4,8 +4,8 @@ Demonstration of the prime number detector example from the
https://docs.python.org/3/library/concurrent.futures.html#processpoolexecutor-example https://docs.python.org/3/library/concurrent.futures.html#processpoolexecutor-example
This uses no extra threads or fancy semaphores besides ``tractor``'s This uses no extra threads, fancy semaphores or futures; all we need
(TCP) channels. is ``tractor``'s channels.
""" """
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
@ -46,11 +46,12 @@ def is_prime(n):
@asynccontextmanager @asynccontextmanager
async def worker_pool(workers=4): async def worker_pool(workers=4):
"""Though it's a trivial special case for ``tractor``, the well """Though it's a trivial special case for ``tractor``, the well
known "worker pool" seems to be the defacto "I want this process known "worker pool" seems to be the defacto "but, I want this
pattern" for most parallelism pilgrims. process pattern!" for most parallelism pilgrims.
Yes, the workers stay alive (and ready for work) until you close
the context.
""" """
async with tractor.open_nursery() as tn: async with tractor.open_nursery() as tn:
portals = [] portals = []
@ -67,7 +68,7 @@ async def worker_pool(workers=4):
) )
) )
async def map( async def _map(
worker_func: Callable[[int], bool], worker_func: Callable[[int], bool],
sequence: List[int] sequence: List[int]
) -> List[bool]: ) -> List[bool]:
@ -90,7 +91,8 @@ async def worker_pool(workers=4):
for _ in range(len(sequence)): for _ in range(len(sequence)):
yield await recv_chan.receive() yield await recv_chan.receive()
yield map # deliver the parallel "worker mapper" to user code
yield _map
# tear down all "workers" on pool close # tear down all "workers" on pool close
await tn.cancel() await tn.cancel()
@ -110,7 +112,6 @@ async def main():
if __name__ == '__main__': if __name__ == '__main__':
start = time.time() start = time.time()
trio.run(main) trio.run(main)
print(f'script took {time.time() - start} seconds') print(f'script took {time.time() - start} seconds')