Given a list of asyncio tasks, return the ones that are finished with an optional timeout
(self, tasks, timeout=None)
| 631 | return task |
| 632 | |
| 633 | async def finished_tasks(self, tasks, timeout=None): |
| 634 | """ |
| 635 | Given a list of asyncio tasks, return the ones that are finished with an optional timeout |
| 636 | """ |
| 637 | if tasks: |
| 638 | try: |
| 639 | done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED, timeout=timeout) |
| 640 | return done |
| 641 | except BaseException as e: |
| 642 | if isinstance(e, (TimeoutError, asyncio.exceptions.TimeoutError)): |
| 643 | self.log.warning(f"{self.name}: Timeout after {timeout:,} seconds in finished_tasks({tasks})") |
| 644 | for task in list(tasks): |
| 645 | task.cancel() |
| 646 | self._await_cancelled_task(task) |
| 647 | else: |
| 648 | if not in_exception_chain(e, (KeyboardInterrupt, asyncio.CancelledError)): |
| 649 | self.log.error(f"{self.name}: Unhandled exception in finished_tasks({tasks}): {e}") |
| 650 | self.log.trace(traceback.format_exc()) |
| 651 | raise |
| 652 | return set() |
| 653 | |
| 654 | async def cancel_task(self, client_id): |
| 655 | parent_task = self.tasks.pop(client_id, None) |
no test coverage detected