-
-
Notifications
You must be signed in to change notification settings - Fork 323
/
_generated_io_windows.py
202 lines (168 loc) · 7.58 KB
/
_generated_io_windows.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
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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
# ***********************************************************
# ******* WARNING: AUTOGENERATED! ALL EDITS WILL BE LOST ******
# *************************************************************
from __future__ import annotations
from typing import TYPE_CHECKING, ContextManager
from ._ki import LOCALS_KEY_KI_PROTECTION_ENABLED
from ._run import GLOBAL_RUN_CONTEXT
if TYPE_CHECKING:
from typing_extensions import Buffer
from .._file_io import _HasFileNo
from ._unbounded_queue import UnboundedQueue
from ._windows_cffi import CData, Handle
import sys
assert not TYPE_CHECKING or sys.platform == "win32"
__all__ = [
"current_iocp",
"monitor_completion_key",
"notify_closing",
"readinto_overlapped",
"register_with_iocp",
"wait_overlapped",
"wait_readable",
"wait_writable",
"write_overlapped",
]
async def wait_readable(sock: (_HasFileNo | int)) -> None:
"""Block until the kernel reports that the given object is readable.
On Unix systems, ``sock`` must either be an integer file descriptor,
or else an object with a ``.fileno()`` method which returns an
integer file descriptor. Any kind of file descriptor can be passed,
though the exact semantics will depend on your kernel. For example,
this probably won't do anything useful for on-disk files.
On Windows systems, ``sock`` must either be an integer ``SOCKET``
handle, or else an object with a ``.fileno()`` method which returns
an integer ``SOCKET`` handle. File descriptors aren't supported,
and neither are handles that refer to anything besides a
``SOCKET``.
:raises trio.BusyResourceError:
if another task is already waiting for the given socket to
become readable.
:raises trio.ClosedResourceError:
if another task calls :func:`notify_closing` while this
function is still working.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return await GLOBAL_RUN_CONTEXT.runner.io_manager.wait_readable(sock)
except AttributeError:
raise RuntimeError("must be called from async context") from None
async def wait_writable(sock: (_HasFileNo | int)) -> None:
"""Block until the kernel reports that the given object is writable.
See `wait_readable` for the definition of ``sock``.
:raises trio.BusyResourceError:
if another task is already waiting for the given socket to
become writable.
:raises trio.ClosedResourceError:
if another task calls :func:`notify_closing` while this
function is still working.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return await GLOBAL_RUN_CONTEXT.runner.io_manager.wait_writable(sock)
except AttributeError:
raise RuntimeError("must be called from async context") from None
def notify_closing(handle: (Handle | int | _HasFileNo)) -> None:
"""Notify waiters of the given object that it will be closed.
Call this before closing a file descriptor (on Unix) or socket (on
Windows). This will cause any `wait_readable` or `wait_writable`
calls on the given object to immediately wake up and raise
`~trio.ClosedResourceError`.
This doesn't actually close the object – you still have to do that
yourself afterwards. Also, you want to be careful to make sure no
new tasks start waiting on the object in between when you call this
and when it's actually closed. So to close something properly, you
usually want to do these steps in order:
1. Explicitly mark the object as closed, so that any new attempts
to use it will abort before they start.
2. Call `notify_closing` to wake up any already-existing users.
3. Actually close the object.
It's also possible to do them in a different order if that's more
convenient, *but only if* you make sure not to have any checkpoints in
between the steps. This way they all happen in a single atomic
step, so other tasks won't be able to tell what order they happened
in anyway.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return GLOBAL_RUN_CONTEXT.runner.io_manager.notify_closing(handle)
except AttributeError:
raise RuntimeError("must be called from async context") from None
def register_with_iocp(handle: (int | CData)) -> None:
"""TODO: these are implemented, but are currently more of a sketch than
anything real. See `#26
<https://github.com/python-trio/trio/issues/26>`__ and `#52
<https://github.com/python-trio/trio/issues/52>`__.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return GLOBAL_RUN_CONTEXT.runner.io_manager.register_with_iocp(handle)
except AttributeError:
raise RuntimeError("must be called from async context") from None
async def wait_overlapped(
handle_: (int | CData), lpOverlapped: (CData | int)
) -> object:
"""TODO: these are implemented, but are currently more of a sketch than
anything real. See `#26
<https://github.com/python-trio/trio/issues/26>`__ and `#52
<https://github.com/python-trio/trio/issues/52>`__.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return await GLOBAL_RUN_CONTEXT.runner.io_manager.wait_overlapped(
handle_, lpOverlapped
)
except AttributeError:
raise RuntimeError("must be called from async context") from None
async def write_overlapped(
handle: (int | CData), data: Buffer, file_offset: int = 0
) -> int:
"""TODO: these are implemented, but are currently more of a sketch than
anything real. See `#26
<https://github.com/python-trio/trio/issues/26>`__ and `#52
<https://github.com/python-trio/trio/issues/52>`__.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return await GLOBAL_RUN_CONTEXT.runner.io_manager.write_overlapped(
handle, data, file_offset
)
except AttributeError:
raise RuntimeError("must be called from async context") from None
async def readinto_overlapped(
handle: (int | CData), buffer: Buffer, file_offset: int = 0
) -> int:
"""TODO: these are implemented, but are currently more of a sketch than
anything real. See `#26
<https://github.com/python-trio/trio/issues/26>`__ and `#52
<https://github.com/python-trio/trio/issues/52>`__.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return await GLOBAL_RUN_CONTEXT.runner.io_manager.readinto_overlapped(
handle, buffer, file_offset
)
except AttributeError:
raise RuntimeError("must be called from async context") from None
def current_iocp() -> int:
"""TODO: these are implemented, but are currently more of a sketch than
anything real. See `#26
<https://github.com/python-trio/trio/issues/26>`__ and `#52
<https://github.com/python-trio/trio/issues/52>`__.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return GLOBAL_RUN_CONTEXT.runner.io_manager.current_iocp()
except AttributeError:
raise RuntimeError("must be called from async context") from None
def monitor_completion_key() -> ContextManager[tuple[int, UnboundedQueue[object]]]:
"""TODO: these are implemented, but are currently more of a sketch than
anything real. See `#26
<https://github.com/python-trio/trio/issues/26>`__ and `#52
<https://github.com/python-trio/trio/issues/52>`__.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return GLOBAL_RUN_CONTEXT.runner.io_manager.monitor_completion_key()
except AttributeError:
raise RuntimeError("must be called from async context") from None