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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
# Test custom pollable objects implemented in Python.
from micropython import const
try:
import socket, select, io
except ImportError:
print("SKIP")
raise SystemExit
_MP_STREAM_POLL = const(3)
_MP_STREAM_GET_FILENO = const(10)
_MP_STREAM_POLL_RD = const(0x0001)
_MP_STREAM_POLL_WR = const(0x0004)
def print_poll_output(lst):
print([(type(obj), flags) for obj, flags in lst])
class CustomPollable(io.IOBase):
def __init__(self):
self.poll_state = 0
def ioctl(self, cmd, arg):
if cmd == _MP_STREAM_GET_FILENO:
# Bare-metal ports don't call this ioctl, so don't print it.
return -1
print("CustomPollable.ioctl", cmd, arg)
if cmd == _MP_STREAM_POLL:
if self.poll_state == "delay_rd":
self.poll_state = _MP_STREAM_POLL_RD
return 0
elif self.poll_state < 0:
return self.poll_state
else:
return self.poll_state & arg
poller = select.poll()
# Use a new UDP socket for tests, which should be writable but not readable.
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.bind(socket.getaddrinfo("127.0.0.1", 8000)[0][-1])
except OSError:
print("SKIP")
raise SystemExit
x = CustomPollable()
# Register both a file-descriptor-based object and a custom pure-Python object.
poller.register(s)
poller.register(x)
# Modify the flags for the custom object.
poller.modify(x, select.POLLIN)
# Test polling.
print_poll_output(poller.poll(0))
x.poll_state = _MP_STREAM_POLL_WR
print_poll_output(poller.poll(0))
x.poll_state = _MP_STREAM_POLL_RD
print_poll_output(poller.poll(0))
# The custom object becomes readable only after being polled.
poller.modify(s, select.POLLIN)
x.poll_state = "delay_rd"
print_poll_output(poller.poll())
# The custom object returns an error.
x.poll_state = -1000
try:
poller.poll(0)
except OSError as er:
print("OSError", er.errno)
# Register then unregister a socket (a native stream), then test
# that the Python object is still pollable.
s2 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
x.poll_state = _MP_STREAM_POLL_RD
poller.register(s2)
poller.unregister(s2)
print_poll_output(poller.poll())
# Test registering and unregistering multiple times.
for _ in range(2):
poller.unregister(s)
poller.unregister(x)
poller.register(s2)
poller.register(s, select.POLLIN)
poller.register(x, select.POLLIN)
poller.unregister(s2)
print_poll_output(poller.poll())
# Clean up.
poller.unregister(x)
poller.unregister(s)
s2.close()
s.close()
|
