klippy — reactor模块

该模块负责管理事件的注册、调度和处理，充当事件驱动的核心引擎，驱动整个klippy系统的运行。
该模块提供了一个统一的接口register_callback，使各个模块能够注册自己的回调函数以响应特定的事件。
使用事件循环的方式，不断地检查事件的状态并触发相应的回调函数。

reactor模式

Reactor 模式是一个事件驱动的编程模式，它允许程序以非阻塞的方式处理多个 I/O 操作。这个模式主要包含四个核心组件：

事件循环（Event Loop）
：它负责不断监听事件，并将其分发给相应的处理器。
反应堆（Reactor）
：作为事件循环的管理者，它监视一组资源，等待事件发生。
资源（Resources）
：通常是网络套接字或文件描述符，是反应堆监视的对象。
事件处理器（Event Handlers）
：每个事件都有相应的处理器来响应。

入口

try:
    select.poll
    Reactor = PollReactor
except:
    Reactor = SelectReactor

如果支持 select.poll，则调用PollReactor，否则，调用SelectReactor

初始化

class PollReactor(SelectReactor):
    def __init__(self, gc_checking=False):
        SelectReactor.__init__(self, gc_checking)
        self._poll = select.poll()
        self._fds = {}

class SelectReactor:
    NOW = _NOW
    NEVER = _NEVER
    def __init__(self, gc_checking=False):
        # Main code
        self._process = False
        self.monotonic = chelper.get_ffi()[1].get_monotonic
        # Python garbage collection
        self._check_gc = gc_checking
        self._last_gc_times = [0., 0., 0.]
        # Timers
        self._timers = []
        self._next_timer = self.NEVER
        # Callbacks
        self._pipe_fds = None
        self._async_queue = queue.Queue()
        # File descriptors
        self._fds = []
        # Greenlets
        self._g_dispatch = None
        self._greenlets = []
        self._all_greenlets = []

PollReactor 继承自SelectReactor，在初始化的时候，调用SelectReactor初始化，对timers、文件描述符、Greenlets等进行初始化。

def run(self):
    if self._pipe_fds is None:
        self._setup_async_callbacks()
    self._process = True
    g_next = ReactorGreenlet(run=self._dispatch_loop)
    self._all_greenlets.append(g_next)
    g_next.switch()

run方法在klippy.py 主循环打印机对象初始化后，会调用该方法。
self._process：设置True 为运行状态。
ReactorGreenlet(run=self._dispatch_loop)：生成协程，进入事件主循环协程。
g_next.switch()：切换到协程g_next。

事件驱动主循环

def _dispatch_loop(self):
    self._g_dispatch = g_dispatch = greenlet.getcurrent()
    busy = True
    eventtime = self.monotonic()
    while self._process:
        timeout = self._check_timers(eventtime, busy)
        busy = False
        res = self._poll.poll(int(math.ceil(timeout * 1000.)))
        eventtime = self.monotonic()
        for fd, event in res:
            busy = True
            self._fds[fd](eventtime)
            if g_dispatch is not self._g_dispatch:
                self._end_greenlet(g_dispatch)
                eventtime = self.monotonic()
                break
    self._g_dispatch = None

该方法实现了基于poll函数的事件循环，通过轮询文件描述符上的事件，并根据事件类型调用相应的回调函数进行处理。在处理完一个事件后，如果需要切换到其他协程进行执行，则使用greenlet实现切换。整个过程不断循环，直到_process 为False，则结束事件循环。

eventtime = self.monotonic()：获取当前的时间
timeout = self._check_timers(eventtime, busy)：检查是否有任何定时器事件需要处理。该方法返回下一个定时器事件的超时时间。
self._poll.poll(int(math.ceil(timeout * 1000.)))：轮询操作，等待文件描述符上的事件发生。
self._fds
fd
：调用文件描述符相应的回调函数。主要监听本地socket文件和一个reactor的双向管道文件描述符。
if g_dispatch is not self._g_dispatch：判断当前的greenlet与事件循环的greenlet是否相同，不同则切换协程。
self._end_greenlet(g_dispatch)：结束当前协程，并切换到其他协程。

协程

greenlet 是python实现协程的一个三方库，是一种基于协作式的多任务编程模型，允许在同一线程内实现多个并发执行的任务。
greenlet 是协作和顺序的。当一个 greenlet 运行时，其他 greenlet 都不能运行；开发者可以完全控制何时在 greenlet 之间切换执行。
协作式调度: 协程通常使用协作式调度，这意味着它们在适当的时机主动让出控制权，从而允许其他协程执行。这种方式可以有效避免线程切换的开销。
事件循环: 在许多协程实现中（如 Python 的
asyncio
），协程是在事件循环中运行的。事件循环负责调度和管理协程的执行。
I/O 密集型任务: 协程特别适合处理 I/O 密集型任务（如网络请求和文件操作），因为在等待 I/O 操作完成时，协程可以挂起自己，让其他协程继续执行。

定时器注册

def register_timer(self, callback, waketime=NEVER):

    timer_handler = ReactorTimer(callback, waketime)
    timers = list(self._timers)
    timers.append(timer_handler)
    self._timers = timers
    self._next_timer = min(self._next_timer, waketime)
    return timer_handler

将一个回调函数注册到定时器队列中

定时器检查

def _check_timers(self, eventtime, busy):
    if eventtime < self._next_timer:
        if busy:
            return 0.
        if self._check_gc:
            gi = gc.get_count()
            if gi[0] >= 700:
                # Reactor looks idle and gc is due - run it
                gc_level = 0
                if gi[1] >= 10:
                    gc_level = 1
                    if gi[2] >= 10:
                        gc_level = 2
                self._last_gc_times[gc_level] = eventtime
                gc.collect(gc_level)
                return 0.
        return min(1., max(.001, self._next_timer - eventtime))
    self._next_timer = self.NEVER
    g_dispatch = self._g_dispatch
    for t in self._timers:
        waketime = t.waketime
        if eventtime >= waketime:
            t.waketime = self.NEVER
            t.waketime = waketime = t.callback(eventtime)
            if g_dispatch is not self._g_dispatch:
                self._next_timer = min(self._next_timer, waketime)
                self._end_greenlet(g_dispatch)
                return 0.
        self._next_timer = min(self._next_timer, waketime)
    return 0.

该函数主要是通过检查定时器的到期时间，决定是否执行定时器的回调函数，以及更新下一次检查定时器的时间。
如果未到定时器执行时间，判断状态是否忙碌，忙碌直接返回；否则进行垃圾回收操作
遍历定时器列表，回调定时器函数进行处理
在触发回调函数后，检查是否发生了协程切换。如果
g_dispatch
和当前的
_g_dispatch
不一致，意味着其他协程已经接管控制，结束当前协程并返回。
循环结束后，更新下次要检查的定时器时间
self._next_timer
，确保系统按时检查到期的定时器。

定时器更新

def update_timer(self, timer_handler, waketime):
    timer_handler.waketime = waketime
    self._next_timer = min(self._next_timer, waketime)

主要是更新定时器的执行时间，是否立即执行。

klippy — reactor模块

reactor模式

入口

初始化

事件驱动主循环

协程

定时器注册

定时器检查

定时器更新

添加新评论

最新文章

最近回复

分类

归档

其它