ยปCore Development>Code coverage>Lib/multiprocessing/connection.py

Python code coverage for Lib/multiprocessing/connection.py

#countcontent
1n/a#
2n/a# A higher level module for using sockets (or Windows named pipes)
3n/a#
4n/a# multiprocessing/connection.py
5n/a#
6n/a# Copyright (c) 2006-2008, R Oudkerk
7n/a# Licensed to PSF under a Contributor Agreement.
8n/a#
9n/a
10n/a__all__ = [ 'Client', 'Listener', 'Pipe', 'wait' ]
11n/a
12n/aimport io
13n/aimport os
14n/aimport sys
15n/aimport socket
16n/aimport struct
17n/aimport time
18n/aimport tempfile
19n/aimport itertools
20n/a
21n/aimport _multiprocessing
22n/a
23n/afrom . import util
24n/a
25n/afrom . import AuthenticationError, BufferTooShort
26n/afrom .context import reduction
27n/a_ForkingPickler = reduction.ForkingPickler
28n/a
29n/atry:
30n/a import _winapi
31n/a from _winapi import WAIT_OBJECT_0, WAIT_ABANDONED_0, WAIT_TIMEOUT, INFINITE
32n/aexcept ImportError:
33n/a if sys.platform == 'win32':
34n/a raise
35n/a _winapi = None
36n/a
37n/a#
38n/a#
39n/a#
40n/a
41n/aBUFSIZE = 8192
42n/a# A very generous timeout when it comes to local connections...
43n/aCONNECTION_TIMEOUT = 20.
44n/a
45n/a_mmap_counter = itertools.count()
46n/a
47n/adefault_family = 'AF_INET'
48n/afamilies = ['AF_INET']
49n/a
50n/aif hasattr(socket, 'AF_UNIX'):
51n/a default_family = 'AF_UNIX'
52n/a families += ['AF_UNIX']
53n/a
54n/aif sys.platform == 'win32':
55n/a default_family = 'AF_PIPE'
56n/a families += ['AF_PIPE']
57n/a
58n/a
59n/adef _init_timeout(timeout=CONNECTION_TIMEOUT):
60n/a return time.time() + timeout
61n/a
62n/adef _check_timeout(t):
63n/a return time.time() > t
64n/a
65n/a#
66n/a#
67n/a#
68n/a
69n/adef arbitrary_address(family):
70n/a '''
71n/a Return an arbitrary free address for the given family
72n/a '''
73n/a if family == 'AF_INET':
74n/a return ('localhost', 0)
75n/a elif family == 'AF_UNIX':
76n/a return tempfile.mktemp(prefix='listener-', dir=util.get_temp_dir())
77n/a elif family == 'AF_PIPE':
78n/a return tempfile.mktemp(prefix=r'\\.\pipe\pyc-%d-%d-' %
79n/a (os.getpid(), next(_mmap_counter)), dir="")
80n/a else:
81n/a raise ValueError('unrecognized family')
82n/a
83n/adef _validate_family(family):
84n/a '''
85n/a Checks if the family is valid for the current environment.
86n/a '''
87n/a if sys.platform != 'win32' and family == 'AF_PIPE':
88n/a raise ValueError('Family %s is not recognized.' % family)
89n/a
90n/a if sys.platform == 'win32' and family == 'AF_UNIX':
91n/a # double check
92n/a if not hasattr(socket, family):
93n/a raise ValueError('Family %s is not recognized.' % family)
94n/a
95n/adef address_type(address):
96n/a '''
97n/a Return the types of the address
98n/a
99n/a This can be 'AF_INET', 'AF_UNIX', or 'AF_PIPE'
100n/a '''
101n/a if type(address) == tuple:
102n/a return 'AF_INET'
103n/a elif type(address) is str and address.startswith('\\\\'):
104n/a return 'AF_PIPE'
105n/a elif type(address) is str:
106n/a return 'AF_UNIX'
107n/a else:
108n/a raise ValueError('address type of %r unrecognized' % address)
109n/a
110n/a#
111n/a# Connection classes
112n/a#
113n/a
114n/aclass _ConnectionBase:
115n/a _handle = None
116n/a
117n/a def __init__(self, handle, readable=True, writable=True):
118n/a handle = handle.__index__()
119n/a if handle < 0:
120n/a raise ValueError("invalid handle")
121n/a if not readable and not writable:
122n/a raise ValueError(
123n/a "at least one of `readable` and `writable` must be True")
124n/a self._handle = handle
125n/a self._readable = readable
126n/a self._writable = writable
127n/a
128n/a # XXX should we use util.Finalize instead of a __del__?
129n/a
130n/a def __del__(self):
131n/a if self._handle is not None:
132n/a self._close()
133n/a
134n/a def _check_closed(self):
135n/a if self._handle is None:
136n/a raise OSError("handle is closed")
137n/a
138n/a def _check_readable(self):
139n/a if not self._readable:
140n/a raise OSError("connection is write-only")
141n/a
142n/a def _check_writable(self):
143n/a if not self._writable:
144n/a raise OSError("connection is read-only")
145n/a
146n/a def _bad_message_length(self):
147n/a if self._writable:
148n/a self._readable = False
149n/a else:
150n/a self.close()
151n/a raise OSError("bad message length")
152n/a
153n/a @property
154n/a def closed(self):
155n/a """True if the connection is closed"""
156n/a return self._handle is None
157n/a
158n/a @property
159n/a def readable(self):
160n/a """True if the connection is readable"""
161n/a return self._readable
162n/a
163n/a @property
164n/a def writable(self):
165n/a """True if the connection is writable"""
166n/a return self._writable
167n/a
168n/a def fileno(self):
169n/a """File descriptor or handle of the connection"""
170n/a self._check_closed()
171n/a return self._handle
172n/a
173n/a def close(self):
174n/a """Close the connection"""
175n/a if self._handle is not None:
176n/a try:
177n/a self._close()
178n/a finally:
179n/a self._handle = None
180n/a
181n/a def send_bytes(self, buf, offset=0, size=None):
182n/a """Send the bytes data from a bytes-like object"""
183n/a self._check_closed()
184n/a self._check_writable()
185n/a m = memoryview(buf)
186n/a # HACK for byte-indexing of non-bytewise buffers (e.g. array.array)
187n/a if m.itemsize > 1:
188n/a m = memoryview(bytes(m))
189n/a n = len(m)
190n/a if offset < 0:
191n/a raise ValueError("offset is negative")
192n/a if n < offset:
193n/a raise ValueError("buffer length < offset")
194n/a if size is None:
195n/a size = n - offset
196n/a elif size < 0:
197n/a raise ValueError("size is negative")
198n/a elif offset + size > n:
199n/a raise ValueError("buffer length < offset + size")
200n/a self._send_bytes(m[offset:offset + size])
201n/a
202n/a def send(self, obj):
203n/a """Send a (picklable) object"""
204n/a self._check_closed()
205n/a self._check_writable()
206n/a self._send_bytes(_ForkingPickler.dumps(obj))
207n/a
208n/a def recv_bytes(self, maxlength=None):
209n/a """
210n/a Receive bytes data as a bytes object.
211n/a """
212n/a self._check_closed()
213n/a self._check_readable()
214n/a if maxlength is not None and maxlength < 0:
215n/a raise ValueError("negative maxlength")
216n/a buf = self._recv_bytes(maxlength)
217n/a if buf is None:
218n/a self._bad_message_length()
219n/a return buf.getvalue()
220n/a
221n/a def recv_bytes_into(self, buf, offset=0):
222n/a """
223n/a Receive bytes data into a writeable bytes-like object.
224n/a Return the number of bytes read.
225n/a """
226n/a self._check_closed()
227n/a self._check_readable()
228n/a with memoryview(buf) as m:
229n/a # Get bytesize of arbitrary buffer
230n/a itemsize = m.itemsize
231n/a bytesize = itemsize * len(m)
232n/a if offset < 0:
233n/a raise ValueError("negative offset")
234n/a elif offset > bytesize:
235n/a raise ValueError("offset too large")
236n/a result = self._recv_bytes()
237n/a size = result.tell()
238n/a if bytesize < offset + size:
239n/a raise BufferTooShort(result.getvalue())
240n/a # Message can fit in dest
241n/a result.seek(0)
242n/a result.readinto(m[offset // itemsize :
243n/a (offset + size) // itemsize])
244n/a return size
245n/a
246n/a def recv(self):
247n/a """Receive a (picklable) object"""
248n/a self._check_closed()
249n/a self._check_readable()
250n/a buf = self._recv_bytes()
251n/a return _ForkingPickler.loads(buf.getbuffer())
252n/a
253n/a def poll(self, timeout=0.0):
254n/a """Whether there is any input available to be read"""
255n/a self._check_closed()
256n/a self._check_readable()
257n/a return self._poll(timeout)
258n/a
259n/a def __enter__(self):
260n/a return self
261n/a
262n/a def __exit__(self, exc_type, exc_value, exc_tb):
263n/a self.close()
264n/a
265n/a
266n/aif _winapi:
267n/a
268n/a class PipeConnection(_ConnectionBase):
269n/a """
270n/a Connection class based on a Windows named pipe.
271n/a Overlapped I/O is used, so the handles must have been created
272n/a with FILE_FLAG_OVERLAPPED.
273n/a """
274n/a _got_empty_message = False
275n/a
276n/a def _close(self, _CloseHandle=_winapi.CloseHandle):
277n/a _CloseHandle(self._handle)
278n/a
279n/a def _send_bytes(self, buf):
280n/a ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
281n/a try:
282n/a if err == _winapi.ERROR_IO_PENDING:
283n/a waitres = _winapi.WaitForMultipleObjects(
284n/a [ov.event], False, INFINITE)
285n/a assert waitres == WAIT_OBJECT_0
286n/a except:
287n/a ov.cancel()
288n/a raise
289n/a finally:
290n/a nwritten, err = ov.GetOverlappedResult(True)
291n/a assert err == 0
292n/a assert nwritten == len(buf)
293n/a
294n/a def _recv_bytes(self, maxsize=None):
295n/a if self._got_empty_message:
296n/a self._got_empty_message = False
297n/a return io.BytesIO()
298n/a else:
299n/a bsize = 128 if maxsize is None else min(maxsize, 128)
300n/a try:
301n/a ov, err = _winapi.ReadFile(self._handle, bsize,
302n/a overlapped=True)
303n/a try:
304n/a if err == _winapi.ERROR_IO_PENDING:
305n/a waitres = _winapi.WaitForMultipleObjects(
306n/a [ov.event], False, INFINITE)
307n/a assert waitres == WAIT_OBJECT_0
308n/a except:
309n/a ov.cancel()
310n/a raise
311n/a finally:
312n/a nread, err = ov.GetOverlappedResult(True)
313n/a if err == 0:
314n/a f = io.BytesIO()
315n/a f.write(ov.getbuffer())
316n/a return f
317n/a elif err == _winapi.ERROR_MORE_DATA:
318n/a return self._get_more_data(ov, maxsize)
319n/a except OSError as e:
320n/a if e.winerror == _winapi.ERROR_BROKEN_PIPE:
321n/a raise EOFError
322n/a else:
323n/a raise
324n/a raise RuntimeError("shouldn't get here; expected KeyboardInterrupt")
325n/a
326n/a def _poll(self, timeout):
327n/a if (self._got_empty_message or
328n/a _winapi.PeekNamedPipe(self._handle)[0] != 0):
329n/a return True
330n/a return bool(wait([self], timeout))
331n/a
332n/a def _get_more_data(self, ov, maxsize):
333n/a buf = ov.getbuffer()
334n/a f = io.BytesIO()
335n/a f.write(buf)
336n/a left = _winapi.PeekNamedPipe(self._handle)[1]
337n/a assert left > 0
338n/a if maxsize is not None and len(buf) + left > maxsize:
339n/a self._bad_message_length()
340n/a ov, err = _winapi.ReadFile(self._handle, left, overlapped=True)
341n/a rbytes, err = ov.GetOverlappedResult(True)
342n/a assert err == 0
343n/a assert rbytes == left
344n/a f.write(ov.getbuffer())
345n/a return f
346n/a
347n/a
348n/aclass Connection(_ConnectionBase):
349n/a """
350n/a Connection class based on an arbitrary file descriptor (Unix only), or
351n/a a socket handle (Windows).
352n/a """
353n/a
354n/a if _winapi:
355n/a def _close(self, _close=_multiprocessing.closesocket):
356n/a _close(self._handle)
357n/a _write = _multiprocessing.send
358n/a _read = _multiprocessing.recv
359n/a else:
360n/a def _close(self, _close=os.close):
361n/a _close(self._handle)
362n/a _write = os.write
363n/a _read = os.read
364n/a
365n/a def _send(self, buf, write=_write):
366n/a remaining = len(buf)
367n/a while True:
368n/a n = write(self._handle, buf)
369n/a remaining -= n
370n/a if remaining == 0:
371n/a break
372n/a buf = buf[n:]
373n/a
374n/a def _recv(self, size, read=_read):
375n/a buf = io.BytesIO()
376n/a handle = self._handle
377n/a remaining = size
378n/a while remaining > 0:
379n/a chunk = read(handle, remaining)
380n/a n = len(chunk)
381n/a if n == 0:
382n/a if remaining == size:
383n/a raise EOFError
384n/a else:
385n/a raise OSError("got end of file during message")
386n/a buf.write(chunk)
387n/a remaining -= n
388n/a return buf
389n/a
390n/a def _send_bytes(self, buf):
391n/a n = len(buf)
392n/a # For wire compatibility with 3.2 and lower
393n/a header = struct.pack("!i", n)
394n/a if n > 16384:
395n/a # The payload is large so Nagle's algorithm won't be triggered
396n/a # and we'd better avoid the cost of concatenation.
397n/a self._send(header)
398n/a self._send(buf)
399n/a else:
400n/a # Issue #20540: concatenate before sending, to avoid delays due
401n/a # to Nagle's algorithm on a TCP socket.
402n/a # Also note we want to avoid sending a 0-length buffer separately,
403n/a # to avoid "broken pipe" errors if the other end closed the pipe.
404n/a self._send(header + buf)
405n/a
406n/a def _recv_bytes(self, maxsize=None):
407n/a buf = self._recv(4)
408n/a size, = struct.unpack("!i", buf.getvalue())
409n/a if maxsize is not None and size > maxsize:
410n/a return None
411n/a return self._recv(size)
412n/a
413n/a def _poll(self, timeout):
414n/a r = wait([self], timeout)
415n/a return bool(r)
416n/a
417n/a
418n/a#
419n/a# Public functions
420n/a#
421n/a
422n/aclass Listener(object):
423n/a '''
424n/a Returns a listener object.
425n/a
426n/a This is a wrapper for a bound socket which is 'listening' for
427n/a connections, or for a Windows named pipe.
428n/a '''
429n/a def __init__(self, address=None, family=None, backlog=1, authkey=None):
430n/a family = family or (address and address_type(address)) \
431n/a or default_family
432n/a address = address or arbitrary_address(family)
433n/a
434n/a _validate_family(family)
435n/a if family == 'AF_PIPE':
436n/a self._listener = PipeListener(address, backlog)
437n/a else:
438n/a self._listener = SocketListener(address, family, backlog)
439n/a
440n/a if authkey is not None and not isinstance(authkey, bytes):
441n/a raise TypeError('authkey should be a byte string')
442n/a
443n/a self._authkey = authkey
444n/a
445n/a def accept(self):
446n/a '''
447n/a Accept a connection on the bound socket or named pipe of `self`.
448n/a
449n/a Returns a `Connection` object.
450n/a '''
451n/a if self._listener is None:
452n/a raise OSError('listener is closed')
453n/a c = self._listener.accept()
454n/a if self._authkey:
455n/a deliver_challenge(c, self._authkey)
456n/a answer_challenge(c, self._authkey)
457n/a return c
458n/a
459n/a def close(self):
460n/a '''
461n/a Close the bound socket or named pipe of `self`.
462n/a '''
463n/a listener = self._listener
464n/a if listener is not None:
465n/a self._listener = None
466n/a listener.close()
467n/a
468n/a address = property(lambda self: self._listener._address)
469n/a last_accepted = property(lambda self: self._listener._last_accepted)
470n/a
471n/a def __enter__(self):
472n/a return self
473n/a
474n/a def __exit__(self, exc_type, exc_value, exc_tb):
475n/a self.close()
476n/a
477n/a
478n/adef Client(address, family=None, authkey=None):
479n/a '''
480n/a Returns a connection to the address of a `Listener`
481n/a '''
482n/a family = family or address_type(address)
483n/a _validate_family(family)
484n/a if family == 'AF_PIPE':
485n/a c = PipeClient(address)
486n/a else:
487n/a c = SocketClient(address)
488n/a
489n/a if authkey is not None and not isinstance(authkey, bytes):
490n/a raise TypeError('authkey should be a byte string')
491n/a
492n/a if authkey is not None:
493n/a answer_challenge(c, authkey)
494n/a deliver_challenge(c, authkey)
495n/a
496n/a return c
497n/a
498n/a
499n/aif sys.platform != 'win32':
500n/a
501n/a def Pipe(duplex=True):
502n/a '''
503n/a Returns pair of connection objects at either end of a pipe
504n/a '''
505n/a if duplex:
506n/a s1, s2 = socket.socketpair()
507n/a s1.setblocking(True)
508n/a s2.setblocking(True)
509n/a c1 = Connection(s1.detach())
510n/a c2 = Connection(s2.detach())
511n/a else:
512n/a fd1, fd2 = os.pipe()
513n/a c1 = Connection(fd1, writable=False)
514n/a c2 = Connection(fd2, readable=False)
515n/a
516n/a return c1, c2
517n/a
518n/aelse:
519n/a
520n/a def Pipe(duplex=True):
521n/a '''
522n/a Returns pair of connection objects at either end of a pipe
523n/a '''
524n/a address = arbitrary_address('AF_PIPE')
525n/a if duplex:
526n/a openmode = _winapi.PIPE_ACCESS_DUPLEX
527n/a access = _winapi.GENERIC_READ | _winapi.GENERIC_WRITE
528n/a obsize, ibsize = BUFSIZE, BUFSIZE
529n/a else:
530n/a openmode = _winapi.PIPE_ACCESS_INBOUND
531n/a access = _winapi.GENERIC_WRITE
532n/a obsize, ibsize = 0, BUFSIZE
533n/a
534n/a h1 = _winapi.CreateNamedPipe(
535n/a address, openmode | _winapi.FILE_FLAG_OVERLAPPED |
536n/a _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE,
537n/a _winapi.PIPE_TYPE_MESSAGE | _winapi.PIPE_READMODE_MESSAGE |
538n/a _winapi.PIPE_WAIT,
539n/a 1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER,
540n/a # default security descriptor: the handle cannot be inherited
541n/a _winapi.NULL
542n/a )
543n/a h2 = _winapi.CreateFile(
544n/a address, access, 0, _winapi.NULL, _winapi.OPEN_EXISTING,
545n/a _winapi.FILE_FLAG_OVERLAPPED, _winapi.NULL
546n/a )
547n/a _winapi.SetNamedPipeHandleState(
548n/a h2, _winapi.PIPE_READMODE_MESSAGE, None, None
549n/a )
550n/a
551n/a overlapped = _winapi.ConnectNamedPipe(h1, overlapped=True)
552n/a _, err = overlapped.GetOverlappedResult(True)
553n/a assert err == 0
554n/a
555n/a c1 = PipeConnection(h1, writable=duplex)
556n/a c2 = PipeConnection(h2, readable=duplex)
557n/a
558n/a return c1, c2
559n/a
560n/a#
561n/a# Definitions for connections based on sockets
562n/a#
563n/a
564n/aclass SocketListener(object):
565n/a '''
566n/a Representation of a socket which is bound to an address and listening
567n/a '''
568n/a def __init__(self, address, family, backlog=1):
569n/a self._socket = socket.socket(getattr(socket, family))
570n/a try:
571n/a # SO_REUSEADDR has different semantics on Windows (issue #2550).
572n/a if os.name == 'posix':
573n/a self._socket.setsockopt(socket.SOL_SOCKET,
574n/a socket.SO_REUSEADDR, 1)
575n/a self._socket.setblocking(True)
576n/a self._socket.bind(address)
577n/a self._socket.listen(backlog)
578n/a self._address = self._socket.getsockname()
579n/a except OSError:
580n/a self._socket.close()
581n/a raise
582n/a self._family = family
583n/a self._last_accepted = None
584n/a
585n/a if family == 'AF_UNIX':
586n/a self._unlink = util.Finalize(
587n/a self, os.unlink, args=(address,), exitpriority=0
588n/a )
589n/a else:
590n/a self._unlink = None
591n/a
592n/a def accept(self):
593n/a s, self._last_accepted = self._socket.accept()
594n/a s.setblocking(True)
595n/a return Connection(s.detach())
596n/a
597n/a def close(self):
598n/a try:
599n/a self._socket.close()
600n/a finally:
601n/a unlink = self._unlink
602n/a if unlink is not None:
603n/a self._unlink = None
604n/a unlink()
605n/a
606n/a
607n/adef SocketClient(address):
608n/a '''
609n/a Return a connection object connected to the socket given by `address`
610n/a '''
611n/a family = address_type(address)
612n/a with socket.socket( getattr(socket, family) ) as s:
613n/a s.setblocking(True)
614n/a s.connect(address)
615n/a return Connection(s.detach())
616n/a
617n/a#
618n/a# Definitions for connections based on named pipes
619n/a#
620n/a
621n/aif sys.platform == 'win32':
622n/a
623n/a class PipeListener(object):
624n/a '''
625n/a Representation of a named pipe
626n/a '''
627n/a def __init__(self, address, backlog=None):
628n/a self._address = address
629n/a self._handle_queue = [self._new_handle(first=True)]
630n/a
631n/a self._last_accepted = None
632n/a util.sub_debug('listener created with address=%r', self._address)
633n/a self.close = util.Finalize(
634n/a self, PipeListener._finalize_pipe_listener,
635n/a args=(self._handle_queue, self._address), exitpriority=0
636n/a )
637n/a
638n/a def _new_handle(self, first=False):
639n/a flags = _winapi.PIPE_ACCESS_DUPLEX | _winapi.FILE_FLAG_OVERLAPPED
640n/a if first:
641n/a flags |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE
642n/a return _winapi.CreateNamedPipe(
643n/a self._address, flags,
644n/a _winapi.PIPE_TYPE_MESSAGE | _winapi.PIPE_READMODE_MESSAGE |
645n/a _winapi.PIPE_WAIT,
646n/a _winapi.PIPE_UNLIMITED_INSTANCES, BUFSIZE, BUFSIZE,
647n/a _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL
648n/a )
649n/a
650n/a def accept(self):
651n/a self._handle_queue.append(self._new_handle())
652n/a handle = self._handle_queue.pop(0)
653n/a try:
654n/a ov = _winapi.ConnectNamedPipe(handle, overlapped=True)
655n/a except OSError as e:
656n/a if e.winerror != _winapi.ERROR_NO_DATA:
657n/a raise
658n/a # ERROR_NO_DATA can occur if a client has already connected,
659n/a # written data and then disconnected -- see Issue 14725.
660n/a else:
661n/a try:
662n/a res = _winapi.WaitForMultipleObjects(
663n/a [ov.event], False, INFINITE)
664n/a except:
665n/a ov.cancel()
666n/a _winapi.CloseHandle(handle)
667n/a raise
668n/a finally:
669n/a _, err = ov.GetOverlappedResult(True)
670n/a assert err == 0
671n/a return PipeConnection(handle)
672n/a
673n/a @staticmethod
674n/a def _finalize_pipe_listener(queue, address):
675n/a util.sub_debug('closing listener with address=%r', address)
676n/a for handle in queue:
677n/a _winapi.CloseHandle(handle)
678n/a
679n/a def PipeClient(address):
680n/a '''
681n/a Return a connection object connected to the pipe given by `address`
682n/a '''
683n/a t = _init_timeout()
684n/a while 1:
685n/a try:
686n/a _winapi.WaitNamedPipe(address, 1000)
687n/a h = _winapi.CreateFile(
688n/a address, _winapi.GENERIC_READ | _winapi.GENERIC_WRITE,
689n/a 0, _winapi.NULL, _winapi.OPEN_EXISTING,
690n/a _winapi.FILE_FLAG_OVERLAPPED, _winapi.NULL
691n/a )
692n/a except OSError as e:
693n/a if e.winerror not in (_winapi.ERROR_SEM_TIMEOUT,
694n/a _winapi.ERROR_PIPE_BUSY) or _check_timeout(t):
695n/a raise
696n/a else:
697n/a break
698n/a else:
699n/a raise
700n/a
701n/a _winapi.SetNamedPipeHandleState(
702n/a h, _winapi.PIPE_READMODE_MESSAGE, None, None
703n/a )
704n/a return PipeConnection(h)
705n/a
706n/a#
707n/a# Authentication stuff
708n/a#
709n/a
710n/aMESSAGE_LENGTH = 20
711n/a
712n/aCHALLENGE = b'#CHALLENGE#'
713n/aWELCOME = b'#WELCOME#'
714n/aFAILURE = b'#FAILURE#'
715n/a
716n/adef deliver_challenge(connection, authkey):
717n/a import hmac
718n/a assert isinstance(authkey, bytes)
719n/a message = os.urandom(MESSAGE_LENGTH)
720n/a connection.send_bytes(CHALLENGE + message)
721n/a digest = hmac.new(authkey, message, 'md5').digest()
722n/a response = connection.recv_bytes(256) # reject large message
723n/a if response == digest:
724n/a connection.send_bytes(WELCOME)
725n/a else:
726n/a connection.send_bytes(FAILURE)
727n/a raise AuthenticationError('digest received was wrong')
728n/a
729n/adef answer_challenge(connection, authkey):
730n/a import hmac
731n/a assert isinstance(authkey, bytes)
732n/a message = connection.recv_bytes(256) # reject large message
733n/a assert message[:len(CHALLENGE)] == CHALLENGE, 'message = %r' % message
734n/a message = message[len(CHALLENGE):]
735n/a digest = hmac.new(authkey, message, 'md5').digest()
736n/a connection.send_bytes(digest)
737n/a response = connection.recv_bytes(256) # reject large message
738n/a if response != WELCOME:
739n/a raise AuthenticationError('digest sent was rejected')
740n/a
741n/a#
742n/a# Support for using xmlrpclib for serialization
743n/a#
744n/a
745n/aclass ConnectionWrapper(object):
746n/a def __init__(self, conn, dumps, loads):
747n/a self._conn = conn
748n/a self._dumps = dumps
749n/a self._loads = loads
750n/a for attr in ('fileno', 'close', 'poll', 'recv_bytes', 'send_bytes'):
751n/a obj = getattr(conn, attr)
752n/a setattr(self, attr, obj)
753n/a def send(self, obj):
754n/a s = self._dumps(obj)
755n/a self._conn.send_bytes(s)
756n/a def recv(self):
757n/a s = self._conn.recv_bytes()
758n/a return self._loads(s)
759n/a
760n/adef _xml_dumps(obj):
761n/a return xmlrpclib.dumps((obj,), None, None, None, 1).encode('utf-8')
762n/a
763n/adef _xml_loads(s):
764n/a (obj,), method = xmlrpclib.loads(s.decode('utf-8'))
765n/a return obj
766n/a
767n/aclass XmlListener(Listener):
768n/a def accept(self):
769n/a global xmlrpclib
770n/a import xmlrpc.client as xmlrpclib
771n/a obj = Listener.accept(self)
772n/a return ConnectionWrapper(obj, _xml_dumps, _xml_loads)
773n/a
774n/adef XmlClient(*args, **kwds):
775n/a global xmlrpclib
776n/a import xmlrpc.client as xmlrpclib
777n/a return ConnectionWrapper(Client(*args, **kwds), _xml_dumps, _xml_loads)
778n/a
779n/a#
780n/a# Wait
781n/a#
782n/a
783n/aif sys.platform == 'win32':
784n/a
785n/a def _exhaustive_wait(handles, timeout):
786n/a # Return ALL handles which are currently signalled. (Only
787n/a # returning the first signalled might create starvation issues.)
788n/a L = list(handles)
789n/a ready = []
790n/a while L:
791n/a res = _winapi.WaitForMultipleObjects(L, False, timeout)
792n/a if res == WAIT_TIMEOUT:
793n/a break
794n/a elif WAIT_OBJECT_0 <= res < WAIT_OBJECT_0 + len(L):
795n/a res -= WAIT_OBJECT_0
796n/a elif WAIT_ABANDONED_0 <= res < WAIT_ABANDONED_0 + len(L):
797n/a res -= WAIT_ABANDONED_0
798n/a else:
799n/a raise RuntimeError('Should not get here')
800n/a ready.append(L[res])
801n/a L = L[res+1:]
802n/a timeout = 0
803n/a return ready
804n/a
805n/a _ready_errors = {_winapi.ERROR_BROKEN_PIPE, _winapi.ERROR_NETNAME_DELETED}
806n/a
807n/a def wait(object_list, timeout=None):
808n/a '''
809n/a Wait till an object in object_list is ready/readable.
810n/a
811n/a Returns list of those objects in object_list which are ready/readable.
812n/a '''
813n/a if timeout is None:
814n/a timeout = INFINITE
815n/a elif timeout < 0:
816n/a timeout = 0
817n/a else:
818n/a timeout = int(timeout * 1000 + 0.5)
819n/a
820n/a object_list = list(object_list)
821n/a waithandle_to_obj = {}
822n/a ov_list = []
823n/a ready_objects = set()
824n/a ready_handles = set()
825n/a
826n/a try:
827n/a for o in object_list:
828n/a try:
829n/a fileno = getattr(o, 'fileno')
830n/a except AttributeError:
831n/a waithandle_to_obj[o.__index__()] = o
832n/a else:
833n/a # start an overlapped read of length zero
834n/a try:
835n/a ov, err = _winapi.ReadFile(fileno(), 0, True)
836n/a except OSError as e:
837n/a ov, err = None, e.winerror
838n/a if err not in _ready_errors:
839n/a raise
840n/a if err == _winapi.ERROR_IO_PENDING:
841n/a ov_list.append(ov)
842n/a waithandle_to_obj[ov.event] = o
843n/a else:
844n/a # If o.fileno() is an overlapped pipe handle and
845n/a # err == 0 then there is a zero length message
846n/a # in the pipe, but it HAS NOT been consumed...
847n/a if ov and sys.getwindowsversion()[:2] >= (6, 2):
848n/a # ... except on Windows 8 and later, where
849n/a # the message HAS been consumed.
850n/a try:
851n/a _, err = ov.GetOverlappedResult(False)
852n/a except OSError as e:
853n/a err = e.winerror
854n/a if not err and hasattr(o, '_got_empty_message'):
855n/a o._got_empty_message = True
856n/a ready_objects.add(o)
857n/a timeout = 0
858n/a
859n/a ready_handles = _exhaustive_wait(waithandle_to_obj.keys(), timeout)
860n/a finally:
861n/a # request that overlapped reads stop
862n/a for ov in ov_list:
863n/a ov.cancel()
864n/a
865n/a # wait for all overlapped reads to stop
866n/a for ov in ov_list:
867n/a try:
868n/a _, err = ov.GetOverlappedResult(True)
869n/a except OSError as e:
870n/a err = e.winerror
871n/a if err not in _ready_errors:
872n/a raise
873n/a if err != _winapi.ERROR_OPERATION_ABORTED:
874n/a o = waithandle_to_obj[ov.event]
875n/a ready_objects.add(o)
876n/a if err == 0:
877n/a # If o.fileno() is an overlapped pipe handle then
878n/a # a zero length message HAS been consumed.
879n/a if hasattr(o, '_got_empty_message'):
880n/a o._got_empty_message = True
881n/a
882n/a ready_objects.update(waithandle_to_obj[h] for h in ready_handles)
883n/a return [o for o in object_list if o in ready_objects]
884n/a
885n/aelse:
886n/a
887n/a import selectors
888n/a
889n/a # poll/select have the advantage of not requiring any extra file
890n/a # descriptor, contrarily to epoll/kqueue (also, they require a single
891n/a # syscall).
892n/a if hasattr(selectors, 'PollSelector'):
893n/a _WaitSelector = selectors.PollSelector
894n/a else:
895n/a _WaitSelector = selectors.SelectSelector
896n/a
897n/a def wait(object_list, timeout=None):
898n/a '''
899n/a Wait till an object in object_list is ready/readable.
900n/a
901n/a Returns list of those objects in object_list which are ready/readable.
902n/a '''
903n/a with _WaitSelector() as selector:
904n/a for obj in object_list:
905n/a selector.register(obj, selectors.EVENT_READ)
906n/a
907n/a if timeout is not None:
908n/a deadline = time.time() + timeout
909n/a
910n/a while True:
911n/a ready = selector.select(timeout)
912n/a if ready:
913n/a return [key.fileobj for (key, events) in ready]
914n/a else:
915n/a if timeout is not None:
916n/a timeout = deadline - time.time()
917n/a if timeout < 0:
918n/a return ready
919n/a
920n/a#
921n/a# Make connection and socket objects sharable if possible
922n/a#
923n/a
924n/aif sys.platform == 'win32':
925n/a def reduce_connection(conn):
926n/a handle = conn.fileno()
927n/a with socket.fromfd(handle, socket.AF_INET, socket.SOCK_STREAM) as s:
928n/a from . import resource_sharer
929n/a ds = resource_sharer.DupSocket(s)
930n/a return rebuild_connection, (ds, conn.readable, conn.writable)
931n/a def rebuild_connection(ds, readable, writable):
932n/a sock = ds.detach()
933n/a return Connection(sock.detach(), readable, writable)
934n/a reduction.register(Connection, reduce_connection)
935n/a
936n/a def reduce_pipe_connection(conn):
937n/a access = ((_winapi.FILE_GENERIC_READ if conn.readable else 0) |
938n/a (_winapi.FILE_GENERIC_WRITE if conn.writable else 0))
939n/a dh = reduction.DupHandle(conn.fileno(), access)
940n/a return rebuild_pipe_connection, (dh, conn.readable, conn.writable)
941n/a def rebuild_pipe_connection(dh, readable, writable):
942n/a handle = dh.detach()
943n/a return PipeConnection(handle, readable, writable)
944n/a reduction.register(PipeConnection, reduce_pipe_connection)
945n/a
946n/aelse:
947n/a def reduce_connection(conn):
948n/a df = reduction.DupFd(conn.fileno())
949n/a return rebuild_connection, (df, conn.readable, conn.writable)
950n/a def rebuild_connection(df, readable, writable):
951n/a fd = df.detach()
952n/a return Connection(fd, readable, writable)
953n/a reduction.register(Connection, reduce_connection)