Python Asyncio排程原理詳情

# 已經移除了一些不想關的程式碼
class Handle:
    def __init__(self, callback, args, loop, context=None):
        # 初始化上下文，確保執行的時候能找到Handle所在的上下文
        if context is None:
            context = contextvars.copy_context()
        self._context = context
        self._loop = loop
        self._callback = callback
        self._args = args
        self._cancelled = False

    def cancel(self):
        # 設定當前Handle為取消狀態
        if not self._cancelled:
            self._cancelled = True
            self._callback = None
            self._args = None
    def cancelled(self):
        return self._cancelled
    def _run(self):
        # 用於執行真正的函數，且通過context.run方法來確保在自己的上下文內執行。
        try:
            # 保持在自己持有的上下文中執行對應的回撥
            self._context.run(self._callback, *self._args)
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            cb = format_helpers._format_callback_source(
                self._callback, self._args)
            msg = f'Exception in callback {cb}'
            context = {
                'message': msg,
                'exception': exc,
                'handle': self,
            }
            self._loop.call_exception_handler(context)

class TimerHandle(Handle):
    def __init__(self, when, callback, args, loop, context=None):
        super().__init__(callback, args, loop, context)
        self._when = when
        self._scheduled = False
    def __hash__(self):
        return hash(self._when)
    def __lt__(self, other):
        if isinstance(other, TimerHandle):
            return self._when < other._when
        return NotImplemented
    def __le__(self, other):
        if isinstance(other, TimerHandle):
            return self._when < other._when or self.__eq__(other)
        return NotImplemented
    def __gt__(self, other):
        if isinstance(other, TimerHandle):
            return self._when > other._when
        return NotImplemented
    def __ge__(self, other):
        if isinstance(other, TimerHandle):
            return self._when > other._when or self.__eq__(other)
        return NotImplemented
    def __eq__(self, other):
        if isinstance(other, TimerHandle):
            return (self._when == other._when and
                    self._callback == other._callback and
                    self._args == other._args and
                    self._cancelled == other._cancelled)
        return NotImplemented
    def cancel(self):
        if not self._cancelled:
            # 用於通知事件迴圈當前Handle已經退出了
            self._loop._timer_handle_cancelled(self)
        super().cancel()
    def when(self):
        return self._when

class BaseEventLoop:
    ...
    def call_soon(self, callback, *args, context=None):
        # 檢查是否事件迴圈是否關閉，如果是則直接丟擲異常
        self._check_closed()
        handle = self._call_soon(callback, args, context)
        return handle

   def _call_soon(self, callback, args, context):
        # 把呼叫封裝成一個handle，這樣方便被事件迴圈呼叫
        handle = events.Handle(callback, args, self, context)
        # 新增一個handle到_ready，等待被呼叫
        self._ready.append(handle)
        return handle

class BaseEventLoop:
    ...
    def call_later(self, delay, callback, *args, context=None):
        if delay is None:
            raise TypeError('delay must not be None')
        timer = self.call_at(self.time() + delay, callback, *args, context=context)
        return timer

    def call_at(self, when, callback, *args, context=None):
        if when is None:
            raise TypeError("when cannot be None")
        self._check_closed()
        # 建立一個timer handle，然後新增到事件迴圈的_scheduled中，等待被呼叫
        timer = events.TimerHandle(when, callback, args, self, context)
        heapq.heappush(self._scheduled, timer)
        timer._scheduled = True
        return timer

class BaseEventLoop:
    def run_until_complete(self, future):
        ...
        new_task = not futures.isfuture(future)
        # 把coroutine轉換成task，這樣事件迴圈就可以排程了，事件迴圈的最小排程單位為task
        # 需要注意的是此時事件迴圈並沒註冊到全域性變數中，所以需要顯示的傳進去，
        # 同時Task物件註冊的時候，已經通過loop.call_soon把自己註冊到事件迴圈中，等待排程
        future = tasks.ensure_future(future, loop=self)
        if new_task:
            # An exception is raised if the future didn't complete, so there
            # is no need to log the "destroy pending task" message
            future._log_destroy_pending = False
        # 當該task完成時，意味著當前事件迴圈失去了排程物件，無法繼續排程，所以需要關閉當前事件迴圈，程式會由協程模式返回到執行緒模式
        future.add_done_callback(_run_until_complete_cb)
        try:
            # 事件迴圈開始執行
            self.run_forever()
        except:
            if new_task and future.done() and not future.cancelled():
                # The coroutine raised a BaseException. Consume the exception
                # to not log a warning, the caller doesn't have access to the
                # local task.
                future.exception()
            raise
        finally:
            future.remove_done_callback(_run_until_complete_cb)
        if not future.done():
            raise RuntimeError('Event loop stopped before Future completed.')

        return future.result()

    def run_forever(self):
        # 進行一些初始化工作
        self._check_closed()
        self._check_running()
        self._set_coroutine_origin_tracking(self._debug)
        self._thread_id = threading.get_ident()

        old_agen_hooks = sys.get_asyncgen_hooks()
        # 通過asyncgen勾點來自動關閉asyncgen函數，這樣可以提醒使用者生成器還未關閉
        sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook,
                               finalizer=self._asyncgen_finalizer_hook)
        try:
            # 設定當前在執行的事件迴圈到全域性變數中，這樣就可以在任一階段獲取到當前的事件迴圈了
            events._set_running_loop(self)
            while True:
                # 正真執行任務的邏輯
                self._run_once()
                if self._stopping:
                    break
        finally:
            # 關閉迴圈, 並且清理一些資源
            self._stopping = False
            self._thread_id = None
            events._set_running_loop(None)
            self._set_coroutine_origin_tracking(False)
            sys.set_asyncgen_hooks(*old_agen_hooks)

while True:
    # 正真執行任務的邏輯
    self._run_once()
    if self._stopping:
        break

class BaseEventLoop:
    ...
    def _run_once(self):
        # self._scheduled是一個列表，它只存放TimerHandle
        sched_count = len(self._scheduled)
        ###############################
        # 第一階段，整理self._scheduled #
        ###############################
        if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
            self._timer_cancelled_count / sched_count > _MIN_CANCELLED_TIMER_HANDLES_FRACTION):
            # 當待排程的任務數量超過100且待取消的任務佔總任務的50%時，才進入這個邏輯
            # 把需要取消的任務移除
            new_scheduled = []
            for handle in self._scheduled:
                if handle._cancelled:
                    # 設定handle的_cancelled為True，並且把handle從_scheduled中移除
                    handle._scheduled = False
                else:
                    new_scheduled.append(handle)

            # 重新排列堆
            heapq.heapify(new_scheduled)
            self._scheduled = new_scheduled
            self._timer_cancelled_count = 0
        else:
            # 需要取消的handle不多，則只會走這個邏輯，這裡會把堆頂的handle彈出，並標記為不可排程，但不會存取整個堆
            while self._scheduled and self._scheduled[0]._cancelled:
                self._timer_cancelled_count -= 1
                handle = heapq.heappop(self._scheduled)
                handle._scheduled = False

        #################################
        # 第二階段，計算超時值以及等待事件IO #
        #################################
        timeout = None
        # 當有準備排程的handle或者是正在關閉時，不等待，方便儘快的排程
        if self._ready or self._stopping:
            timeout = 0
        elif self._scheduled:
            # Compute the desired timeout.
            # 如果堆有資料時，通過堆頂的handle計算最短的超時時間，但是最多不能超過MAXIMUM_SELECT_TIMEOUT，以免超過系統限制
            when = self._scheduled[0]._when
            timeout = min(max(0, when - self.time()), MAXIMUM_SELECT_TIMEOUT)

        # 事件迴圈等待事件，直到有事件或者超時
        event_list = self._selector.select(timeout)

        ##################################################
        # 第三階段，把滿足條件的TimeHandle放入到self._ready中 #
        ##################################################
        # 獲取得到的事件的回撥，然後裝填到_ready
        self._process_events(event_list)

        # 把一些在self._scheduled且滿足排程條件的handle放到_ready中，比如TimerHandle。
        # end_time為當前時間+一個時間單位，猜測是能多處理一些這段時間內產生的事件
        end_time = self.time() + self._clock_resolution
        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)

        ################################################################################
        # 第四階段，遍歷所有準備排程的handle，並且通過handle的context來執行handle對應的callback #
        ################################################################################
        ntodo = len(self._ready)
        for i in range(ntodo):
            handle = self._ready.popleft()
            # 如果handle已經被取消，則不呼叫
            if handle._cancelled:
                continue
            if self._debug:
                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)
                finally:
                    self._current_handle = None
            else:
                handle._run()
        handle = None  # Needed to break cycles when an exception occurs.

import asyncio
import threading
def task():
    print("task")
def run_loop_inside_thread(loop):
    loop.run_forever()
loop = asyncio.get_event_loop()
threading.Thread(target=run_loop_inside_thread, args=(loop,)).start()
loop.call_soon(task)

event_list = self._selector.select(timeout)

class BaseEventLoop:
    ...
    def call_soon_threadsafe(self, callback, *args, context=None):
        """Like call_soon(), but thread-safe."""
        self._check_closed()
        handle = self._call_soon(callback, args, context)
        self._write_to_self()
        return handle

class BaseSelectorEventLoop(base_events.BaseEventLoop):
    #######
    # 建立 #
    #######
    def __init__(self, selector=None):
        super().__init__()

        if selector is None:
            # 獲取最優的selector
            selector = selectors.DefaultSelector()
        self._selector = selector
        # 建立pipe
        self._make_self_pipe()
        self._transports = weakref.WeakValueDictionary()
    def _make_self_pipe(self):
        # 建立Pipe對應的sock 
        self._ssock, self._csock = socket.socketpair()
        # 設定sock為非阻塞
        self._ssock.setblocking(False)
        self._csock.setblocking(False)
        self._internal_fds += 1
        # 阻塞伺服器端sock讀事件對應的回撥
        self._add_reader(self._ssock.fileno(), self._read_from_self)
    def _add_reader(self, fd, callback, *args):
        # 檢查事件迴圈是否關閉
        self._check_closed()
        # 封裝回撥為handle物件
        handle = events.Handle(callback, args, self, None)
        try:
            key = self._selector.get_key(fd)
        except KeyError:
            # 如果沒有註冊到系統的事件迴圈，則註冊
            self._selector.register(fd, selectors.EVENT_READ,
                                    (handle, None))
        else:
            # 如果已經註冊過，則更新
            mask, (reader, writer) = key.events, key.data
            self._selector.modify(fd, mask | selectors.EVENT_READ,
                                  (handle, writer))
            if reader is not None:
                reader.cancel()
        return handle

    def _read_from_self(self):
        # 負責消費sock資料
        while True:
            try:
                data = self._ssock.recv(4096)
                if not data:
                    break
                self._process_self_data(data)
            except InterruptedError:
                continue
            except BlockingIOError:
                break
    #######
    # 刪除 #
    #######
    def _close_self_pipe(self):
        # 登出Pipe對應的描述符 
        self._remove_reader(self._ssock.fileno())
        # 關閉sock
        self._ssock.close()
        self._ssock = None
        self._csock.close()
        self._csock = None
        self._internal_fds -= 1

    def _remove_reader(self, fd):
        # 如果事件迴圈已經關閉了，就不用操作了
        if self.is_closed():
            return False
        try:
            # 查詢檔案描述符是否在selector中
            key = self._selector.get_key(fd)
        except KeyError:
            # 不存在則返回
            return False
        else:
            # 存在則進入移除的工作
            mask, (reader, writer) = key.events, key.data
            # 通過事件掩碼判斷是否有其它事件
            mask &= ~selectors.EVENT_READ
            if not mask:
                # 移除已經註冊到selector的檔案描述符
                self._selector.unregister(fd)
            else:
                # 移除已經註冊到selector的檔案描述符，並註冊新的事件
                self._selector.modify(fd, mask, (None, writer))

            # 如果reader不為空，則取消reader
            if reader is not None:
                reader.cancel()
                return True
            else:
                return False

event_list = self._selector.select(timeout)

    def _write_to_self(self):
        csock = self._csock
        if csock is None:
            return
        try:
            csock.send(b'')
        except OSError:
            if self._debug:
                logger.debug("Fail to write a null byte into the self-pipe socket", exc_info=True)

class BaseSelectorEventLoop:
    def _process_events(self, event_list):
        for key, mask in event_list:
            # 從回撥事件中獲取到對應的資料，key.data在註冊時是一個元祖，所以這裡要對元祖進行解包
            fileobj, (reader, writer) = key.fileobj, key.data
            if mask & selectors.EVENT_READ and reader is not None:
                # 得到reader handle，如果是被標記為取消，就移除對應的檔案描述符
                if reader._cancelled:
                    self._remove_reader(fileobj)
                else:
                    # 如果沒被標記為取消，則安排到self._ready中
                    self._add_callback(reader)
            if mask & selectors.EVENT_WRITE and writer is not None:
                # 對於寫物件，也是同樣的道理。
                if writer._cancelled:
                    self._remove_writer(fileobj)
                else:
                    self._add_callback(writer)

    def _add_callback(self, handle):
        # 把回撥的handle新增到_ready中
        assert isinstance(handle, events.Handle), 'A Handle is required here'
        if handle._cancelled:
            return
        assert not isinstance(handle, events.TimerHandle)
        self._ready.append(handle)

    def _remove_reader(self, fd):
        # 如果事件迴圈已經關閉了，就不用操作了
        if self.is_closed():
            return False
        try:
            # 查詢檔案描述符是否在selector中
            key = self._selector.get_key(fd)
        except KeyError:
            # 不存在則返回
            return False
        else:
            # 存在則進入移除的工作
            mask, (reader, writer) = key.events, key.data
            mask &= ~selectors.EVENT_READ
            if not mask:
                # 移除已經註冊到selector的檔案描述符
                self._selector.unregister(fd)
            else:
                self._selector.modify(fd, mask, (None, writer))

            if reader is not None:
                reader.cancel()
                return True
            else:
                return False