Backup-bdg/OpenHands

1import asyncio2from concurrent import futures3from concurrent.futures import ThreadPoolExecutor4from typing import Callable, Coroutine, Iterable5 6GENERAL_TIMEOUT: int = 157EXECUTOR = ThreadPoolExecutor()8 9 10async def call_sync_from_async(fn: Callable, *args, **kwargs):11    """12    Shorthand for running a function in the default background thread pool executor13    and awaiting the result. The nature of synchronous code is that the future14    returned by this function is not cancellable15    """16    loop = asyncio.get_event_loop()17    coro = loop.run_in_executor(None, lambda: fn(*args, **kwargs))18    result = await coro19    return result20 21 22def call_async_from_sync(23    corofn: Callable, timeout: float = GENERAL_TIMEOUT, *args, **kwargs24):25    """26    Shorthand for running a coroutine in the default background thread pool executor27    and awaiting the result28    """29 30    if corofn is None:31        raise ValueError('corofn is None')32    if not asyncio.iscoroutinefunction(corofn):33        raise ValueError('corofn is not a coroutine function')34 35    async def arun():36        coro = corofn(*args, **kwargs)37        result = await coro38        return result39 40    def run():41        loop_for_thread = asyncio.new_event_loop()42        try:43            asyncio.set_event_loop(loop_for_thread)44            return asyncio.run(arun())45        finally:46            loop_for_thread.close()47 48    if getattr(EXECUTOR, '_shutdown', False):49        result = run()50        return result51 52    future = EXECUTOR.submit(run)53    futures.wait([future], timeout=timeout or None)54    result = future.result()55    return result56 57 58async def call_coro_in_bg_thread(59    corofn: Callable, timeout: float = GENERAL_TIMEOUT, *args, **kwargs60):61    """Function for running a coroutine in a background thread."""62    await call_sync_from_async(call_async_from_sync, corofn, timeout, *args, **kwargs)63 64 65async def wait_all(66    iterable: Iterable[Coroutine], timeout: int = GENERAL_TIMEOUT67) -> list:68    """69    Shorthand for waiting for all the coroutines in the iterable given in parallel. Creates70    a task for each coroutine.71    Returns a list of results in the original order. If any single task raised an exception, this is raised.72    If multiple tasks raised exceptions, an AsyncException is raised containing all exceptions.73    """74    tasks = [asyncio.create_task(c) for c in iterable]75    if not tasks:76        return []77    _, pending = await asyncio.wait(tasks, timeout=timeout)78    if pending:79        for task in pending:80            task.cancel()81        raise asyncio.TimeoutError()82    results = []83    errors = []84    for task in tasks:85        try:86            results.append(task.result())87        except Exception as e:88            errors.append(e)89    if errors:90        if len(errors) == 1:91            raise errors[0]92        raise AsyncException(errors)93    return [task.result() for task in tasks]94 95 96class AsyncException(Exception):97    def __init__(self, exceptions):98        self.exceptions = exceptions99 100    def __str__(self):101        return '\n'.join(str(e) for e in self.exceptions)102