(self)
Pick up work, execute work, return results, rinse, repeat.
Pick up work, execute work, return results, rinse, repeat.
Pick up work execute work return results rinse repeat .
async def run(self) -> None:
        """Pick up work, execute work, return results, rinse, repeat."""
        pending: Dict[asyncio.Future, TaskID] = {}
        completed = 0
        running = True
        while running or pending:
            # TTL, Tasks To Live, determines how many tasks to execute before dying
            if self.ttl and completed >= self.ttl:
                running = False

            # pick up new work as long as we're "running" and we have open slots
            while running and len(pending) < self.concurrency:
                try:
                    task: PoolTask = self.tx.get_nowait()
                except queue.Empty:
                    break

                if task is None:
                    running = False
                    break

                tid, func, args, kwargs = task
                future = asyncio.ensure_future(func(*args, **kwargs))
                pending[future] = tid

            if not pending:
                await asyncio.sleep(0.005)
                continue

            # return results and/or exceptions when completed
            done, _ = await asyncio.wait(
                pending.keys(), timeout=0.05, return_when=asyncio.FIRST_COMPLETED
            )
            for future in done:
                tid = pending.pop(future)

                result = None
                tb = None
                try:
                    result = future.result()
                except BaseException as e:
                    if self.exception_handler is not None:
                        self.exception_handler(e)

                    tb = traceback.format_exc()

                self.rx.put_nowait((tid, result, tb))
                completed += 1
async def run ( self ) -> None : pending : Dict [ asyncio . Future , TaskID ] = { } completed = 0 running = True while running or pending : # TTL, Tasks To Live, determines how many tasks to execute before dying if self . ttl and completed >= self . ttl : running = False # pick up new work as long as we're "running" and we have open slots while running and len ( pending ) < self . concurrency : try : task : PoolTask = self . tx . get_nowait ( ) except queue . Empty : break if task is None : running = False break tid , func , args , kwargs = task future = asyncio . ensure_future ( func ( * args , * * kwargs ) ) pending [ future ] = tid if not pending : await asyncio . sleep ( 0.005 ) continue # return results and/or exceptions when completed done , _ = await asyncio . wait ( pending . keys ( ) , timeout = 0.05 , return_when = asyncio . FIRST_COMPLETED ) for future in done : tid = pending . pop ( future ) result = None tb = None try : result = future . result ( ) except BaseException as e : if self . exception_handler is not None : self . exception_handler ( e ) tb = traceback . format_exc ( ) self . rx . put_nowait ( ( tid , result , tb ) ) completed += 1