replacement_epoll_selector.py


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import selectors
import sys
import time
from typing import List, Tuple


class ReplacementEpollSelector(selectors.EpollSelector):
    def select(
        self, timeout=None
    ) -> List[Tuple[selectors.SelectorKey, int]]:
        start_time = time.perf_counter()
        if not timeout or timeout < 0:
            interval = sys.getswitchinterval()
        else:
            interval = min(timeout, sys.getswitchinterval())
        while True:
            selected = super().select(interval)
            if selected or timeout == 0 or not timeout:
                return selected
            end_time = time.perf_counter()
            if end_time - start_time >= timeout:
                return []  # None

ReplacementEpollSelector.__qualname__ = (
    "replacement_epoll_selector.ReplacementEpollSelector"
)
selectors.DefaultSelector = ReplacementEpollSelector