Merged revisions 64104,64117 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/trunk
........
r64104 | benjamin.peterson | 2008-06-10 21:40:25 -0500 (Tue, 10 Jun 2008) | 2 lines
add the multiprocessing package to fulfill PEP 371
........
r64117 | benjamin.peterson | 2008-06-11 07:26:31 -0500 (Wed, 11 Jun 2008) | 2 lines
fix import of multiprocessing by juggling imports
........
diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py
new file mode 100644
index 0000000..27c44cd
--- /dev/null
+++ b/Lib/multiprocessing/connection.py
@@ -0,0 +1,425 @@
+#
+# A higher level module for using sockets (or Windows named pipes)
+#
+# multiprocessing/connection.py
+#
+# Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
+#
+
+__all__ = [ 'Client', 'Listener', 'Pipe' ]
+
+import os
+import sys
+import socket
+import time
+import tempfile
+import itertools
+
+import _multiprocessing
+from multiprocessing import current_process
+from multiprocessing.util import get_temp_dir, Finalize, sub_debug, debug
+from multiprocessing.forking import duplicate, close
+
+
+#
+#
+#
+
+BUFSIZE = 8192
+
+_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 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=get_temp_dir())
+ elif family == 'AF_PIPE':
+ return tempfile.mktemp(prefix=r'\\.\pipe\pyc-%d-%d-' %
+ (os.getpid(), next(_mmap_counter)))
+ else:
+ raise ValueError('unrecognized 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)
+
+#
+# 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)
+
+ 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.
+ '''
+ 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`.
+ '''
+ return self._listener.close()
+
+ address = property(lambda self: self._listener._address)
+ last_accepted = property(lambda self: self._listener._last_accepted)
+
+
+def Client(address, family=None, authkey=None):
+ '''
+ Returns a connection to the address of a `Listener`
+ '''
+ family = family or address_type(address)
+ 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
+
+
+if sys.platform != 'win32':
+
+ def Pipe(duplex=True):
+ '''
+ Returns pair of connection objects at either end of a pipe
+ '''
+ if duplex:
+ s1, s2 = socket.socketpair()
+ c1 = _multiprocessing.Connection(os.dup(s1.fileno()))
+ c2 = _multiprocessing.Connection(os.dup(s2.fileno()))
+ s1.close()
+ s2.close()
+ else:
+ fd1, fd2 = os.pipe()
+ c1 = _multiprocessing.Connection(fd1, writable=False)
+ c2 = _multiprocessing.Connection(fd2, readable=False)
+
+ return c1, c2
+
+else:
+
+ from ._multiprocessing import win32
+
+ def Pipe(duplex=True):
+ '''
+ Returns pair of connection objects at either end of a pipe
+ '''
+ address = arbitrary_address('AF_PIPE')
+ if duplex:
+ openmode = win32.PIPE_ACCESS_DUPLEX
+ access = win32.GENERIC_READ | win32.GENERIC_WRITE
+ obsize, ibsize = BUFSIZE, BUFSIZE
+ else:
+ openmode = win32.PIPE_ACCESS_INBOUND
+ access = win32.GENERIC_WRITE
+ obsize, ibsize = 0, BUFSIZE
+
+ h1 = win32.CreateNamedPipe(
+ address, openmode,
+ win32.PIPE_TYPE_MESSAGE | win32.PIPE_READMODE_MESSAGE |
+ win32.PIPE_WAIT,
+ 1, obsize, ibsize, win32.NMPWAIT_WAIT_FOREVER, win32.NULL
+ )
+ h2 = win32.CreateFile(
+ address, access, 0, win32.NULL, win32.OPEN_EXISTING, 0, win32.NULL
+ )
+ win32.SetNamedPipeHandleState(
+ h2, win32.PIPE_READMODE_MESSAGE, None, None
+ )
+
+ try:
+ win32.ConnectNamedPipe(h1, win32.NULL)
+ except WindowsError as e:
+ if e.args[0] != win32.ERROR_PIPE_CONNECTED:
+ raise
+
+ c1 = _multiprocessing.PipeConnection(h1, writable=duplex)
+ c2 = _multiprocessing.PipeConnection(h2, readable=duplex)
+
+ return c1, c2
+
+#
+# Definitions for connections based on sockets
+#
+
+class SocketListener(object):
+ '''
+ Represtation 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))
+ self._socket.bind(address)
+ self._socket.listen(backlog)
+ address = self._socket.getsockname()
+ if type(address) is tuple:
+ address = (socket.getfqdn(address[0]),) + address[1:]
+ self._address = address
+ self._family = family
+ self._last_accepted = None
+
+ sub_debug('listener bound to address %r', self._address)
+
+ if family == 'AF_UNIX':
+ self._unlink = Finalize(
+ self, os.unlink, args=(self._address,), exitpriority=0
+ )
+ else:
+ self._unlink = None
+
+ def accept(self):
+ s, self._last_accepted = self._socket.accept()
+ fd = duplicate(s.fileno())
+ conn = _multiprocessing.Connection(fd)
+ s.close()
+ return conn
+
+ def close(self):
+ self._socket.close()
+ if self._unlink is not None:
+ self._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) )
+
+ while 1:
+ try:
+ s.connect(address)
+ except socket.error as e:
+ if e.args[0] != 10061: # 10061 => connection refused
+ debug('failed to connect to address %s', address)
+ raise
+ time.sleep(0.01)
+ else:
+ break
+ else:
+ raise
+
+ fd = duplicate(s.fileno())
+ conn = _multiprocessing.Connection(fd)
+ s.close()
+ return conn
+
+#
+# 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
+ handle = win32.CreateNamedPipe(
+ address, win32.PIPE_ACCESS_DUPLEX,
+ win32.PIPE_TYPE_MESSAGE | win32.PIPE_READMODE_MESSAGE |
+ win32.PIPE_WAIT,
+ win32.PIPE_UNLIMITED_INSTANCES, BUFSIZE, BUFSIZE,
+ win32.NMPWAIT_WAIT_FOREVER, win32.NULL
+ )
+ self._handle_queue = [handle]
+ self._last_accepted = None
+
+ sub_debug('listener created with address=%r', self._address)
+
+ self.close = Finalize(
+ self, PipeListener._finalize_pipe_listener,
+ args=(self._handle_queue, self._address), exitpriority=0
+ )
+
+ def accept(self):
+ newhandle = win32.CreateNamedPipe(
+ self._address, win32.PIPE_ACCESS_DUPLEX,
+ win32.PIPE_TYPE_MESSAGE | win32.PIPE_READMODE_MESSAGE |
+ win32.PIPE_WAIT,
+ win32.PIPE_UNLIMITED_INSTANCES, BUFSIZE, BUFSIZE,
+ win32.NMPWAIT_WAIT_FOREVER, win32.NULL
+ )
+ self._handle_queue.append(newhandle)
+ handle = self._handle_queue.pop(0)
+ try:
+ win32.ConnectNamedPipe(handle, win32.NULL)
+ except WindowsError as e:
+ if e.args[0] != win32.ERROR_PIPE_CONNECTED:
+ raise
+ return _multiprocessing.PipeConnection(handle)
+
+ @staticmethod
+ def _finalize_pipe_listener(queue, address):
+ sub_debug('closing listener with address=%r', address)
+ for handle in queue:
+ close(handle)
+
+ def PipeClient(address):
+ '''
+ Return a connection object connected to the pipe given by `address`
+ '''
+ while 1:
+ try:
+ win32.WaitNamedPipe(address, 1000)
+ h = win32.CreateFile(
+ address, win32.GENERIC_READ | win32.GENERIC_WRITE,
+ 0, win32.NULL, win32.OPEN_EXISTING, 0, win32.NULL
+ )
+ except WindowsError as e:
+ if e.args[0] not in (win32.ERROR_SEM_TIMEOUT,
+ win32.ERROR_PIPE_BUSY):
+ raise
+ else:
+ break
+ else:
+ raise
+
+ win32.SetNamedPipeHandleState(
+ h, win32.PIPE_READMODE_MESSAGE, None, None
+ )
+ return _multiprocessing.PipeConnection(h)
+
+#
+# Authentication stuff
+#
+
+MESSAGE_LENGTH = 20
+
+CHALLENGE = '#CHALLENGE#'
+WELCOME = '#WELCOME#'
+FAILURE = '#FAILURE#'
+
+if sys.version_info >= (3, 0): # XXX can use bytes literals in 2.6/3.0
+ CHALLENGE = CHALLENGE.encode('ascii')
+ WELCOME = WELCOME.encode('ascii')
+ FAILURE = FAILURE.encode('ascii')
+
+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).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).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):
+ return xmlrpclib.dumps((obj,), None, None, None, 1).encode('utf8')
+
+def _xml_loads(s):
+ (obj,), method = xmlrpclib.loads(s.decode('utf8'))
+ return obj
+
+class XmlListener(Listener):
+ def accept(self):
+ global xmlrpclib
+ import xmlrpc.client as xmlrpclib
+ obj = Listener.accept(self)
+ return ConnectionWrapper(obj, _xml_dumps, _xml_loads)
+
+def XmlClient(*args, **kwds):
+ global xmlrpclib
+ import xmlrpc.client as xmlrpclib
+ return ConnectionWrapper(Client(*args, **kwds), _xml_dumps, _xml_loads)