blob: 0813a6614957db64b8d650504640df05676f0857 (
plain)
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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
|
import selectors
import sys
import threading
import time
from typing import List, Optional, Tuple
from mini_scalene import MiniScalene
@MiniScalene.shim
def replacement_poll_selector(mini_scalene: MiniScalene) -> None:
"""
A replacement for selectors.PollSelector that
periodically wakes up to accept signals
"""
class ReplacementPollSelector(selectors.PollSelector):
def select(
self, timeout: Optional[float] = -1
) -> List[Tuple[selectors.SelectorKey, int]]:
tident = threading.get_ident()
start_time = time.perf_counter()
if not timeout or timeout < 0:
interval = sys.getswitchinterval()
else:
interval = min(timeout, sys.getswitchinterval())
while True:
scalene.set_thread_sleeping(tident)
selected = super().select(interval)
scalene.reset_thread_sleeping(tident)
if selected or timeout == 0:
return selected
end_time = time.perf_counter()
if timeout and timeout != -1:
if end_time - start_time >= timeout:
return [] # None
ReplacementPollSelector.__qualname__ = (
"replacement_poll_selector.ReplacementPollSelector"
)
selectors.PollSelector = ReplacementPollSelector # type: ignore
|
