Task

control how much coroutines are running simultaneously with ability to stop the process from inside.

Use case

when scraping websites you want to control how much memory program consumes and how much load on target it produces.

Usage

1. create an instance


2. call attach_coroutines_generator to attach some generator which will create new coroutines from queue which is accessible from these coroutines (for example, every coroutine downloads html and queues new downloads for extracted links)


3. call run method

More description

_sleep method, _do_stop and stop_gracefully all exists to allow code inside _runner method stop run method call (because we can't cancel future from its child).

When runner catches StopIterationError it is put on sleep for delay seconds.

When some "runner" realizes that every other runner is put on sleep, it updates _do_stop attribute so all runners will break their while-loops.

Questions

• I'm not sure about appropriate naming and typing.


• Security issues?


• Is there other/better ways to do the same?

Code

import asyncio

class RunnersPool:

 def __init__(self, pool_size: int, delay: float =1., enable_auto_stop=True):
 self._pool_size = pool_size
 self._delay = delay
 self._auto_stop_enabled = enable_auto_stop

 self._do_stop = False
 self._active_counter = 0
 self._delayed_counter = 0

 self._coroutines_generator = None
 self._runners = None

 # ----------------
 # runner methods

 async def _runner(self) -> None:
 self._active_counter += 1
 while True:
 try:
 coroutine = next(self._coroutines_generator)
 except StopIteration:
 await self._sleep()
 if not self.is_running and self._auto_stop_enabled:
 break
 else:
 try:
 await coroutine
 except Exception as exc:
 self.handle_exception(exc, coroutine)
 self._active_counter -= 1

 async def _sleep(self) -> None:
 if self._delayed_counter + 1 == self._pool_size:
 self.stop_gracefully()
 return

 self._delayed_counter += 1
 await asyncio.sleep(self._delay)
 self._delayed_counter -= 1

 # end
 # ----------------

 async def run(self):
 if not self._coroutines_generator:
 raise RuntimeError('coroutines_generator not attached yet')

 self._runners = asyncio.gather(*[
 self._runner() for _ in range(self._pool_size)
 ])
 await self._runners

 def handle_exception(self, exception: Exception, coroutine: Coroutine):
 raise exception

 def cancel(self):
 self._runners.cancel()

 def stop_gracefully(self):
 self._do_stop = True

 def attach_coroutines_generator(self, coroutines_generator: Iterable[Coroutine]):
 self._coroutines_generator = coroutines_generator

 @property
 def pool_size(self) -> int:
 return self._pool_size

 @property
 def delay(self) -> float:
 return self._delay

 @property
 def is_running(self) -> bool:
 return not self._do_stop

 @property
 def auto_stop_enabled(self) -> bool:
 return self._auto_stop_enabled

 @property
 def delayed_runners(self) -> int:
 return self._delayed_counter

 @property
 def active_runners(self) -> int:
 return self._active_counter

Example usage

class SmartQueue:

 def __init__(self):
 self.queue = collections.deque()

 def put(self, x):
 self.queue.append(x)

 def get(self):
 return self.queue.popleft()

 def __next__(self):
 if self.queue:
 return self.get()
 else:
 raise StopIteration


async def main():
 sq = SmartQueue()

 async def cor(num):
 await asyncio.sleep(1)
 print(num)
 if num > 0:
 sq.put(cor(num - 1))

 for i in range(1, 10+1):
 sq.put(cor(i))

 r = RunnersPool(5)
 r.attach_coroutines_generator(sq)

 await r.run()

if __name__ == '__main__':
 loop = asyncio.new_event_loop()
 loop.run_until_complete(main())