async.py

class WithLoop:
    def __init__(self, loop: asyncio.AbstractEventLoop):
        self._loop = loop
        self._prev_loop: Optional[asyncio.AbstractEventLoop] = None

    def __enter__(self) -> Self:
        self._prev_loop = asyncio._get_running_loop()
        asyncio._set_running_loop(self._loop)
        return self

    def __exit__(self, exception_type, exception_value, exception_traceback) -> None:
        asyncio._set_running_loop(self._prev_loop)

from asyncio.base_events import _MIN_SCHEDULED_TIMER_HANDLES # type: ignore
from asyncio.base_events import _MIN_CANCELLED_TIMER_HANDLES_FRACTION # type: ignore
from asyncio.base_events import MAXIMUM_SELECT_TIMEOUT # type: ignore
import heapq

class MainEventLoop(asyncio.EventLoop):
    def __init__(self) -> None:
        super().__init__()
        self._event_list = None
        self._executor = ThreadPoolExecutor(max_workers=1)

    def _run_once_begin(self) -> None:
        # NOTE: This is the first part of BaseEventLoop.run_once.

        # NOTE: We can't set this as the running loop globally.
        #       It would clash with other event loops or sublime_aio.
        #       Instead, temporarely override it while processing the event loop

        with WithLoop(self):
            sched_count = len(self._scheduled)
            if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
                self._timer_cancelled_count / sched_count >
                    _MIN_CANCELLED_TIMER_HANDLES_FRACTION):
                new_scheduled = []
                for handle in self._scheduled:
                    if handle._cancelled:
                        handle._scheduled = False
                    else:
                        new_scheduled.append(handle)

                heapq.heapify(new_scheduled)
                self._scheduled = new_scheduled
                self._timer_cancelled_count = 0
            else:
                while self._scheduled and self._scheduled[0]._cancelled:
                    self._timer_cancelled_count -= 1
                    handle = heapq.heappop(self._scheduled)
                    handle._scheduled = False

            timeout = None
            if self._ready or self._stopping: # type: ignore
                timeout = 0
            elif self._scheduled:
                timeout = self._scheduled[0]._when - self.time()
                if timeout > MAXIMUM_SELECT_TIMEOUT:
                    timeout = MAXIMUM_SELECT_TIMEOUT
                elif timeout < 0:
                    timeout = 0

            self._select(timeout)

    def _select(self, timeout):
        def do_select():
            event_list = None
            # NOTE: Override running loop, see above
            with WithLoop(self):
                # NOTE: _selector.select(...) blocks until a coroutine gets ready to run or until the timeout expires.
                #       We run this on a separate thread.
                #       When it returns, we schedule a single iteration of the event loop to the ui thread.
                event_list = self._selector.select(timeout) # type: ignore
                def do_run_once_end():
                    self._run_once_end(event_list)
            sublime.set_timeout(do_run_once_end, 0)

        self._executor.submit(do_select)

    def _run_once_end(self, event_list) -> None:
        # NOTE: This is the final part of BaseEventLoop.run_once.

        # NOTE: Override running loop, see above
        with WithLoop(self):
            self._process_events(event_list) # type: ignore
            # Needed to break cycles when an exception occurs.
            event_list = None

            # Handle 'later' callbacks that are ready.
            end_time = self.time() + self._clock_resolution # type: ignore
            while self._scheduled:
                handle = self._scheduled[0]
                if handle._when >= end_time:
                    break
                handle = heapq.heappop(self._scheduled)
                handle._scheduled = False
                self._ready.append(handle) # type: ignore

            ntodo = len(self._ready) # type: ignore
            for i in range(ntodo):
                handle = self._ready.popleft() # type: ignore
                if handle._cancelled:
                    continue
                if self._debug: # type: ignore
                    try:
                        self._current_handle = handle
                        t0 = self.time()
                        handle._run()
                        dt = self.time() - t0
                        if dt >= self.slow_callback_duration:
                            logger.warning('Executing %s took %.3f seconds',
                                           _format_handle(handle), dt) # type: ignore
                    finally:
                        self._current_handle = None
                else:
                    handle._run()
            handle = None

        if not self._stopping: # type: ignore
            sublime.set_timeout(self._run_once_begin, 0)

    def run_forever(self):
        loop = asyncio._get_running_loop()
        self._run_forever_setup() #type: ignore
        asyncio._set_running_loop(loop)
        sublime.set_timeout(self._run_once_begin, 0)

_executor = ThreadPoolExecutor(max_workers=16, thread_name_prefix="ContinueThreadPool")

_background_loop = asyncio.new_event_loop()
_background_thread = threading.Thread(target=lambda: (_background_loop.run_forever(), _background_loop.close()))
_background_thread.start()
_background_loop.set_default_executor(_executor)

_main_loop = MainEventLoop()
_main_loop.set_default_executor(_executor)
_main_loop.run_forever()

def get_ui_loop() -> asyncio.AbstractEventLoop:
    return _main_loop

def get_loop() -> asyncio.AbstractEventLoop:
    return _background_loop