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
|
import sys
from typing import TYPE_CHECKING
from . import _core
from ._abc import SendStream, ReceiveStream
from ._util import ConflictDetector, Final
from ._core._windows_cffi import _handle, raise_winerror, kernel32, ffi
assert sys.platform == "win32" or not TYPE_CHECKING
# XX TODO: don't just make this up based on nothing.
DEFAULT_RECEIVE_SIZE = 65536
# See the comments on _unix_pipes._FdHolder for discussion of why we set the
# handle to -1 when it's closed.
class _HandleHolder:
def __init__(self, handle: int) -> None:
self.handle = -1
if not isinstance(handle, int):
raise TypeError("handle must be an int")
self.handle = handle
_core.register_with_iocp(self.handle)
@property
def closed(self):
return self.handle == -1
def close(self):
if self.closed:
return
handle = self.handle
self.handle = -1
if not kernel32.CloseHandle(_handle(handle)):
raise_winerror()
def __del__(self):
self.close()
class PipeSendStream(SendStream, metaclass=Final):
"""Represents a send stream over a Windows named pipe that has been
opened in OVERLAPPED mode.
"""
def __init__(self, handle: int) -> None:
self._handle_holder = _HandleHolder(handle)
self._conflict_detector = ConflictDetector(
"another task is currently using this pipe"
)
async def send_all(self, data: bytes):
with self._conflict_detector:
if self._handle_holder.closed:
raise _core.ClosedResourceError("this pipe is already closed")
if not data:
await _core.checkpoint()
return
try:
written = await _core.write_overlapped(self._handle_holder.handle, data)
except BrokenPipeError as ex:
raise _core.BrokenResourceError from ex
# By my reading of MSDN, this assert is guaranteed to pass so long
# as the pipe isn't in nonblocking mode, but... let's just
# double-check.
assert written == len(data)
async def wait_send_all_might_not_block(self) -> None:
with self._conflict_detector:
if self._handle_holder.closed:
raise _core.ClosedResourceError("This pipe is already closed")
# not implemented yet, and probably not needed
await _core.checkpoint()
def close(self):
self._handle_holder.close()
async def aclose(self):
self.close()
await _core.checkpoint()
class PipeReceiveStream(ReceiveStream, metaclass=Final):
"""Represents a receive stream over an os.pipe object."""
def __init__(self, handle: int) -> None:
self._handle_holder = _HandleHolder(handle)
self._conflict_detector = ConflictDetector(
"another task is currently using this pipe"
)
async def receive_some(self, max_bytes=None) -> bytes:
with self._conflict_detector:
if self._handle_holder.closed:
raise _core.ClosedResourceError("this pipe is already closed")
if max_bytes is None:
max_bytes = DEFAULT_RECEIVE_SIZE
else:
if not isinstance(max_bytes, int):
raise TypeError("max_bytes must be integer >= 1")
if max_bytes < 1:
raise ValueError("max_bytes must be integer >= 1")
buffer = bytearray(max_bytes)
try:
size = await _core.readinto_overlapped(
self._handle_holder.handle, buffer
)
except BrokenPipeError:
if self._handle_holder.closed:
raise _core.ClosedResourceError(
"another task closed this pipe"
) from None
# Windows raises BrokenPipeError on one end of a pipe
# whenever the other end closes, regardless of direction.
# Convert this to the Unix behavior of returning EOF to the
# reader when the writer closes.
#
# And since we're not raising an exception, we have to
# checkpoint. But readinto_overlapped did raise an exception,
# so it might not have checkpointed for us. So we have to
# checkpoint manually.
await _core.checkpoint()
return b""
else:
del buffer[size:]
return buffer
def close(self):
self._handle_holder.close()
async def aclose(self):
self.close()
await _core.checkpoint()
|
