| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039 |
- # -*- coding: utf-8 -*-
- #
- # A higher level module for using sockets (or Windows named pipes)
- #
- # multiprocessing/connection.py
- #
- # Copyright (c) 2006-2008, R Oudkerk
- # Licensed to PSF under a Contributor Agreement.
- #
- from __future__ import absolute_import
- import errno
- import io
- import os
- import sys
- import socket
- import select
- import struct
- import tempfile
- import itertools
- from . import reduction
- from . import util
- from . import AuthenticationError, BufferTooShort
- from ._ext import _billiard
- from .compat import setblocking, send_offset
- from .five import monotonic
- from .reduction import ForkingPickler
- try:
- from .compat import _winapi
- except ImportError:
- if sys.platform == 'win32':
- raise
- _winapi = None
- else:
- if sys.platform == 'win32':
- WAIT_OBJECT_0 = _winapi.WAIT_OBJECT_0
- try:
- WAIT_ABANDONED_0 = _winapi.WAIT_ABANDONED_0
- except AttributeError:
- WAIT_ABANDONED_0 = 128 # noqa
- WAIT_TIMEOUT = _winapi.WAIT_TIMEOUT
- INFINITE = _winapi.INFINITE
- __all__ = ['Client', 'Listener', 'Pipe', 'wait']
- is_pypy = hasattr(sys, 'pypy_version_info')
- #
- #
- #
- BUFSIZE = 8192
- # A very generous timeout when it comes to local connections...
- CONNECTION_TIMEOUT = 20.
- _mmap_counter = itertools.count()
- default_family = 'AF_INET'
- families = ['AF_INET']
- if hasattr(socket, 'AF_UNIX'):
- default_family = 'AF_UNIX'
- families += ['AF_UNIX']
- if sys.platform == 'win32':
- default_family = 'AF_PIPE'
- families += ['AF_PIPE']
- def _init_timeout(timeout=CONNECTION_TIMEOUT):
- return monotonic() + timeout
- def _check_timeout(t):
- return monotonic() > t
- #
- #
- #
- def arbitrary_address(family):
- '''
- Return an arbitrary free address for the given family
- '''
- if family == 'AF_INET':
- return ('localhost', 0)
- elif family == 'AF_UNIX':
- return tempfile.mktemp(prefix='listener-', dir=util.get_temp_dir())
- elif family == 'AF_PIPE':
- return tempfile.mktemp(prefix=r'\\.\pipe\pyc-%d-%d-' %
- (os.getpid(), next(_mmap_counter)), dir="")
- else:
- raise ValueError('unrecognized family')
- def _validate_family(family):
- '''
- Checks if the family is valid for the current environment.
- '''
- if sys.platform != 'win32' and family == 'AF_PIPE':
- raise ValueError('Family %s is not recognized.' % family)
- if sys.platform == 'win32' and family == 'AF_UNIX':
- # double check
- if not hasattr(socket, family):
- raise ValueError('Family %s is not recognized.' % family)
- def address_type(address):
- '''
- Return the types of the address
- This can be 'AF_INET', 'AF_UNIX', or 'AF_PIPE'
- '''
- if type(address) == tuple:
- return 'AF_INET'
- elif type(address) is str and address.startswith('\\\\'):
- return 'AF_PIPE'
- elif type(address) is str:
- return 'AF_UNIX'
- else:
- raise ValueError('address type of %r unrecognized' % address)
- #
- # Connection classes
- #
- class _SocketContainer(object):
- def __init__(self, sock):
- self.sock = sock
- class _ConnectionBase(object):
- _handle = None
- def __init__(self, handle, readable=True, writable=True):
- if isinstance(handle, _SocketContainer):
- self._socket = handle.sock # keep ref so not collected
- handle = handle.sock.fileno()
- handle = handle.__index__()
- if handle < 0:
- raise ValueError("invalid handle")
- if not readable and not writable:
- raise ValueError(
- "at least one of `readable` and `writable` must be True")
- self._handle = handle
- self._readable = readable
- self._writable = writable
- # XXX should we use util.Finalize instead of a __del__?
- def __del__(self):
- if self._handle is not None:
- self._close()
- def _check_closed(self):
- if self._handle is None:
- raise OSError("handle is closed")
- def _check_readable(self):
- if not self._readable:
- raise OSError("connection is write-only")
- def _check_writable(self):
- if not self._writable:
- raise OSError("connection is read-only")
- def _bad_message_length(self):
- if self._writable:
- self._readable = False
- else:
- self.close()
- raise OSError("bad message length")
- @property
- def closed(self):
- """True if the connection is closed"""
- return self._handle is None
- @property
- def readable(self):
- """True if the connection is readable"""
- return self._readable
- @property
- def writable(self):
- """True if the connection is writable"""
- return self._writable
- def fileno(self):
- """File descriptor or handle of the connection"""
- self._check_closed()
- return self._handle
- def close(self):
- """Close the connection"""
- if self._handle is not None:
- try:
- self._close()
- finally:
- self._handle = None
- def send_bytes(self, buf, offset=0, size=None):
- """Send the bytes data from a bytes-like object"""
- self._check_closed()
- self._check_writable()
- m = memoryview(buf)
- # HACK for byte-indexing of non-bytewise buffers (e.g. array.array)
- if m.itemsize > 1:
- m = memoryview(bytes(m))
- n = len(m)
- if offset < 0:
- raise ValueError("offset is negative")
- if n < offset:
- raise ValueError("buffer length < offset")
- if size is None:
- size = n - offset
- elif size < 0:
- raise ValueError("size is negative")
- elif offset + size > n:
- raise ValueError("buffer length < offset + size")
- self._send_bytes(m[offset:offset + size])
- def send(self, obj):
- """Send a (picklable) object"""
- self._check_closed()
- self._check_writable()
- self._send_bytes(ForkingPickler.dumps(obj))
- def recv_bytes(self, maxlength=None):
- """
- Receive bytes data as a bytes object.
- """
- self._check_closed()
- self._check_readable()
- if maxlength is not None and maxlength < 0:
- raise ValueError("negative maxlength")
- buf = self._recv_bytes(maxlength)
- if buf is None:
- self._bad_message_length()
- return buf.getvalue()
- def recv_bytes_into(self, buf, offset=0):
- """
- Receive bytes data into a writeable bytes-like object.
- Return the number of bytes read.
- """
- self._check_closed()
- self._check_readable()
- with memoryview(buf) as m:
- # Get bytesize of arbitrary buffer
- itemsize = m.itemsize
- bytesize = itemsize * len(m)
- if offset < 0:
- raise ValueError("negative offset")
- elif offset > bytesize:
- raise ValueError("offset too large")
- result = self._recv_bytes()
- size = result.tell()
- if bytesize < offset + size:
- raise BufferTooShort(result.getvalue())
- # Message can fit in dest
- result.seek(0)
- result.readinto(m[
- offset // itemsize:(offset + size) // itemsize
- ])
- return size
- def recv(self):
- """Receive a (picklable) object"""
- self._check_closed()
- self._check_readable()
- buf = self._recv_bytes()
- return ForkingPickler.loadbuf(buf)
- def poll(self, timeout=0.0):
- """Whether there is any input available to be read"""
- self._check_closed()
- self._check_readable()
- return self._poll(timeout)
- def __enter__(self):
- return self
- def __exit__(self, exc_type, exc_value, exc_tb):
- self.close()
- def send_offset(self, buf, offset):
- return send_offset(self.fileno(), buf, offset)
- def setblocking(self, blocking):
- setblocking(self.fileno(), blocking)
- if _winapi:
- class PipeConnection(_ConnectionBase):
- """
- Connection class based on a Windows named pipe.
- Overlapped I/O is used, so the handles must have been created
- with FILE_FLAG_OVERLAPPED.
- """
- _got_empty_message = False
- def _close(self, _CloseHandle=_winapi.CloseHandle):
- _CloseHandle(self._handle)
- def _send_bytes(self, buf):
- ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
- try:
- if err == _winapi.ERROR_IO_PENDING:
- waitres = _winapi.WaitForMultipleObjects(
- [ov.event], False, INFINITE)
- assert waitres == WAIT_OBJECT_0
- except:
- ov.cancel()
- raise
- finally:
- nwritten, err = ov.GetOverlappedResult(True)
- assert err == 0
- assert nwritten == len(buf)
- def _recv_bytes(self, maxsize=None):
- if self._got_empty_message:
- self._got_empty_message = False
- return io.BytesIO()
- else:
- bsize = 128 if maxsize is None else min(maxsize, 128)
- try:
- ov, err = _winapi.ReadFile(
- self._handle, bsize, overlapped=True,
- )
- try:
- if err == _winapi.ERROR_IO_PENDING:
- waitres = _winapi.WaitForMultipleObjects(
- [ov.event], False, INFINITE)
- assert waitres == WAIT_OBJECT_0
- except:
- ov.cancel()
- raise
- finally:
- nread, err = ov.GetOverlappedResult(True)
- if err == 0:
- f = io.BytesIO()
- f.write(ov.getbuffer())
- return f
- elif err == _winapi.ERROR_MORE_DATA:
- return self._get_more_data(ov, maxsize)
- except OSError as e:
- if e.winerror == _winapi.ERROR_BROKEN_PIPE:
- raise EOFError
- else:
- raise
- raise RuntimeError(
- "shouldn't get here; expected KeyboardInterrupt")
- def _poll(self, timeout):
- if (self._got_empty_message or
- _winapi.PeekNamedPipe(self._handle)[0] != 0):
- return True
- return bool(wait([self], timeout))
- def _get_more_data(self, ov, maxsize):
- buf = ov.getbuffer()
- f = io.BytesIO()
- f.write(buf)
- left = _winapi.PeekNamedPipe(self._handle)[1]
- assert left > 0
- if maxsize is not None and len(buf) + left > maxsize:
- self._bad_message_length()
- ov, err = _winapi.ReadFile(self._handle, left, overlapped=True)
- rbytes, err = ov.GetOverlappedResult(True)
- assert err == 0
- assert rbytes == left
- f.write(ov.getbuffer())
- return f
- class Connection(_ConnectionBase):
- """
- Connection class based on an arbitrary file descriptor (Unix only), or
- a socket handle (Windows).
- """
- if _winapi:
- def _close(self, _close=_billiard.closesocket):
- _close(self._handle)
- _write = _billiard.send
- _read = _billiard.recv
- else:
- def _close(self, _close=os.close):
- _close(self._handle)
- _write = os.write
- _read = os.read
- def _send(self, buf, write=_write):
- remaining = len(buf)
- while True:
- try:
- n = write(self._handle, buf)
- except (OSError, IOError, socket.error) as exc:
- if getattr(exc, 'errno', None) != errno.EINTR:
- raise
- else:
- remaining -= n
- if remaining == 0:
- break
- buf = buf[n:]
- def _recv(self, size, read=_read):
- buf = io.BytesIO()
- handle = self._handle
- remaining = size
- while remaining > 0:
- try:
- chunk = read(handle, remaining)
- except (OSError, IOError, socket.error) as exc:
- if getattr(exc, 'errno', None) != errno.EINTR:
- raise
- else:
- n = len(chunk)
- if n == 0:
- if remaining == size:
- raise EOFError
- else:
- raise OSError("got end of file during message")
- buf.write(chunk)
- remaining -= n
- return buf
- def _send_bytes(self, buf, memoryview=memoryview):
- n = len(buf)
- # For wire compatibility with 3.2 and lower
- header = struct.pack("!i", n)
- if n > 16384:
- # The payload is large so Nagle's algorithm won't be triggered
- # and we'd better avoid the cost of concatenation.
- self._send(header)
- self._send(buf)
- else:
- # Issue #20540: concatenate before sending, to avoid delays due
- # to Nagle's algorithm on a TCP socket.
- # Also note we want to avoid sending a 0-length buffer separately,
- # to avoid "broken pipe" errors if the other end closed the pipe.
- if isinstance(buf, memoryview):
- buf = buf.tobytes()
- self._send(header + buf)
- def _recv_bytes(self, maxsize=None):
- buf = self._recv(4)
- size, = struct.unpack("!i", buf.getvalue())
- if maxsize is not None and size > maxsize:
- return None
- return self._recv(size)
- def _poll(self, timeout):
- r = wait([self], timeout)
- return bool(r)
- #
- # Public functions
- #
- class Listener(object):
- '''
- Returns a listener object.
- This is a wrapper for a bound socket which is 'listening' for
- connections, or for a Windows named pipe.
- '''
- def __init__(self, address=None, family=None, backlog=1, authkey=None):
- family = (family or
- (address and address_type(address)) or default_family)
- address = address or arbitrary_address(family)
- _validate_family(family)
- if family == 'AF_PIPE':
- self._listener = PipeListener(address, backlog)
- else:
- self._listener = SocketListener(address, family, backlog)
- if authkey is not None and not isinstance(authkey, bytes):
- raise TypeError('authkey should be a byte string')
- self._authkey = authkey
- def accept(self):
- '''
- Accept a connection on the bound socket or named pipe of `self`.
- Returns a `Connection` object.
- '''
- if self._listener is None:
- raise OSError('listener is closed')
- c = self._listener.accept()
- if self._authkey:
- deliver_challenge(c, self._authkey)
- answer_challenge(c, self._authkey)
- return c
- def close(self):
- '''
- Close the bound socket or named pipe of `self`.
- '''
- listener = self._listener
- if listener is not None:
- self._listener = None
- listener.close()
- address = property(lambda self: self._listener._address)
- last_accepted = property(lambda self: self._listener._last_accepted)
- def __enter__(self):
- return self
- def __exit__(self, exc_type, exc_value, exc_tb):
- self.close()
- def Client(address, family=None, authkey=None):
- '''
- Returns a connection to the address of a `Listener`
- '''
- family = family or address_type(address)
- _validate_family(family)
- if family == 'AF_PIPE':
- c = PipeClient(address)
- else:
- c = SocketClient(address)
- if authkey is not None and not isinstance(authkey, bytes):
- raise TypeError('authkey should be a byte string')
- if authkey is not None:
- answer_challenge(c, authkey)
- deliver_challenge(c, authkey)
- return c
- def detach(sock):
- if hasattr(sock, 'detach'):
- return sock.detach()
- # older socket lib does not have detach. We'll keep a reference around
- # so that it does not get garbage collected.
- return _SocketContainer(sock)
- if sys.platform != 'win32':
- def Pipe(duplex=True, rnonblock=False, wnonblock=False):
- '''
- Returns pair of connection objects at either end of a pipe
- '''
- if duplex:
- s1, s2 = socket.socketpair()
- s1.setblocking(not rnonblock)
- s2.setblocking(not wnonblock)
- c1 = Connection(detach(s1))
- c2 = Connection(detach(s2))
- else:
- fd1, fd2 = os.pipe()
- if rnonblock:
- setblocking(fd1, 0)
- if wnonblock:
- setblocking(fd2, 0)
- c1 = Connection(fd1, writable=False)
- c2 = Connection(fd2, readable=False)
- return c1, c2
- else:
- def Pipe(duplex=True, rnonblock=False, wnonblock=False):
- '''
- Returns pair of connection objects at either end of a pipe
- '''
- assert not rnonblock, 'rnonblock not supported on windows'
- assert not wnonblock, 'wnonblock not supported on windows'
- address = arbitrary_address('AF_PIPE')
- if duplex:
- openmode = _winapi.PIPE_ACCESS_DUPLEX
- access = _winapi.GENERIC_READ | _winapi.GENERIC_WRITE
- obsize, ibsize = BUFSIZE, BUFSIZE
- else:
- openmode = _winapi.PIPE_ACCESS_INBOUND
- access = _winapi.GENERIC_WRITE
- obsize, ibsize = 0, BUFSIZE
- h1 = _winapi.CreateNamedPipe(
- address, openmode | _winapi.FILE_FLAG_OVERLAPPED |
- _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE,
- _winapi.PIPE_TYPE_MESSAGE | _winapi.PIPE_READMODE_MESSAGE |
- _winapi.PIPE_WAIT,
- 1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER,
- # default security descriptor: the handle cannot be inherited
- _winapi.NULL
- )
- h2 = _winapi.CreateFile(
- address, access, 0, _winapi.NULL, _winapi.OPEN_EXISTING,
- _winapi.FILE_FLAG_OVERLAPPED, _winapi.NULL
- )
- _winapi.SetNamedPipeHandleState(
- h2, _winapi.PIPE_READMODE_MESSAGE, None, None
- )
- overlapped = _winapi.ConnectNamedPipe(h1, overlapped=True)
- _, err = overlapped.GetOverlappedResult(True)
- assert err == 0
- c1 = PipeConnection(h1, writable=duplex)
- c2 = PipeConnection(h2, readable=duplex)
- return c1, c2
- #
- # Definitions for connections based on sockets
- #
- class SocketListener(object):
- '''
- Representation of a socket which is bound to an address and listening
- '''
- def __init__(self, address, family, backlog=1):
- self._socket = socket.socket(getattr(socket, family))
- try:
- # SO_REUSEADDR has different semantics on Windows (issue #2550).
- if os.name == 'posix':
- self._socket.setsockopt(socket.SOL_SOCKET,
- socket.SO_REUSEADDR, 1)
- self._socket.setblocking(True)
- self._socket.bind(address)
- self._socket.listen(backlog)
- self._address = self._socket.getsockname()
- except OSError:
- self._socket.close()
- raise
- self._family = family
- self._last_accepted = None
- if family == 'AF_UNIX':
- self._unlink = util.Finalize(
- self, os.unlink, args=(address,), exitpriority=0
- )
- else:
- self._unlink = None
- def accept(self):
- while True:
- try:
- s, self._last_accepted = self._socket.accept()
- except (OSError, IOError, socket.error) as exc:
- if getattr(exc, 'errno', None) != errno.EINTR:
- raise
- else:
- break
- s.setblocking(True)
- return Connection(detach(s))
- def close(self):
- try:
- self._socket.close()
- finally:
- unlink = self._unlink
- if unlink is not None:
- self._unlink = None
- unlink()
- def SocketClient(address):
- '''
- Return a connection object connected to the socket given by `address`
- '''
- family = address_type(address)
- s = socket.socket(getattr(socket, family))
- s.setblocking(True)
- s.connect(address)
- return Connection(detach(s))
- #
- # Definitions for connections based on named pipes
- #
- if sys.platform == 'win32':
- class PipeListener(object):
- '''
- Representation of a named pipe
- '''
- def __init__(self, address, backlog=None):
- self._address = address
- self._handle_queue = [self._new_handle(first=True)]
- self._last_accepted = None
- util.sub_debug('listener created with address=%r', self._address)
- self.close = util.Finalize(
- self, PipeListener._finalize_pipe_listener,
- args=(self._handle_queue, self._address), exitpriority=0
- )
- def _new_handle(self, first=False):
- flags = _winapi.PIPE_ACCESS_DUPLEX | _winapi.FILE_FLAG_OVERLAPPED
- if first:
- flags |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE
- return _winapi.CreateNamedPipe(
- self._address, flags,
- _winapi.PIPE_TYPE_MESSAGE | _winapi.PIPE_READMODE_MESSAGE |
- _winapi.PIPE_WAIT,
- _winapi.PIPE_UNLIMITED_INSTANCES, BUFSIZE, BUFSIZE,
- _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL
- )
- def accept(self):
- self._handle_queue.append(self._new_handle())
- handle = self._handle_queue.pop(0)
- try:
- ov = _winapi.ConnectNamedPipe(handle, overlapped=True)
- except OSError as e:
- if e.winerror != _winapi.ERROR_NO_DATA:
- raise
- # ERROR_NO_DATA can occur if a client has already connected,
- # written data and then disconnected -- see Issue 14725.
- else:
- try:
- _winapi.WaitForMultipleObjects(
- [ov.event], False, INFINITE)
- except:
- ov.cancel()
- _winapi.CloseHandle(handle)
- raise
- finally:
- _, err = ov.GetOverlappedResult(True)
- assert err == 0
- return PipeConnection(handle)
- @staticmethod
- def _finalize_pipe_listener(queue, address):
- util.sub_debug('closing listener with address=%r', address)
- for handle in queue:
- _winapi.CloseHandle(handle)
- def PipeClient(address, _ignore=(_winapi.ERROR_SEM_TIMEOUT,
- _winapi.ERROR_PIPE_BUSY)):
- '''
- Return a connection object connected to the pipe given by `address`
- '''
- t = _init_timeout()
- while 1:
- try:
- _winapi.WaitNamedPipe(address, 1000)
- h = _winapi.CreateFile(
- address, _winapi.GENERIC_READ | _winapi.GENERIC_WRITE,
- 0, _winapi.NULL, _winapi.OPEN_EXISTING,
- _winapi.FILE_FLAG_OVERLAPPED, _winapi.NULL
- )
- except OSError as e:
- if e.winerror not in _ignore or _check_timeout(t):
- raise
- else:
- break
- else:
- raise
- _winapi.SetNamedPipeHandleState(
- h, _winapi.PIPE_READMODE_MESSAGE, None, None
- )
- return PipeConnection(h)
- #
- # Authentication stuff
- #
- MESSAGE_LENGTH = 20
- CHALLENGE = b'#CHALLENGE#'
- WELCOME = b'#WELCOME#'
- FAILURE = b'#FAILURE#'
- def deliver_challenge(connection, authkey):
- import hmac
- assert isinstance(authkey, bytes)
- message = os.urandom(MESSAGE_LENGTH)
- connection.send_bytes(CHALLENGE + message)
- digest = hmac.new(authkey, message, 'md5').digest()
- response = connection.recv_bytes(256) # reject large message
- if response == digest:
- connection.send_bytes(WELCOME)
- else:
- connection.send_bytes(FAILURE)
- raise AuthenticationError('digest received was wrong')
- def answer_challenge(connection, authkey):
- import hmac
- assert isinstance(authkey, bytes)
- message = connection.recv_bytes(256) # reject large message
- assert message[:len(CHALLENGE)] == CHALLENGE, 'message = %r' % message
- message = message[len(CHALLENGE):]
- digest = hmac.new(authkey, message, 'md5').digest()
- connection.send_bytes(digest)
- response = connection.recv_bytes(256) # reject large message
- if response != WELCOME:
- raise AuthenticationError('digest sent was rejected')
- #
- # Support for using xmlrpclib for serialization
- #
- class ConnectionWrapper(object):
- def __init__(self, conn, dumps, loads):
- self._conn = conn
- self._dumps = dumps
- self._loads = loads
- for attr in ('fileno', 'close', 'poll', 'recv_bytes', 'send_bytes'):
- obj = getattr(conn, attr)
- setattr(self, attr, obj)
- def send(self, obj):
- s = self._dumps(obj)
- self._conn.send_bytes(s)
- def recv(self):
- s = self._conn.recv_bytes()
- return self._loads(s)
- def _xml_dumps(obj):
- o = xmlrpclib.dumps((obj, ), None, None, None, 1) # noqa
- return o.encode('utf-8')
- def _xml_loads(s):
- (obj,), method = xmlrpclib.loads(s.decode('utf-8')) # noqa
- return obj
- class XmlListener(Listener):
- def accept(self):
- global xmlrpclib
- import xmlrpc.client as xmlrpclib # noqa
- obj = Listener.accept(self)
- return ConnectionWrapper(obj, _xml_dumps, _xml_loads)
- def XmlClient(*args, **kwds):
- global xmlrpclib
- import xmlrpc.client as xmlrpclib # noqa
- return ConnectionWrapper(Client(*args, **kwds), _xml_dumps, _xml_loads)
- #
- # Wait
- #
- if sys.platform == 'win32':
- def _exhaustive_wait(handles, timeout):
- # Return ALL handles which are currently signaled. (Only
- # returning the first signaled might create starvation issues.)
- L = list(handles)
- ready = []
- while L:
- res = _winapi.WaitForMultipleObjects(L, False, timeout)
- if res == WAIT_TIMEOUT:
- break
- elif WAIT_OBJECT_0 <= res < WAIT_OBJECT_0 + len(L):
- res -= WAIT_OBJECT_0
- elif WAIT_ABANDONED_0 <= res < WAIT_ABANDONED_0 + len(L):
- res -= WAIT_ABANDONED_0
- else:
- raise RuntimeError('Should not get here')
- ready.append(L[res])
- L = L[res + 1:]
- timeout = 0
- return ready
- _ready_errors = {_winapi.ERROR_BROKEN_PIPE, _winapi.ERROR_NETNAME_DELETED}
- def wait(object_list, timeout=None):
- '''
- Wait till an object in object_list is ready/readable.
- Returns list of those objects in object_list which are ready/readable.
- '''
- if timeout is None:
- timeout = INFINITE
- elif timeout < 0:
- timeout = 0
- else:
- timeout = int(timeout * 1000 + 0.5)
- object_list = list(object_list)
- waithandle_to_obj = {}
- ov_list = []
- ready_objects = set()
- ready_handles = set()
- try:
- for o in object_list:
- try:
- fileno = getattr(o, 'fileno')
- except AttributeError:
- waithandle_to_obj[o.__index__()] = o
- else:
- # start an overlapped read of length zero
- try:
- ov, err = _winapi.ReadFile(fileno(), 0, True)
- except OSError as e:
- err = e.winerror
- if err not in _ready_errors:
- raise
- if err == _winapi.ERROR_IO_PENDING:
- ov_list.append(ov)
- waithandle_to_obj[ov.event] = o
- else:
- # If o.fileno() is an overlapped pipe handle and
- # err == 0 then there is a zero length message
- # in the pipe, but it HAS NOT been consumed.
- ready_objects.add(o)
- timeout = 0
- ready_handles = _exhaustive_wait(waithandle_to_obj.keys(), timeout)
- finally:
- # request that overlapped reads stop
- for ov in ov_list:
- ov.cancel()
- # wait for all overlapped reads to stop
- for ov in ov_list:
- try:
- _, err = ov.GetOverlappedResult(True)
- except OSError as e:
- err = e.winerror
- if err not in _ready_errors:
- raise
- if err != _winapi.ERROR_OPERATION_ABORTED:
- o = waithandle_to_obj[ov.event]
- ready_objects.add(o)
- if err == 0:
- # If o.fileno() is an overlapped pipe handle then
- # a zero length message HAS been consumed.
- if hasattr(o, '_got_empty_message'):
- o._got_empty_message = True
- ready_objects.update(waithandle_to_obj[h] for h in ready_handles)
- return [p for p in object_list if p in ready_objects]
- else:
- if hasattr(select, 'poll'):
- def _poll(fds, timeout):
- if timeout is not None:
- timeout = int(timeout * 1000) # timeout is in milliseconds
- fd_map = {}
- pollster = select.poll()
- for fd in fds:
- pollster.register(fd, select.POLLIN)
- if hasattr(fd, 'fileno'):
- fd_map[fd.fileno()] = fd
- else:
- fd_map[fd] = fd
- ls = []
- for fd, event in pollster.poll(timeout):
- if event & select.POLLNVAL:
- raise ValueError('invalid file descriptor %i' % fd)
- ls.append(fd_map[fd])
- return ls
- else:
- def _poll(fds, timeout): # noqa
- return select.select(fds, [], [], timeout)[0]
- def wait(object_list, timeout=None): # noqa
- '''
- Wait till an object in object_list is ready/readable.
- Returns list of those objects in object_list which are ready/readable.
- '''
- if timeout is not None:
- if timeout <= 0:
- return _poll(object_list, 0)
- else:
- deadline = monotonic() + timeout
- while True:
- try:
- return _poll(object_list, timeout)
- except (OSError, IOError, socket.error) as e:
- if e.errno != errno.EINTR:
- raise
- if timeout is not None:
- timeout = deadline - monotonic()
- #
- # Make connection and socket objects sharable if possible
- #
- if sys.platform == 'win32':
- def reduce_connection(conn):
- handle = conn.fileno()
- with socket.fromfd(handle, socket.AF_INET, socket.SOCK_STREAM) as s:
- from . import resource_sharer
- ds = resource_sharer.DupSocket(s)
- return rebuild_connection, (ds, conn.readable, conn.writable)
- def rebuild_connection(ds, readable, writable):
- sock = ds.detach()
- return Connection(detach(sock), readable, writable)
- reduction.register(Connection, reduce_connection)
- def reduce_pipe_connection(conn):
- access = ((_winapi.FILE_GENERIC_READ if conn.readable else 0) |
- (_winapi.FILE_GENERIC_WRITE if conn.writable else 0))
- dh = reduction.DupHandle(conn.fileno(), access)
- return rebuild_pipe_connection, (dh, conn.readable, conn.writable)
- def rebuild_pipe_connection(dh, readable, writable):
- return PipeConnection(detach(dh), readable, writable)
- reduction.register(PipeConnection, reduce_pipe_connection)
- else:
- def reduce_connection(conn):
- df = reduction.DupFd(conn.fileno())
- return rebuild_connection, (df, conn.readable, conn.writable)
- def rebuild_connection(df, readable, writable):
- return Connection(detach(df), readable, writable)
- reduction.register(Connection, reduce_connection)
|