Fri Jun 13 21:13:41 CEST 2008

Author: benjamin.peterson
Date: Fri Jun 13 21:13:39 2008
New Revision: 64248
Log:
convert multiprocessing to unix line endings
Modified:
 python/trunk/Lib/multiprocessing/__init__.py
 python/trunk/Lib/multiprocessing/connection.py
 python/trunk/Lib/multiprocessing/dummy/__init__.py
 python/trunk/Lib/multiprocessing/dummy/connection.py
 python/trunk/Lib/multiprocessing/forking.py
 python/trunk/Lib/multiprocessing/heap.py
 python/trunk/Lib/multiprocessing/managers.py
 python/trunk/Lib/multiprocessing/pool.py
 python/trunk/Lib/multiprocessing/process.py
 python/trunk/Lib/multiprocessing/queues.py
 python/trunk/Lib/multiprocessing/reduction.py
 python/trunk/Lib/multiprocessing/sharedctypes.py
 python/trunk/Lib/multiprocessing/synchronize.py
 python/trunk/Lib/multiprocessing/util.py
 python/trunk/Lib/test/test_multiprocessing.py
 python/trunk/Modules/_multiprocessing/multiprocessing.c
 python/trunk/Modules/_multiprocessing/multiprocessing.h
 python/trunk/Modules/_multiprocessing/pipe_connection.c
 python/trunk/Modules/_multiprocessing/win32_functions.c
Modified: python/trunk/Lib/multiprocessing/__init__.py
==============================================================================

--- python/trunk/Lib/multiprocessing/__init__.py	(original)
+++ python/trunk/Lib/multiprocessing/__init__.py	Fri Jun 13 21:13:39 2008
@@ -68,10 +68,10 @@
 
 class ProcessError(Exception):
 pass
- 
+
 class BufferTooShort(ProcessError):
 pass
- 
+
 class TimeoutError(ProcessError):
 pass
 
@@ -123,7 +123,7 @@
 num = os.sysconf('SC_NPROCESSORS_ONLN')
 except (ValueError, OSError, AttributeError):
 num = 0
- 
+
 if num >= 1:
 return num
 else:
@@ -151,13 +151,13 @@
 '''
 from multiprocessing.util import log_to_stderr
 return log_to_stderr(level)
- 
+
 def allow_connection_pickling():
 '''
 Install support for sending connections and sockets between processes
 '''
 from multiprocessing import reduction
- 
+
 #
 # Definitions depending on native semaphores
 #
@@ -263,7 +263,7 @@
 '''
 Sets the path to a python.exe or pythonw.exe binary used to run
 child processes on Windows instead of sys.executable.
- Useful for people embedding Python. 
+ Useful for people embedding Python.
 '''
 from multiprocessing.forking import set_executable
 set_executable(executable)
Modified: python/trunk/Lib/multiprocessing/connection.py
==============================================================================
--- python/trunk/Lib/multiprocessing/connection.py	(original)
+++ python/trunk/Lib/multiprocessing/connection.py	Fri Jun 13 21:13:39 2008
@@ -50,7 +50,7 @@
 '''
 if family == 'AF_INET':
 return ('localhost', 0)
- elif family == 'AF_UNIX': 
+ 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-' %
@@ -160,7 +160,7 @@
 c2 = _multiprocessing.Connection(fd2, readable=False)
 
 return c1, c2
- 
+
 else:
 
 from ._multiprocessing import win32
@@ -200,7 +200,7 @@
 
 c1 = _multiprocessing.PipeConnection(h1, writable=duplex)
 c2 = _multiprocessing.PipeConnection(h2, readable=duplex)
- 
+
 return c1, c2
 
 #
@@ -290,14 +290,14 @@
 )
 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,
@@ -320,7 +320,7 @@
 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`
@@ -397,7 +397,7 @@
 self._loads = loads
 for attr in ('fileno', 'close', 'poll', 'recv_bytes', 'send_bytes'):
 obj = getattr(conn, attr)
- setattr(self, attr, obj) 
+ setattr(self, attr, obj)
 def send(self, obj):
 s = self._dumps(obj)
 self._conn.send_bytes(s)
Modified: python/trunk/Lib/multiprocessing/dummy/__init__.py
==============================================================================
--- python/trunk/Lib/multiprocessing/dummy/__init__.py	(original)
+++ python/trunk/Lib/multiprocessing/dummy/__init__.py	Fri Jun 13 21:13:39 2008
@@ -1,143 +1,143 @@
-#
-# Support for the API of the multiprocessing package using threads
-#
-# multiprocessing/dummy/__init__.py
-#
-# Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
-#
-
-__all__ = [
- 'Process', 'current_process', 'active_children', 'freeze_support',
- 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Condition',
- 'Event', 'Queue', 'Manager', 'Pipe', 'Pool', 'JoinableQueue'
- ]
-
-#
-# Imports
-#
-
-import threading
-import sys
-import weakref
-import array
-import itertools
-
-from multiprocessing import TimeoutError, cpu_count
-from multiprocessing.dummy.connection import Pipe
-from threading import Lock, RLock, Semaphore, BoundedSemaphore
-from threading import Event
-from Queue import Queue
-
-#
-#
-#
-
-class DummyProcess(threading.Thread):
-
- def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
- threading.Thread.__init__(self, group, target, name, args, kwargs)
- self._pid = None
- self._children = weakref.WeakKeyDictionary()
- self._start_called = False
- self._parent = current_process()
-
- def start(self):
- assert self._parent is current_process()
- self._start_called = True
- self._parent._children[self] = None
- threading.Thread.start(self)
-
- def get_exitcode(self):
- if self._start_called and not self.is_alive():
- return 0
- else:
- return None
-
- # XXX
- if sys.version_info < (3, 0):
- is_alive = threading.Thread.is_alive.im_func
- get_name = threading.Thread.get_name.im_func
- set_name = threading.Thread.set_name.im_func
- is_daemon = threading.Thread.is_daemon.im_func
- set_daemon = threading.Thread.set_daemon.im_func
- else:
- is_alive = threading.Thread.is_alive
- get_name = threading.Thread.get_name
- set_name = threading.Thread.set_name
- is_daemon = threading.Thread.is_daemon
- set_daemon = threading.Thread.set_daemon
-
-#
-#
-#
- 
-class Condition(threading._Condition):
- # XXX
- if sys.version_info < (3, 0):
- notify_all = threading._Condition.notify_all.im_func
- else:
- notify_all = threading._Condition.notify_all
-
-#
-#
-#
-
-Process = DummyProcess
-current_process = threading.current_thread
-current_process()._children = weakref.WeakKeyDictionary()
-
-def active_children():
- children = current_process()._children
- for p in list(children):
- if not p.is_alive():
- children.pop(p, None)
- return list(children)
-
-def freeze_support():
- pass
-
-#
-#
-#
-
-class Namespace(object):
- def __init__(self, **kwds):
- self.__dict__.update(kwds)
- def __repr__(self):
- items = self.__dict__.items()
- temp = []
- for name, value in items:
- if not name.startswith('_'):
- temp.append('%s=%r' % (name, value))
- temp.sort()
- return 'Namespace(%s)' % str.join(', ', temp)
-
-dict = dict
-list = list
-
-def Array(typecode, sequence, lock=True):
- return array.array(typecode, sequence)
-
-class Value(object):
- def __init__(self, typecode, value, lock=True):
- self._typecode = typecode
- self._value = value
- def _get(self):
- return self._value
- def _set(self, value):
- self._value = value
- value = property(_get, _set)
- def __repr__(self):
- return '<%r(%r, %r)>'%(type(self).__name__,self._typecode,self._value)
-
-def Manager():
- return sys.modules[__name__]
-
-def shutdown():
- pass
-
-def Pool(processes=None, initializer=None, initargs=()):
- from multiprocessing.pool import ThreadPool
- return ThreadPool(processes, initializer, initargs)
-
-JoinableQueue = Queue
+#
+# Support for the API of the multiprocessing package using threads
+#
+# multiprocessing/dummy/__init__.py
+#
+# Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
+#
+
+__all__ = [
+ 'Process', 'current_process', 'active_children', 'freeze_support',
+ 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Condition',
+ 'Event', 'Queue', 'Manager', 'Pipe', 'Pool', 'JoinableQueue'
+ ]
+
+#
+# Imports
+#
+
+import threading
+import sys
+import weakref
+import array
+import itertools
+
+from multiprocessing import TimeoutError, cpu_count
+from multiprocessing.dummy.connection import Pipe
+from threading import Lock, RLock, Semaphore, BoundedSemaphore
+from threading import Event
+from Queue import Queue
+
+#
+#
+#
+
+class DummyProcess(threading.Thread):
+
+ def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
+ threading.Thread.__init__(self, group, target, name, args, kwargs)
+ self._pid = None
+ self._children = weakref.WeakKeyDictionary()
+ self._start_called = False
+ self._parent = current_process()
+
+ def start(self):
+ assert self._parent is current_process()
+ self._start_called = True
+ self._parent._children[self] = None
+ threading.Thread.start(self)
+
+ def get_exitcode(self):
+ if self._start_called and not self.is_alive():
+ return 0
+ else:
+ return None
+
+ # XXX
+ if sys.version_info < (3, 0):
+ is_alive = threading.Thread.is_alive.im_func
+ get_name = threading.Thread.get_name.im_func
+ set_name = threading.Thread.set_name.im_func
+ is_daemon = threading.Thread.is_daemon.im_func
+ set_daemon = threading.Thread.set_daemon.im_func
+ else:
+ is_alive = threading.Thread.is_alive
+ get_name = threading.Thread.get_name
+ set_name = threading.Thread.set_name
+ is_daemon = threading.Thread.is_daemon
+ set_daemon = threading.Thread.set_daemon
+
+#
+#
+#
+
+class Condition(threading._Condition):
+ # XXX
+ if sys.version_info < (3, 0):
+ notify_all = threading._Condition.notify_all.im_func
+ else:
+ notify_all = threading._Condition.notify_all
+
+#
+#
+#
+
+Process = DummyProcess
+current_process = threading.current_thread
+current_process()._children = weakref.WeakKeyDictionary()
+
+def active_children():
+ children = current_process()._children
+ for p in list(children):
+ if not p.is_alive():
+ children.pop(p, None)
+ return list(children)
+
+def freeze_support():
+ pass
+
+#
+#
+#
+
+class Namespace(object):
+ def __init__(self, **kwds):
+ self.__dict__.update(kwds)
+ def __repr__(self):
+ items = self.__dict__.items()
+ temp = []
+ for name, value in items:
+ if not name.startswith('_'):
+ temp.append('%s=%r' % (name, value))
+ temp.sort()
+ return 'Namespace(%s)' % str.join(', ', temp)
+
+dict = dict
+list = list
+
+def Array(typecode, sequence, lock=True):
+ return array.array(typecode, sequence)
+
+class Value(object):
+ def __init__(self, typecode, value, lock=True):
+ self._typecode = typecode
+ self._value = value
+ def _get(self):
+ return self._value
+ def _set(self, value):
+ self._value = value
+ value = property(_get, _set)
+ def __repr__(self):
+ return '<%r(%r, %r)>'%(type(self).__name__,self._typecode,self._value)
+
+def Manager():
+ return sys.modules[__name__]
+
+def shutdown():
+ pass
+
+def Pool(processes=None, initializer=None, initargs=()):
+ from multiprocessing.pool import ThreadPool
+ return ThreadPool(processes, initializer, initargs)
+
+JoinableQueue = Queue
Modified: python/trunk/Lib/multiprocessing/dummy/connection.py
==============================================================================
--- python/trunk/Lib/multiprocessing/dummy/connection.py	(original)
+++ python/trunk/Lib/multiprocessing/dummy/connection.py	Fri Jun 13 21:13:39 2008
@@ -1,61 +1,61 @@
-#
-# Analogue of `multiprocessing.connection` which uses queues instead of sockets
-#
-# multiprocessing/dummy/connection.py
-#
-# Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
-#
-
-__all__ = [ 'Client', 'Listener', 'Pipe' ]
-
-from Queue import Queue
-
-
-families = [None]
-
-
-class Listener(object):
-
- def __init__(self, address=None, family=None, backlog=1):
- self._backlog_queue = Queue(backlog)
-
- def accept(self):
- return Connection(*self._backlog_queue.get())
-
- def close(self):
- self._backlog_queue = None
-
- address = property(lambda self: self._backlog_queue)
-
-
-def Client(address):
- _in, _out = Queue(), Queue()
- address.put((_out, _in))
- return Connection(_in, _out)
-
-
-def Pipe(duplex=True):
- a, b = Queue(), Queue()
- return Connection(a, b), Connection(b, a)
-
-
-class Connection(object):
-
- def __init__(self, _in, _out):
- self._out = _out
- self._in = _in
- self.send = self.send_bytes = _out.put
- self.recv = self.recv_bytes = _in.get
-
- def poll(self, timeout=0.0):
- if self._in.qsize() > 0:
- return True
- if timeout <= 0.0:
- return False
- self._in.not_empty.acquire()
- self._in.not_empty.wait(timeout)
- self._in.not_empty.release()
- return self._in.qsize() > 0
-
- def close(self):
- pass
+#
+# Analogue of `multiprocessing.connection` which uses queues instead of sockets
+#
+# multiprocessing/dummy/connection.py
+#
+# Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
+#
+
+__all__ = [ 'Client', 'Listener', 'Pipe' ]
+
+from Queue import Queue
+
+
+families = [None]
+
+
+class Listener(object):
+
+ def __init__(self, address=None, family=None, backlog=1):
+ self._backlog_queue = Queue(backlog)
+
+ def accept(self):
+ return Connection(*self._backlog_queue.get())
+
+ def close(self):
+ self._backlog_queue = None
+
+ address = property(lambda self: self._backlog_queue)
+
+
+def Client(address):
+ _in, _out = Queue(), Queue()
+ address.put((_out, _in))
+ return Connection(_in, _out)
+
+
+def Pipe(duplex=True):
+ a, b = Queue(), Queue()
+ return Connection(a, b), Connection(b, a)
+
+
+class Connection(object):
+
+ def __init__(self, _in, _out):
+ self._out = _out
+ self._in = _in
+ self.send = self.send_bytes = _out.put
+ self.recv = self.recv_bytes = _in.get
+
+ def poll(self, timeout=0.0):
+ if self._in.qsize() > 0:
+ return True
+ if timeout <= 0.0:
+ return False
+ self._in.not_empty.acquire()
+ self._in.not_empty.wait(timeout)
+ self._in.not_empty.release()
+ return self._in.qsize() > 0
+
+ def close(self):
+ pass
Modified: python/trunk/Lib/multiprocessing/forking.py
==============================================================================
--- python/trunk/Lib/multiprocessing/forking.py	(original)
+++ python/trunk/Lib/multiprocessing/forking.py	Fri Jun 13 21:13:39 2008
@@ -92,7 +92,7 @@
 except OSError, e:
 if self.wait(timeout=0.1) is None:
 raise
- 
+
 @staticmethod
 def thread_is_spawning():
 return False
@@ -107,10 +107,10 @@
 import _subprocess
 import copy_reg
 import time
- 
+
 from ._multiprocessing import win32, Connection, PipeConnection
 from .util import Finalize
- 
+
 try:
 from cPickle import dump, load, HIGHEST_PROTOCOL
 except ImportError:
@@ -217,7 +217,7 @@
 if code == TERMINATE:
 code = -signal.SIGTERM
 self.returncode = code
- 
+
 return self.returncode
 
 def poll(self):
@@ -230,7 +230,7 @@
 except WindowsError:
 if self.wait(timeout=0.1) is None:
 raise
- 
+
 #
 #
 #
@@ -308,7 +308,7 @@
 Return info about parent needed by child to unpickle process object
 '''
 from .util import _logger, _log_to_stderr
- 
+
 d = dict(
 name=name,
 sys_path=sys.path,
@@ -317,7 +317,7 @@
 orig_dir=process.ORIGINAL_DIR,
 authkey=process.current_process().get_authkey(),
 )
- 
+
 if _logger is not None:
 d['log_level'] = _logger.getEffectiveLevel()
 
@@ -336,7 +336,7 @@
 #
 # Make (Pipe)Connection picklable
 #
- 
+
 def reduce_connection(conn):
 if not Popen.thread_is_spawning():
 raise RuntimeError(
@@ -345,7 +345,7 @@
 )
 return type(conn), (Popen.duplicate_for_child(conn.fileno()),
 conn.readable, conn.writable)
- 
+
 copy_reg.pickle(Connection, reduce_connection)
 copy_reg.pickle(PipeConnection, reduce_connection)
 
@@ -367,7 +367,7 @@
 
 if 'authkey' in data:
 process.current_process()._authkey = data['authkey']
- 
+
 if 'log_to_stderr' in data and data['log_to_stderr']:
 util.log_to_stderr()
 
Modified: python/trunk/Lib/multiprocessing/heap.py
==============================================================================
--- python/trunk/Lib/multiprocessing/heap.py	(original)
+++ python/trunk/Lib/multiprocessing/heap.py	Fri Jun 13 21:13:39 2008
@@ -1,201 +1,201 @@
-#
-# Module which supports allocation of memory from an mmap
-#
-# multiprocessing/heap.py
-#
-# Copyright (c) 2007-2008, R Oudkerk --- see COPYING.txt
-#
-
-import bisect
-import mmap
-import tempfile
-import os
-import sys
-import threading
-import itertools
-
-import _multiprocessing
-from multiprocessing.util import Finalize, info
-from multiprocessing.forking import assert_spawning
-
-__all__ = ['BufferWrapper']
-
-#
-# Inheirtable class which wraps an mmap, and from which blocks can be allocated
-#
-
-if sys.platform == 'win32':
-
- from ._multiprocessing import win32
-
- class Arena(object):
-
- _counter = itertools.count()
-
- def __init__(self, size):
- self.size = size
- self.name = 'pym-%d-%d' % (os.getpid(), Arena._counter.next())
- self.buffer = mmap.mmap(-1, self.size, tagname=self.name)
- assert win32.GetLastError() == 0, 'tagname already in use'
- self._state = (self.size, self.name)
-
- def __getstate__(self):
- assert_spawning(self)
- return self._state
-
- def __setstate__(self, state):
- self.size, self.name = self._state = state
- self.buffer = mmap.mmap(-1, self.size, tagname=self.name)
- assert win32.GetLastError() == win32.ERROR_ALREADY_EXISTS
-
-else:
-
- class Arena(object):
-
- def __init__(self, size):
- self.buffer = mmap.mmap(-1, size)
- self.size = size
- self.name = None
-
-#
-# Class allowing allocation of chunks of memory from arenas
-#
-
-class Heap(object):
-
- _alignment = 8
-
- def __init__(self, size=mmap.PAGESIZE):
- self._lastpid = os.getpid()
- self._lock = threading.Lock()
- self._size = size
- self._lengths = []
- self._len_to_seq = {}
- self._start_to_block = {}
- self._stop_to_block = {}
- self._allocated_blocks = set()
- self._arenas = []
-
- @staticmethod
- def _roundup(n, alignment):
- # alignment must be a power of 2
- mask = alignment - 1
- return (n + mask) & ~mask
-
- def _malloc(self, size):
- # returns a large enough block -- it might be much larger
- i = bisect.bisect_left(self._lengths, size)
- if i == len(self._lengths):
- length = self._roundup(max(self._size, size), mmap.PAGESIZE)
- self._size *= 2
- info('allocating a new mmap of length %d', length)
- arena = Arena(length)
- self._arenas.append(arena)
- return (arena, 0, length)
- else:
- length = self._lengths[i]
- seq = self._len_to_seq[length]
- block = seq.pop()
- if not seq:
- del self._len_to_seq[length], self._lengths[i]
-
- (arena, start, stop) = block
- del self._start_to_block[(arena, start)]
- del self._stop_to_block[(arena, stop)]
- return block
-
- def _free(self, block):
- # free location and try to merge with neighbours
- (arena, start, stop) = block
-
- try:
- prev_block = self._stop_to_block[(arena, start)]
- except KeyError:
- pass
- else:
- start, _ = self._absorb(prev_block)
-
- try:
- next_block = self._start_to_block[(arena, stop)]
- except KeyError:
- pass
- else:
- _, stop = self._absorb(next_block)
-
- block = (arena, start, stop)
- length = stop - start
-
- try:
- self._len_to_seq[length].append(block)
- except KeyError:
- self._len_to_seq[length] = [block]
- bisect.insort(self._lengths, length)
-
- self._start_to_block[(arena, start)] = block
- self._stop_to_block[(arena, stop)] = block
-
- def _absorb(self, block):
- # deregister this block so it can be merged with a neighbour
- (arena, start, stop) = block
- del self._start_to_block[(arena, start)]
- del self._stop_to_block[(arena, stop)]
-
- length = stop - start
- seq = self._len_to_seq[length]
- seq.remove(block)
- if not seq:
- del self._len_to_seq[length]
- self._lengths.remove(length)
-
- return start, stop
-
- def free(self, block):
- # free a block returned by malloc()
- assert os.getpid() == self._lastpid
- self._lock.acquire()
- try:
- self._allocated_blocks.remove(block)
- self._free(block)
- finally:
- self._lock.release()
-
- def malloc(self, size):
- # return a block of right size (possibly rounded up)
- assert 0 <= size < sys.maxint
- if os.getpid() != self._lastpid:
- self.__init__() # reinitialize after fork
- self._lock.acquire()
- try:
- size = self._roundup(max(size,1), self._alignment)
- (arena, start, stop) = self._malloc(size)
- new_stop = start + size
- if new_stop < stop:
- self._free((arena, new_stop, stop))
- block = (arena, start, new_stop)
- self._allocated_blocks.add(block)
- return block
- finally:
- self._lock.release()
-
-#
-# Class representing a chunk of an mmap -- can be inherited
-#
-
-class BufferWrapper(object):
-
- _heap = Heap()
-
- def __init__(self, size):
- assert 0 <= size < sys.maxint
- block = BufferWrapper._heap.malloc(size)
- self._state = (block, size)
- Finalize(self, BufferWrapper._heap.free, args=(block,))
-
- def get_address(self):
- (arena, start, stop), size = self._state
- address, length = _multiprocessing.address_of_buffer(arena.buffer)
- assert size <= length
- return address + start
-
- def get_size(self):
- return self._state[1]
+#
+# Module which supports allocation of memory from an mmap
+#
+# multiprocessing/heap.py
+#
+# Copyright (c) 2007-2008, R Oudkerk --- see COPYING.txt
+#
+
+import bisect
+import mmap
+import tempfile
+import os
+import sys
+import threading
+import itertools
+
+import _multiprocessing
+from multiprocessing.util import Finalize, info
+from multiprocessing.forking import assert_spawning
+
+__all__ = ['BufferWrapper']
+
+#
+# Inheirtable class which wraps an mmap, and from which blocks can be allocated
+#
+
+if sys.platform == 'win32':
+
+ from ._multiprocessing import win32
+
+ class Arena(object):
+
+ _counter = itertools.count()
+
+ def __init__(self, size):
+ self.size = size
+ self.name = 'pym-%d-%d' % (os.getpid(), Arena._counter.next())
+ self.buffer = mmap.mmap(-1, self.size, tagname=self.name)
+ assert win32.GetLastError() == 0, 'tagname already in use'
+ self._state = (self.size, self.name)
+
+ def __getstate__(self):
+ assert_spawning(self)
+ return self._state
+
+ def __setstate__(self, state):
+ self.size, self.name = self._state = state
+ self.buffer = mmap.mmap(-1, self.size, tagname=self.name)
+ assert win32.GetLastError() == win32.ERROR_ALREADY_EXISTS
+
+else:
+
+ class Arena(object):
+
+ def __init__(self, size):
+ self.buffer = mmap.mmap(-1, size)
+ self.size = size
+ self.name = None
+
+#
+# Class allowing allocation of chunks of memory from arenas
+#
+
+class Heap(object):
+
+ _alignment = 8
+
+ def __init__(self, size=mmap.PAGESIZE):
+ self._lastpid = os.getpid()
+ self._lock = threading.Lock()
+ self._size = size
+ self._lengths = []
+ self._len_to_seq = {}
+ self._start_to_block = {}
+ self._stop_to_block = {}
+ self._allocated_blocks = set()
+ self._arenas = []
+
+ @staticmethod
+ def _roundup(n, alignment):
+ # alignment must be a power of 2
+ mask = alignment - 1
+ return (n + mask) & ~mask
+
+ def _malloc(self, size):
+ # returns a large enough block -- it might be much larger
+ i = bisect.bisect_left(self._lengths, size)
+ if i == len(self._lengths):
+ length = self._roundup(max(self._size, size), mmap.PAGESIZE)
+ self._size *= 2
+ info('allocating a new mmap of length %d', length)
+ arena = Arena(length)
+ self._arenas.append(arena)
+ return (arena, 0, length)
+ else:
+ length = self._lengths[i]
+ seq = self._len_to_seq[length]
+ block = seq.pop()
+ if not seq:
+ del self._len_to_seq[length], self._lengths[i]
+
+ (arena, start, stop) = block
+ del self._start_to_block[(arena, start)]
+ del self._stop_to_block[(arena, stop)]
+ return block
+
+ def _free(self, block):
+ # free location and try to merge with neighbours
+ (arena, start, stop) = block
+
+ try:
+ prev_block = self._stop_to_block[(arena, start)]
+ except KeyError:
+ pass
+ else:
+ start, _ = self._absorb(prev_block)
+
+ try:
+ next_block = self._start_to_block[(arena, stop)]
+ except KeyError:
+ pass
+ else:
+ _, stop = self._absorb(next_block)
+
+ block = (arena, start, stop)
+ length = stop - start
+
+ try:
+ self._len_to_seq[length].append(block)
+ except KeyError:
+ self._len_to_seq[length] = [block]
+ bisect.insort(self._lengths, length)
+
+ self._start_to_block[(arena, start)] = block
+ self._stop_to_block[(arena, stop)] = block
+
+ def _absorb(self, block):
+ # deregister this block so it can be merged with a neighbour
+ (arena, start, stop) = block
+ del self._start_to_block[(arena, start)]
+ del self._stop_to_block[(arena, stop)]
+
+ length = stop - start
+ seq = self._len_to_seq[length]
+ seq.remove(block)
+ if not seq:
+ del self._len_to_seq[length]
+ self._lengths.remove(length)
+
+ return start, stop
+
+ def free(self, block):
+ # free a block returned by malloc()
+ assert os.getpid() == self._lastpid
+ self._lock.acquire()
+ try:
+ self._allocated_blocks.remove(block)
+ self._free(block)
+ finally:
+ self._lock.release()
+
+ def malloc(self, size):
+ # return a block of right size (possibly rounded up)
+ assert 0 <= size < sys.maxint
+ if os.getpid() != self._lastpid:
+ self.__init__() # reinitialize after fork
+ self._lock.acquire()
+ try:
+ size = self._roundup(max(size,1), self._alignment)
+ (arena, start, stop) = self._malloc(size)
+ new_stop = start + size
+ if new_stop < stop:
+ self._free((arena, new_stop, stop))
+ block = (arena, start, new_stop)
+ self._allocated_blocks.add(block)
+ return block
+ finally:
+ self._lock.release()
+
+#
+# Class representing a chunk of an mmap -- can be inherited
+#
+
+class BufferWrapper(object):
+
+ _heap = Heap()
+
+ def __init__(self, size):
+ assert 0 <= size < sys.maxint
+ block = BufferWrapper._heap.malloc(size)
+ self._state = (block, size)
+ Finalize(self, BufferWrapper._heap.free, args=(block,))
+
+ def get_address(self):
+ (arena, start, stop), size = self._state
+ address, length = _multiprocessing.address_of_buffer(arena.buffer)
+ assert size <= length
+ return address + start
+
+ def get_size(self):
+ return self._state[1]
Modified: python/trunk/Lib/multiprocessing/managers.py
==============================================================================
--- python/trunk/Lib/multiprocessing/managers.py	(original)
+++ python/trunk/Lib/multiprocessing/managers.py	Fri Jun 13 21:13:39 2008
@@ -40,7 +40,7 @@
 bytes
 except NameError:
 bytes = str # XXX not needed in Py2.6 and Py3.0
- 
+
 #
 # Register some things for pickling
 #
@@ -55,7 +55,7 @@
 return list, (list(obj),)
 for view_type in view_types:
 copy_reg.pickle(view_type, rebuild_as_list)
- 
+
 #
 # Type for identifying shared objects
 #
@@ -104,7 +104,7 @@
 return RemoteError('Unserializable message: %s\n' % result)
 else:
 return ValueError('Unrecognized message type')
- 
+
 class RemoteError(Exception):
 def __str__(self):
 return ('\n' + '-'*75 + '\n' + str(self.args[0]) + '-'*75)
@@ -340,7 +340,7 @@
 util.debug('resetting stdout, stderr')
 sys.stdout = sys.__stdout__
 sys.stderr = sys.__stderr__
- 
+
 util._run_finalizers(0)
 
 for p in active_children():
@@ -358,7 +358,7 @@
 traceback.print_exc()
 finally:
 exit(0)
- 
+
 def create(self, c, typeid, *args, **kwds):
 '''
 Create a new shared object and return its id
@@ -367,7 +367,7 @@
 try:
 callable, exposed, method_to_typeid, proxytype = \
 self.registry[typeid]
- 
+
 if callable is None:
 assert len(args) == 1 and not kwds
 obj = args[0]
@@ -456,7 +456,7 @@
 '''
 _registry = {}
 _Server = Server
- 
+
 def __init__(self, address=None, authkey=None, serializer='pickle'):
 if authkey is None:
 authkey = current_process().get_authkey()
@@ -487,7 +487,7 @@
 conn = Client(self._address, authkey=self._authkey)
 dispatch(conn, None, 'dummy')
 self._state.value = State.STARTED
- 
+
 def start(self):
 '''
 Spawn a server process for this manager object
@@ -570,10 +570,10 @@
 Return the number of shared objects
 '''
 conn = self._Client(self._address, authkey=self._authkey)
- try: 
+ try:
 return dispatch(conn, None, 'number_of_objects')
 finally:
- conn.close() 
+ conn.close()
 
 def __enter__(self):
 return self
@@ -612,7 +612,7 @@
 del BaseProxy._address_to_local[address]
 except KeyError:
 pass
- 
+
 address = property(lambda self: self._address)
 
 @classmethod
@@ -640,7 +640,7 @@
 cls._registry[typeid] = (
 callable, exposed, method_to_typeid, proxytype
 )
- 
+
 if create_method:
 def temp(self, *args, **kwds):
 util.debug('requesting creation of a shared %r object', typeid)
@@ -709,9 +709,9 @@
 
 if incref:
 self._incref()
- 
+
 util.register_after_fork(self, BaseProxy._after_fork)
- 
+
 def _connect(self):
 util.debug('making connection to manager')
 name = current_process().get_name()
@@ -720,7 +720,7 @@
 conn = self._Client(self._token.address, authkey=self._authkey)
 dispatch(conn, None, 'accept_connection', (name,))
 self._tls.connection = conn
- 
+
 def _callmethod(self, methodname, args=(), kwds={}):
 '''
 Try to call a method of the referrent and return a copy of the result
@@ -735,7 +735,7 @@
 
 conn.send((self._id, methodname, args, kwds))
 kind, result = conn.recv()
- 
+
 if kind == '#RETURN':
 return result
 elif kind == '#PROXY':
@@ -793,7 +793,7 @@
 threading.current_thread().get_name())
 tls.connection.close()
 del tls.connection
- 
+
 def _after_fork(self):
 self._manager = None
 try:
@@ -806,7 +806,7 @@
 kwds = {}
 if Popen.thread_is_spawning():
 kwds['authkey'] = self._authkey
- 
+
 if getattr(self, '_isauto', False):
 kwds['exposed'] = self._exposed_
 return (RebuildProxy,
@@ -817,7 +817,7 @@
 
 def __deepcopy__(self, memo):
 return self._getvalue()
- 
+
 def __repr__(self):
 return '<%s object, typeid %r at %s>' % \
 (type(self).__name__, self._token.typeid, '0x%x' % id(self))
@@ -842,7 +842,7 @@
 If possible the shared object is returned, or otherwise a proxy for it.
 '''
 server = getattr(current_process(), '_manager_server', None)
- 
+
 if server and server.address == token.address:
 return server.id_to_obj[token.id][0]
 else:
@@ -884,7 +884,7 @@
 Return an auto-proxy for `token`
 '''
 _Client = listener_client[serializer][1]
- 
+
 if exposed is None:
 conn = _Client(token.address, authkey=authkey)
 try:
@@ -995,7 +995,7 @@
 if key[0] == '_':
 return object.__getattribute__(self, key)
 callmethod = object.__getattribute__(self, '_callmethod')
- return callmethod('__getattribute__', (key,)) 
+ return callmethod('__getattribute__', (key,))
 def __setattr__(self, key, value):
 if key[0] == '_':
 return object.__setattr__(self, key, value)
@@ -1007,7 +1007,7 @@
 callmethod = object.__getattribute__(self, '_callmethod')
 return callmethod('__delattr__', (key,))
 
- 
+
 class ValueProxy(BaseProxy):
 _exposed_ = ('get', 'set')
 def get(self):
@@ -1063,10 +1063,10 @@
 class SyncManager(BaseManager):
 '''
 Subclass of `BaseManager` which supports a number of shared object types.
- 
+
 The types registered are those intended for the synchronization
 of threads, plus `dict`, `list` and `Namespace`.
- 
+
 The `multiprocessing.Manager()` function creates started instances of
 this class.
 '''
Modified: python/trunk/Lib/multiprocessing/pool.py
==============================================================================
--- python/trunk/Lib/multiprocessing/pool.py	(original)
+++ python/trunk/Lib/multiprocessing/pool.py	Fri Jun 13 21:13:39 2008
@@ -58,18 +58,18 @@
 except (EOFError, IOError):
 debug('worker got EOFError or IOError -- exiting')
 break
- 
+
 if task is None:
 debug('worker got sentinel -- exiting')
 break
- 
+
 job, i, func, args, kwds = task
 try:
 result = (True, func(*args, **kwds))
 except Exception, e:
 result = (False, e)
 put((job, i, result))
- 
+
 #
 # Class representing a process pool
 #
@@ -91,7 +91,7 @@
 processes = cpu_count()
 except NotImplementedError:
 processes = 1
- 
+
 self._pool = []
 for i in range(processes):
 w = self.Process(
@@ -102,7 +102,7 @@
 w.set_name(w.get_name().replace('Process', 'PoolWorker'))
 w.set_daemon(True)
 w.start()
- 
+
 self._task_handler = threading.Thread(
 target=Pool._handle_tasks,
 args=(self._taskqueue, self._quick_put, self._outqueue, self._pool)
@@ -132,7 +132,7 @@
 self._outqueue = SimpleQueue()
 self._quick_put = self._inqueue._writer.send
 self._quick_get = self._outqueue._reader.recv
- 
+
 def apply(self, func, args=(), kwds={}):
 '''
 Equivalent of `apply()` builtin
@@ -182,7 +182,7 @@
 self._taskqueue.put((((result._job, i, mapstar, (x,), {})
 for i, x in enumerate(task_batches)), result._set_length))
 return (item for chunk in result for item in chunk)
- 
+
 def apply_async(self, func, args=(), kwds={}, callback=None):
 '''
 Asynchronous equivalent of `apply()` builtin
@@ -199,12 +199,12 @@
 assert self._state == RUN
 if not hasattr(iterable, '__len__'):
 iterable = list(iterable)
- 
+
 if chunksize is None:
 chunksize, extra = divmod(len(iterable), len(self._pool) * 4)
 if extra:
 chunksize += 1
- 
+
 task_batches = Pool._get_tasks(func, iterable, chunksize)
 result = MapResult(self._cache, chunksize, len(iterable), callback)
 self._taskqueue.put((((result._job, i, mapstar, (x,), {})
@@ -234,13 +234,13 @@
 break
 else:
 debug('task handler got sentinel')
- 
+
 
 try:
 # tell result handler to finish when cache is empty
 debug('task handler sending sentinel to result handler')
 outqueue.put(None)
- 
+
 # tell workers there is no more work
 debug('task handler sending sentinel to workers')
 for p in pool:
@@ -260,12 +260,12 @@
 except (IOError, EOFError):
 debug('result handler got EOFError/IOError -- exiting')
 return
- 
+
 if thread._state:
 assert thread._state == TERMINATE
 debug('result handler found thread._state=TERMINATE')
 break
- 
+
 if task is None:
 debug('result handler got sentinel')
 break
@@ -321,7 +321,7 @@
 raise NotImplementedError(
 'pool objects cannot be passed between processes or pickled'
 )
- 
+
 def close(self):
 debug('closing pool')
 if self._state == RUN:
@@ -355,7 +355,7 @@
 task_handler, result_handler, cache):
 # this is guaranteed to only be called once
 debug('finalizing pool')
- 
+
 task_handler._state = TERMINATE
 taskqueue.put(None) # sentinel
 
@@ -363,7 +363,7 @@
 cls._help_stuff_finish(inqueue, task_handler, len(pool))
 
 assert result_handler.is_alive() or len(cache) == 0
- 
+
 result_handler._state = TERMINATE
 outqueue.put(None) # sentinel
 
@@ -396,14 +396,14 @@
 self._ready = False
 self._callback = callback
 cache[self._job] = self
- 
+
 def ready(self):
 return self._ready
- 
+
 def successful(self):
 assert self._ready
 return self._success
- 
+
 def wait(self, timeout=None):
 self._cond.acquire()
 try:
@@ -438,7 +438,7 @@
 #
 
 class MapResult(ApplyResult):
- 
+
 def __init__(self, cache, chunksize, length, callback):
 ApplyResult.__init__(self, cache, callback)
 self._success = True
@@ -449,7 +449,7 @@
 self._ready = True
 else:
 self._number_left = length//chunksize + bool(length % chunksize)
- 
+
 def _set(self, i, success_result):
 success, result = success_result
 if success:
@@ -492,10 +492,10 @@
 self._length = None
 self._unsorted = {}
 cache[self._job] = self
- 
+
 def __iter__(self):
 return self
- 
+
 def next(self, timeout=None):
 self._cond.acquire()
 try:
@@ -520,7 +520,7 @@
 raise value
 
 __next__ = next # XXX
- 
+
 def _set(self, i, obj):
 self._cond.acquire()
 try:
@@ -534,12 +534,12 @@
 self._cond.notify()
 else:
 self._unsorted[i] = obj
- 
+
 if self._index == self._length:
 del self._cache[self._job]
 finally:
 self._cond.release()
- 
+
 def _set_length(self, length):
 self._cond.acquire()
 try:
@@ -572,18 +572,18 @@
 #
 
 class ThreadPool(Pool):
- 
+
 from .dummy import Process
- 
+
 def __init__(self, processes=None, initializer=None, initargs=()):
 Pool.__init__(self, processes, initializer, initargs)
- 
+
 def _setup_queues(self):
 self._inqueue = Queue.Queue()
 self._outqueue = Queue.Queue()
 self._quick_put = self._inqueue.put
 self._quick_get = self._outqueue.get
- 
+
 @staticmethod
 def _help_stuff_finish(inqueue, task_handler, size):
 # put sentinels at head of inqueue to make workers finish
Modified: python/trunk/Lib/multiprocessing/process.py
==============================================================================
--- python/trunk/Lib/multiprocessing/process.py	(original)
+++ python/trunk/Lib/multiprocessing/process.py	Fri Jun 13 21:13:39 2008
@@ -47,7 +47,7 @@
 '''
 _cleanup()
 return list(_current_process._children)
- 
+
 #
 #
 #
@@ -69,7 +69,7 @@
 The class is analagous to `threading.Thread`
 '''
 _Popen = None
- 
+
 def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
 assert group is None, 'group argument must be None for now'
 count = _current_process._counter.next()
@@ -91,7 +91,7 @@
 '''
 if self._target:
 self._target(*self._args, **self._kwargs)
- 
+
 def start(self):
 '''
 Start child process
@@ -114,7 +114,7 @@
 Terminate process; sends SIGTERM signal or uses TerminateProcess()
 '''
 self._popen.terminate()
- 
+
 def join(self, timeout=None):
 '''
 Wait until child process terminates
@@ -217,11 +217,11 @@
 status, self._daemonic and ' daemon' or '')
 
 ##
- 
+
 def _bootstrap(self):
 from . import util
 global _current_process
- 
+
 try:
 self._children = set()
 self._counter = itertools.count(1)
Modified: python/trunk/Lib/multiprocessing/queues.py
==============================================================================
--- python/trunk/Lib/multiprocessing/queues.py	(original)
+++ python/trunk/Lib/multiprocessing/queues.py	Fri Jun 13 21:13:39 2008
@@ -41,9 +41,9 @@
 else:
 self._wlock = Lock()
 self._sem = BoundedSemaphore(maxsize)
- 
+
 self._after_fork()
- 
+
 if sys.platform != 'win32':
 register_after_fork(self, Queue._after_fork)
 
@@ -51,12 +51,12 @@
 assert_spawning(self)
 return (self._maxsize, self._reader, self._writer,
 self._rlock, self._wlock, self._sem, self._opid)
- 
+
 def __setstate__(self, state):
 (self._maxsize, self._reader, self._writer,
 self._rlock, self._wlock, self._sem, self._opid) = state
 self._after_fork()
- 
+
 def _after_fork(self):
 debug('Queue._after_fork()')
 self._notempty = threading.Condition(threading.Lock())
@@ -69,7 +69,7 @@
 self._send = self._writer.send
 self._recv = self._reader.recv
 self._poll = self._reader.poll
- 
+
 def put(self, obj, block=True, timeout=None):
 assert not self._closed
 if not self._sem.acquire(block, timeout):
@@ -93,7 +93,7 @@
 return res
 finally:
 self._rlock.release()
- 
+
 else:
 if block:
 deadline = time.time() + timeout
@@ -135,7 +135,7 @@
 assert self._closed
 if self._jointhread:
 self._jointhread()
- 
+
 def cancel_join_thread(self):
 debug('Queue.cancel_join_thread()')
 self._joincancelled = True
@@ -146,7 +146,7 @@
 
 def _start_thread(self):
 debug('Queue._start_thread()')
- 
+
 # Start thread which transfers data from buffer to pipe
 self._buffer.clear()
 self._thread = threading.Thread(
@@ -174,14 +174,14 @@
 [weakref.ref(self._thread)],
 exitpriority=-5
 )
- 
+
 # Send sentinel to the thread queue object when garbage collected
 self._close = Finalize(
 self, Queue._finalize_close,
 [self._buffer, self._notempty],
 exitpriority=10
 )
- 
+
 @staticmethod
 def _finalize_join(twr):
 debug('joining queue thread')
@@ -191,7 +191,7 @@
 debug('... queue thread joined')
 else:
 debug('... queue thread already dead')
- 
+
 @staticmethod
 def _finalize_close(buffer, notempty):
 debug('telling queue thread to quit')
@@ -206,7 +206,7 @@
 def _feed(buffer, notempty, send, writelock, close):
 debug('starting thread to feed data to pipe')
 from .util import is_exiting
- 
+
 nacquire = notempty.acquire
 nrelease = notempty.release
 nwait = notempty.wait
@@ -217,7 +217,7 @@
 wrelease = writelock.release
 else:
 wacquire = None
- 
+
 try:
 while 1:
 nacquire()
@@ -257,7 +257,7 @@
 traceback.print_exc()
 except Exception:
 pass
- 
+
 _sentinel = object()
 
 #
@@ -274,7 +274,7 @@
 Queue.__init__(self, maxsize)
 self._unfinished_tasks = Semaphore(0)
 self._cond = Condition()
- 
+
 def __getstate__(self):
 return Queue.__getstate__(self) + (self._cond, self._unfinished_tasks)
 
@@ -285,7 +285,7 @@
 def put(self, item, block=True, timeout=None):
 Queue.put(self, item, block, timeout)
 self._unfinished_tasks.release()
- 
+
 def task_done(self):
 self._cond.acquire()
 try:
@@ -295,7 +295,7 @@
 self._cond.notify_all()
 finally:
 self._cond.release()
- 
+
 def join(self):
 self._cond.acquire()
 try:
Modified: python/trunk/Lib/multiprocessing/reduction.py
==============================================================================
--- python/trunk/Lib/multiprocessing/reduction.py	(original)
+++ python/trunk/Lib/multiprocessing/reduction.py	Fri Jun 13 21:13:39 2008
@@ -36,7 +36,7 @@
 if sys.platform == 'win32':
 import _subprocess
 from ._multiprocessing import win32
- 
+
 def send_handle(conn, handle, destination_pid):
 process_handle = win32.OpenProcess(
 win32.PROCESS_ALL_ACCESS, False, destination_pid
@@ -46,14 +46,14 @@
 conn.send(new_handle)
 finally:
 close(process_handle)
- 
+
 def recv_handle(conn):
 return conn.recv()
 
 else:
 def send_handle(conn, handle, destination_pid):
 _multiprocessing.sendfd(conn.fileno(), handle)
- 
+
 def recv_handle(conn):
 return _multiprocessing.recvfd(conn.fileno())
 
@@ -93,7 +93,7 @@
 
 def _serve():
 from .util import is_exiting, sub_warning
- 
+
 while 1:
 try:
 conn = _listener.accept()
@@ -109,7 +109,7 @@
 'thread for sharing handles raised exception :\n' +
 '-'*79 + '\n' + traceback.format_exc() + '-'*79
 )
- 
+
 #
 # Functions to be used for pickling/unpickling objects with handles
 #
@@ -176,15 +176,15 @@
 #
 
 if sys.platform == 'win32':
- 
+
 def reduce_pipe_connection(conn):
 rh = reduce_handle(conn.fileno())
 return rebuild_pipe_connection, (rh, conn.readable, conn.writable)
- 
+
 def rebuild_pipe_connection(reduced_handle, readable, writable):
 handle = rebuild_handle(reduced_handle)
 return _multiprocessing.PipeConnection(
 handle, readable=readable, writable=writable
 )
- 
+
 copy_reg.pickle(_multiprocessing.PipeConnection, reduce_pipe_connection)
Modified: python/trunk/Lib/multiprocessing/sharedctypes.py
==============================================================================
--- python/trunk/Lib/multiprocessing/sharedctypes.py	(original)
+++ python/trunk/Lib/multiprocessing/sharedctypes.py	Fri Jun 13 21:13:39 2008
@@ -92,10 +92,10 @@
 new_obj = _new_value(type(obj))
 ctypes.pointer(new_obj)[0] = obj
 return new_obj
- 
+
 def synchronized(obj, lock=None):
 assert not isinstance(obj, SynchronizedBase), 'object already synchronized'
- 
+
 if isinstance(obj, ctypes._SimpleCData):
 return Synchronized(obj, lock)
 elif isinstance(obj, ctypes.Array):
@@ -123,7 +123,7 @@
 return rebuild_ctype, (obj._type_, obj._wrapper, obj._length_)
 else:
 return rebuild_ctype, (type(obj), obj._wrapper, None)
- 
+
 def rebuild_ctype(type_, wrapper, length):
 if length is not None:
 type_ = type_ * length
@@ -170,7 +170,7 @@
 #
 
 class SynchronizedBase(object):
- 
+
 def __init__(self, obj, lock=None):
 self._obj = obj
 self._lock = lock or RLock()
@@ -180,55 +180,55 @@
 def __reduce__(self):
 assert_spawning(self)
 return synchronized, (self._obj, self._lock)
- 
+
 def get_obj(self):
 return self._obj
- 
+
 def get_lock(self):
 return self._lock
- 
+
 def __repr__(self):
 return '<%s wrapper for %s>' % (type(self).__name__, self._obj)
- 
- 
+
+
 class Synchronized(SynchronizedBase):
 value = make_property('value')
- 
- 
+
+
 class SynchronizedArray(SynchronizedBase):
- 
+
 def __len__(self):
 return len(self._obj)
- 
+
 def __getitem__(self, i):
 self.acquire()
 try:
 return self._obj[i]
 finally:
 self.release()
- 
+
 def __setitem__(self, i, value):
 self.acquire()
 try:
 self._obj[i] = value
 finally:
 self.release()
- 
+
 def __getslice__(self, start, stop):
 self.acquire()
 try:
 return self._obj[start:stop]
 finally:
 self.release()
- 
+
 def __setslice__(self, start, stop, values):
 self.acquire()
 try:
 self._obj[start:stop] = values
 finally:
 self.release()
- 
- 
+
+
 class SynchronizedString(SynchronizedArray):
 value = make_property('value')
 raw = make_property('raw')
Modified: python/trunk/Lib/multiprocessing/synchronize.py
==============================================================================
--- python/trunk/Lib/multiprocessing/synchronize.py	(original)
+++ python/trunk/Lib/multiprocessing/synchronize.py	Fri Jun 13 21:13:39 2008
@@ -38,7 +38,7 @@
 sl = self._semlock = _multiprocessing.SemLock(kind, value, maxvalue)
 debug('created semlock with handle %s' % sl.handle)
 self._make_methods()
- 
+
 if sys.platform != 'win32':
 def _after_fork(obj):
 obj._semlock._after_fork()
@@ -129,7 +129,7 @@
 
 def __init__(self):
 SemLock.__init__(self, RECURSIVE_MUTEX, 1, 1)
- 
+
 def __repr__(self):
 try:
 if self._semlock._is_mine():
@@ -210,17 +210,17 @@
 def notify(self):
 assert self._lock._semlock._is_mine(), 'lock is not owned'
 assert not self._wait_semaphore.acquire(False)
- 
+
 # to take account of timeouts since last notify() we subtract
 # woken_count from sleeping_count and rezero woken_count
 while self._woken_count.acquire(False):
 res = self._sleeping_count.acquire(False)
 assert res
- 
+
 if self._sleeping_count.acquire(False): # try grabbing a sleeper
 self._wait_semaphore.release() # wake up one sleeper
 self._woken_count.acquire() # wait for the sleeper to wake
- 
+
 # rezero _wait_semaphore in case a timeout just happened
 self._wait_semaphore.acquire(False)
 
@@ -233,7 +233,7 @@
 while self._woken_count.acquire(False):
 res = self._sleeping_count.acquire(False)
 assert res
- 
+
 sleepers = 0
 while self._sleeping_count.acquire(False):
 self._wait_semaphore.release() # wake up one sleeper
@@ -266,7 +266,7 @@
 return False
 finally:
 self._cond.release()
- 
+
 def set(self):
 self._cond.acquire()
 try:
Modified: python/trunk/Lib/multiprocessing/util.py
==============================================================================
--- python/trunk/Lib/multiprocessing/util.py	(original)
+++ python/trunk/Lib/multiprocessing/util.py	Fri Jun 13 21:13:39 2008
@@ -83,7 +83,7 @@
 import logging
 if hasattr(logging, 'multiprocessing'):
 return
- 
+
 logging._acquireLock()
 try:
 OldLoggerClass = logging.getLoggerClass()
Modified: python/trunk/Lib/test/test_multiprocessing.py
==============================================================================
--- python/trunk/Lib/test/test_multiprocessing.py	(original)
+++ python/trunk/Lib/test/test_multiprocessing.py	Fri Jun 13 21:13:39 2008
@@ -1,1791 +1,1791 @@
-#
-# Unit tests for the multiprocessing package
-#
-
-import unittest
-import threading
-import Queue
-import time
-import sys
-import os
-import gc
-import signal
-import array
-import copy
-import socket
-import random
-import logging
-
-import multiprocessing.dummy
-import multiprocessing.connection
-import multiprocessing.managers
-import multiprocessing.heap
-import multiprocessing.managers
-import multiprocessing.pool
-import _multiprocessing
-
-from multiprocessing import util
-
-#
-#
-#
-
-if sys.version_info >= (3, 0):
- def latin(s):
- return s.encode('latin')
-else:
- latin = str
-
-try:
- bytes
-except NameError:
- bytes = str
- def bytearray(seq):
- return array.array('c', seq)
-
-#
-# Constants
-#
-
-LOG_LEVEL = util.SUBWARNING
-#LOG_LEVEL = logging.WARNING
-
-DELTA = 0.1
-CHECK_TIMINGS = False # making true makes tests take a lot longer
- # and can sometimes cause some non-serious
- # failures because some calls block a bit
- # longer than expected
-if CHECK_TIMINGS:
- TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.82, 0.35, 1.4
-else:
- TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.1, 0.1, 0.1
-
-HAVE_GETVALUE = not getattr(_multiprocessing,
- 'HAVE_BROKEN_SEM_GETVALUE', False)
-
-#
-# Creates a wrapper for a function which records the time it takes to finish
-#
-
-class TimingWrapper(object):
-
- def __init__(self, func):
- self.func = func
- self.elapsed = None
-
- def __call__(self, *args, **kwds):
- t = time.time()
- try:
- return self.func(*args, **kwds)
- finally:
- self.elapsed = time.time() - t
- 
-#
-# Base class for test cases
-#
-
-class BaseTestCase(object):
- 
- ALLOWED_TYPES = ('processes', 'manager', 'threads')
-
- def assertTimingAlmostEqual(self, a, b):
- if CHECK_TIMINGS:
- self.assertAlmostEqual(a, b, 1)
-
- def assertReturnsIfImplemented(self, value, func, *args):
- try:
- res = func(*args)
- except NotImplementedError:
- pass
- else:
- return self.assertEqual(value, res)
-
-#
-# Return the value of a semaphore
-#
-
-def get_value(self):
- try:
- return self.get_value()
- except AttributeError:
- try:
- return self._Semaphore__value
- except AttributeError:
- try:
- return self._value
- except AttributeError:
- raise NotImplementedError
-
-#
-# Testcases
-#
-
-class _TestProcess(BaseTestCase):
- 
- ALLOWED_TYPES = ('processes', 'threads')
- 
- def test_current(self):
- if self.TYPE == 'threads':
- return
-
- current = self.current_process()
- authkey = current.get_authkey()
- 
- self.assertTrue(current.is_alive())
- self.assertTrue(not current.is_daemon()) 
- self.assertTrue(isinstance(authkey, bytes))
- self.assertTrue(len(authkey) > 0)
- self.assertEqual(current.get_ident(), os.getpid())
- self.assertEqual(current.get_exitcode(), None)
-
- def _test(self, q, *args, **kwds):
- current = self.current_process()
- q.put(args)
- q.put(kwds)
- q.put(current.get_name())
- if self.TYPE != 'threads':
- q.put(bytes(current.get_authkey()))
- q.put(current.pid)
-
- def test_process(self):
- q = self.Queue(1)
- e = self.Event()
- args = (q, 1, 2)
- kwargs = {'hello':23, 'bye':2.54}
- name = 'SomeProcess'
- p = self.Process(
- target=self._test, args=args, kwargs=kwargs, name=name
- )
- p.set_daemon(True)
- current = self.current_process()
-
- if self.TYPE != 'threads':
- self.assertEquals(p.get_authkey(), current.get_authkey())
- self.assertEquals(p.is_alive(), False)
- self.assertEquals(p.is_daemon(), True)
- self.assertTrue(p not in self.active_children())
- self.assertTrue(type(self.active_children()) is list)
- self.assertEqual(p.get_exitcode(), None)
- 
- p.start()
- 
- self.assertEquals(p.get_exitcode(), None)
- self.assertEquals(p.is_alive(), True)
- self.assertTrue(p in self.active_children())
- 
- self.assertEquals(q.get(), args[1:])
- self.assertEquals(q.get(), kwargs)
- self.assertEquals(q.get(), p.get_name())
- if self.TYPE != 'threads':
- self.assertEquals(q.get(), current.get_authkey())
- self.assertEquals(q.get(), p.pid)
-
- p.join()
-
- self.assertEquals(p.get_exitcode(), 0)
- self.assertEquals(p.is_alive(), False)
- self.assertTrue(p not in self.active_children()) 
-
- def _test_terminate(self):
- time.sleep(1000)
-
- def test_terminate(self):
- if self.TYPE == 'threads':
- return
- 
- p = self.Process(target=self._test_terminate)
- p.set_daemon(True)
- p.start()
-
- self.assertEqual(p.is_alive(), True)
- self.assertTrue(p in self.active_children())
- self.assertEqual(p.get_exitcode(), None)
-
- p.terminate()
-
- join = TimingWrapper(p.join)
- self.assertEqual(join(), None)
- self.assertTimingAlmostEqual(join.elapsed, 0.0)
- 
- self.assertEqual(p.is_alive(), False)
- self.assertTrue(p not in self.active_children())
-
- p.join()
-
- # XXX sometimes get p.get_exitcode() == 0 on Windows ...
- #self.assertEqual(p.get_exitcode(), -signal.SIGTERM)
-
- def test_cpu_count(self):
- try:
- cpus = multiprocessing.cpu_count()
- except NotImplementedError:
- cpus = 1
- self.assertTrue(type(cpus) is int)
- self.assertTrue(cpus >= 1)
-
- def test_active_children(self):
- self.assertEqual(type(self.active_children()), list)
-
- p = self.Process(target=time.sleep, args=(DELTA,))
- self.assertTrue(p not in self.active_children())
- 
- p.start()
- self.assertTrue(p in self.active_children())
-
- p.join()
- self.assertTrue(p not in self.active_children())
-
- def _test_recursion(self, wconn, id):
- from multiprocessing import forking
- wconn.send(id)
- if len(id) < 2:
- for i in range(2):
- p = self.Process(
- target=self._test_recursion, args=(wconn, id+[i])
- )
- p.start()
- p.join()
-
- def test_recursion(self):
- rconn, wconn = self.Pipe(duplex=False)
- self._test_recursion(wconn, [])
- 
- time.sleep(DELTA)
- result = []
- while rconn.poll():
- result.append(rconn.recv())
- 
- expected = [
- [],
- [0],
- [0, 0],
- [0, 1],
- [1],
- [1, 0],
- [1, 1]
- ]
- self.assertEqual(result, expected)
-
-#
-#
-#
-
-class _UpperCaser(multiprocessing.Process):
-
- def __init__(self):
- multiprocessing.Process.__init__(self)
- self.child_conn, self.parent_conn = multiprocessing.Pipe()
-
- def run(self):
- self.parent_conn.close()
- for s in iter(self.child_conn.recv, None):
- self.child_conn.send(s.upper())
- self.child_conn.close()
-
- def submit(self, s):
- assert type(s) is str
- self.parent_conn.send(s)
- return self.parent_conn.recv()
-
- def stop(self):
- self.parent_conn.send(None)
- self.parent_conn.close()
- self.child_conn.close()
-
-class _TestSubclassingProcess(BaseTestCase):
-
- ALLOWED_TYPES = ('processes',)
-
- def test_subclassing(self):
- uppercaser = _UpperCaser()
- uppercaser.start()
- self.assertEqual(uppercaser.submit('hello'), 'HELLO')
- self.assertEqual(uppercaser.submit('world'), 'WORLD')
- uppercaser.stop()
- uppercaser.join()
- 
-#
-#
-#
-
-def queue_empty(q):
- if hasattr(q, 'empty'):
- return q.empty()
- else:
- return q.qsize() == 0
-
-def queue_full(q, maxsize):
- if hasattr(q, 'full'):
- return q.full()
- else:
- return q.qsize() == maxsize
-
-
-class _TestQueue(BaseTestCase):
-
-
- def _test_put(self, queue, child_can_start, parent_can_continue):
- child_can_start.wait()
- for i in range(6):
- queue.get()
- parent_can_continue.set()
-
- def test_put(self):
- MAXSIZE = 6
- queue = self.Queue(maxsize=MAXSIZE)
- child_can_start = self.Event()
- parent_can_continue = self.Event()
-
- proc = self.Process(
- target=self._test_put,
- args=(queue, child_can_start, parent_can_continue)
- )
- proc.set_daemon(True)
- proc.start()
- 
- self.assertEqual(queue_empty(queue), True)
- self.assertEqual(queue_full(queue, MAXSIZE), False)
-
- queue.put(1)
- queue.put(2, True)
- queue.put(3, True, None)
- queue.put(4, False)
- queue.put(5, False, None)
- queue.put_nowait(6)
-
- # the values may be in buffer but not yet in pipe so sleep a bit
- time.sleep(DELTA) 
-
- self.assertEqual(queue_empty(queue), False)
- self.assertEqual(queue_full(queue, MAXSIZE), True)
-
- put = TimingWrapper(queue.put)
- put_nowait = TimingWrapper(queue.put_nowait)
-
- self.assertRaises(Queue.Full, put, 7, False)
- self.assertTimingAlmostEqual(put.elapsed, 0)
-
- self.assertRaises(Queue.Full, put, 7, False, None)
- self.assertTimingAlmostEqual(put.elapsed, 0)
-
- self.assertRaises(Queue.Full, put_nowait, 7)
- self.assertTimingAlmostEqual(put_nowait.elapsed, 0)
-
- self.assertRaises(Queue.Full, put, 7, True, TIMEOUT1)
- self.assertTimingAlmostEqual(put.elapsed, TIMEOUT1)
-
- self.assertRaises(Queue.Full, put, 7, False, TIMEOUT2)
- self.assertTimingAlmostEqual(put.elapsed, 0)
-
- self.assertRaises(Queue.Full, put, 7, True, timeout=TIMEOUT3)
- self.assertTimingAlmostEqual(put.elapsed, TIMEOUT3)
-
- child_can_start.set()
- parent_can_continue.wait()
-
- self.assertEqual(queue_empty(queue), True)
- self.assertEqual(queue_full(queue, MAXSIZE), False)
-
- proc.join()
-
- def _test_get(self, queue, child_can_start, parent_can_continue):
- child_can_start.wait()
- queue.put(1)
- queue.put(2)
- queue.put(3)
- queue.put(4)
- queue.put(5)
- parent_can_continue.set()
- 
- def test_get(self):
- queue = self.Queue()
- child_can_start = self.Event()
- parent_can_continue = self.Event()
- 
- proc = self.Process(
- target=self._test_get,
- args=(queue, child_can_start, parent_can_continue)
- )
- proc.set_daemon(True)
- proc.start()
- 
- self.assertEqual(queue_empty(queue), True)
- 
- child_can_start.set()
- parent_can_continue.wait()
-
- time.sleep(DELTA)
- self.assertEqual(queue_empty(queue), False)
-
- self.assertEqual(queue.get(), 1)
- self.assertEqual(queue.get(True, None), 2)
- self.assertEqual(queue.get(True), 3)
- self.assertEqual(queue.get(timeout=1), 4)
- self.assertEqual(queue.get_nowait(), 5)
- 
- self.assertEqual(queue_empty(queue), True)
-
- get = TimingWrapper(queue.get)
- get_nowait = TimingWrapper(queue.get_nowait)
- 
- self.assertRaises(Queue.Empty, get, False)
- self.assertTimingAlmostEqual(get.elapsed, 0)
-
- self.assertRaises(Queue.Empty, get, False, None)
- self.assertTimingAlmostEqual(get.elapsed, 0)
-
- self.assertRaises(Queue.Empty, get_nowait)
- self.assertTimingAlmostEqual(get_nowait.elapsed, 0)
-
- self.assertRaises(Queue.Empty, get, True, TIMEOUT1)
- self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1)
-
- self.assertRaises(Queue.Empty, get, False, TIMEOUT2)
- self.assertTimingAlmostEqual(get.elapsed, 0)
-
- self.assertRaises(Queue.Empty, get, timeout=TIMEOUT3)
- self.assertTimingAlmostEqual(get.elapsed, TIMEOUT3)
-
- proc.join()
- 
- def _test_fork(self, queue):
- for i in range(10, 20):
- queue.put(i)
- # note that at this point the items may only be buffered, so the
- # process cannot shutdown until the feeder thread has finished
- # pushing items onto the pipe.
-
- def test_fork(self):
- # Old versions of Queue would fail to create a new feeder
- # thread for a forked process if the original process had its
- # own feeder thread. This test checks that this no longer
- # happens.
-
- queue = self.Queue()
-
- # put items on queue so that main process starts a feeder thread
- for i in range(10):
- queue.put(i)
-
- # wait to make sure thread starts before we fork a new process
- time.sleep(DELTA)
-
- # fork process
- p = self.Process(target=self._test_fork, args=(queue,))
- p.start()
-
- # check that all expected items are in the queue
- for i in range(20):
- self.assertEqual(queue.get(), i)
- self.assertRaises(Queue.Empty, queue.get, False)
-
- p.join()
-
- def test_qsize(self):
- q = self.Queue()
- try:
- self.assertEqual(q.qsize(), 0)
- except NotImplementedError:
- return
- q.put(1)
- self.assertEqual(q.qsize(), 1)
- q.put(5)
- self.assertEqual(q.qsize(), 2)
- q.get()
- self.assertEqual(q.qsize(), 1)
- q.get()
- self.assertEqual(q.qsize(), 0)
-
- def _test_task_done(self, q):
- for obj in iter(q.get, None):
- time.sleep(DELTA)
- q.task_done()
-
- def test_task_done(self):
- queue = self.JoinableQueue()
-
- if sys.version_info < (2, 5) and not hasattr(queue, 'task_done'):
- return
-
- workers = [self.Process(target=self._test_task_done, args=(queue,))
- for i in xrange(4)]
- 
- for p in workers:
- p.start()
-
- for i in xrange(10):
- queue.put(i)
-
- queue.join()
-
- for p in workers:
- queue.put(None)
- 
- for p in workers:
- p.join()
-
-#
-#
-#
-
-class _TestLock(BaseTestCase):
-
- def test_lock(self):
- lock = self.Lock()
- self.assertEqual(lock.acquire(), True)
- self.assertEqual(lock.acquire(False), False)
- self.assertEqual(lock.release(), None)
- self.assertRaises((ValueError, threading.ThreadError), lock.release)
-
- def test_rlock(self):
- lock = self.RLock()
- self.assertEqual(lock.acquire(), True)
- self.assertEqual(lock.acquire(), True)
- self.assertEqual(lock.acquire(), True)
- self.assertEqual(lock.release(), None)
- self.assertEqual(lock.release(), None)
- self.assertEqual(lock.release(), None)
- self.assertRaises((AssertionError, RuntimeError), lock.release)
- 
- 
-class _TestSemaphore(BaseTestCase):
-
- def _test_semaphore(self, sem):
- self.assertReturnsIfImplemented(2, get_value, sem)
- self.assertEqual(sem.acquire(), True)
- self.assertReturnsIfImplemented(1, get_value, sem)
- self.assertEqual(sem.acquire(), True)
- self.assertReturnsIfImplemented(0, get_value, sem)
- self.assertEqual(sem.acquire(False), False)
- self.assertReturnsIfImplemented(0, get_value, sem)
- self.assertEqual(sem.release(), None)
- self.assertReturnsIfImplemented(1, get_value, sem)
- self.assertEqual(sem.release(), None)
- self.assertReturnsIfImplemented(2, get_value, sem)
- 
- def test_semaphore(self):
- sem = self.Semaphore(2)
- self._test_semaphore(sem)
- self.assertEqual(sem.release(), None)
- self.assertReturnsIfImplemented(3, get_value, sem)
- self.assertEqual(sem.release(), None)
- self.assertReturnsIfImplemented(4, get_value, sem)
-
- def test_bounded_semaphore(self):
- sem = self.BoundedSemaphore(2)
- self._test_semaphore(sem)
- # Currently fails on OS/X
- #if HAVE_GETVALUE:
- # self.assertRaises(ValueError, sem.release)
- # self.assertReturnsIfImplemented(2, get_value, sem)
-
- def test_timeout(self):
- if self.TYPE != 'processes':
- return
-
- sem = self.Semaphore(0)
- acquire = TimingWrapper(sem.acquire)
-
- self.assertEqual(acquire(False), False)
- self.assertTimingAlmostEqual(acquire.elapsed, 0.0)
-
- self.assertEqual(acquire(False, None), False)
- self.assertTimingAlmostEqual(acquire.elapsed, 0.0)
-
- self.assertEqual(acquire(False, TIMEOUT1), False)
- self.assertTimingAlmostEqual(acquire.elapsed, 0)
-
- self.assertEqual(acquire(True, TIMEOUT2), False)
- self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT2)
-
- self.assertEqual(acquire(timeout=TIMEOUT3), False)
- self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT3)
-
-
-class _TestCondition(BaseTestCase):
- 
- def f(self, cond, sleeping, woken, timeout=None):
- cond.acquire()
- sleeping.release()
- cond.wait(timeout)
- woken.release()
- cond.release()
- 
- def check_invariant(self, cond):
- # this is only supposed to succeed when there are no sleepers
- if self.TYPE == 'processes':
- try:
- sleepers = (cond._sleeping_count.get_value() -
- cond._woken_count.get_value())
- self.assertEqual(sleepers, 0)
- self.assertEqual(cond._wait_semaphore.get_value(), 0)
- except NotImplementedError:
- pass
- 
- def test_notify(self):
- cond = self.Condition()
- sleeping = self.Semaphore(0)
- woken = self.Semaphore(0)
- 
- p = self.Process(target=self.f, args=(cond, sleeping, woken))
- p.set_daemon(True)
- p.start()
-
- p = threading.Thread(target=self.f, args=(cond, sleeping, woken))
- p.set_daemon(True)
- p.start()
- 
- # wait for both children to start sleeping
- sleeping.acquire()
- sleeping.acquire()
- 
- # check no process/thread has woken up
- time.sleep(DELTA)
- self.assertReturnsIfImplemented(0, get_value, woken)
-
- # wake up one process/thread
- cond.acquire()
- cond.notify()
- cond.release()
- 
- # check one process/thread has woken up
- time.sleep(DELTA)
- self.assertReturnsIfImplemented(1, get_value, woken)
-
- # wake up another
- cond.acquire()
- cond.notify()
- cond.release()
- 
- # check other has woken up
- time.sleep(DELTA)
- self.assertReturnsIfImplemented(2, get_value, woken)
- 
- # check state is not mucked up
- self.check_invariant(cond)
- p.join()
- 
- def test_notify_all(self):
- cond = self.Condition()
- sleeping = self.Semaphore(0)
- woken = self.Semaphore(0)
-
- # start some threads/processes which will timeout
- for i in range(3):
- p = self.Process(target=self.f,
- args=(cond, sleeping, woken, TIMEOUT1))
- p.set_daemon(True)
- p.start()
-
- t = threading.Thread(target=self.f,
- args=(cond, sleeping, woken, TIMEOUT1))
- t.set_daemon(True)
- t.start()
-
- # wait for them all to sleep
- for i in xrange(6):
- sleeping.acquire()
-
- # check they have all timed out
- for i in xrange(6):
- woken.acquire()
- self.assertReturnsIfImplemented(0, get_value, woken)
-
- # check state is not mucked up
- self.check_invariant(cond)
-
- # start some more threads/processes
- for i in range(3):
- p = self.Process(target=self.f, args=(cond, sleeping, woken))
- p.set_daemon(True)
- p.start()
- 
- t = threading.Thread(target=self.f, args=(cond, sleeping, woken))
- t.set_daemon(True)
- t.start()
- 
- # wait for them to all sleep
- for i in xrange(6):
- sleeping.acquire()
- 
- # check no process/thread has woken up
- time.sleep(DELTA)
- self.assertReturnsIfImplemented(0, get_value, woken)
-
- # wake them all up
- cond.acquire()
- cond.notify_all()
- cond.release()
-
- # check they have all woken
- time.sleep(DELTA)
- self.assertReturnsIfImplemented(6, get_value, woken)
-
- # check state is not mucked up
- self.check_invariant(cond)
-
- def test_timeout(self):
- cond = self.Condition()
- wait = TimingWrapper(cond.wait)
- cond.acquire()
- res = wait(TIMEOUT1)
- cond.release()
- self.assertEqual(res, None)
- self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1)
-
- 
-class _TestEvent(BaseTestCase):
-
- def _test_event(self, event):
- time.sleep(TIMEOUT2)
- event.set()
-
- def test_event(self):
- event = self.Event()
- wait = TimingWrapper(event.wait)
- 
- # Removed temporaily, due to API shear, this does not 
- # work with threading._Event objects. is_set == isSet
- #self.assertEqual(event.is_set(), False)
- 
- self.assertEqual(wait(0.0), None)
- self.assertTimingAlmostEqual(wait.elapsed, 0.0)
- self.assertEqual(wait(TIMEOUT1), None)
- self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1)
-
- event.set()
-
- # See note above on the API differences
- # self.assertEqual(event.is_set(), True)
- self.assertEqual(wait(), None)
- self.assertTimingAlmostEqual(wait.elapsed, 0.0)
- self.assertEqual(wait(TIMEOUT1), None)
- self.assertTimingAlmostEqual(wait.elapsed, 0.0)
- # self.assertEqual(event.is_set(), True)
-
- event.clear()
-
- #self.assertEqual(event.is_set(), False)
-
- self.Process(target=self._test_event, args=(event,)).start()
- self.assertEqual(wait(), None)
-
-#
-#
-#
-
-class _TestValue(BaseTestCase):
-
- codes_values = [
- ('i', 4343, 24234),
- ('d', 3.625, -4.25),
- ('h', -232, 234),
- ('c', latin('x'), latin('y'))
- ]
-
- def _test(self, values):
- for sv, cv in zip(values, self.codes_values):
- sv.value = cv[2]
- 
- 
- def test_value(self, raw=False):
- if self.TYPE != 'processes':
- return
-
- if raw:
- values = [self.RawValue(code, value)
- for code, value, _ in self.codes_values]
- else:
- values = [self.Value(code, value)
- for code, value, _ in self.codes_values]
- 
- for sv, cv in zip(values, self.codes_values):
- self.assertEqual(sv.value, cv[1])
- 
- proc = self.Process(target=self._test, args=(values,))
- proc.start()
- proc.join()
-
- for sv, cv in zip(values, self.codes_values):
- self.assertEqual(sv.value, cv[2])
-
- def test_rawvalue(self):
- self.test_value(raw=True)
-
- def test_getobj_getlock(self):
- if self.TYPE != 'processes':
- return
-
- val1 = self.Value('i', 5)
- lock1 = val1.get_lock()
- obj1 = val1.get_obj()
-
- val2 = self.Value('i', 5, lock=None)
- lock2 = val2.get_lock()
- obj2 = val2.get_obj()
-
- lock = self.Lock()
- val3 = self.Value('i', 5, lock=lock)
- lock3 = val3.get_lock()
- obj3 = val3.get_obj()
- self.assertEqual(lock, lock3)
- 
- arr4 = self.RawValue('i', 5)
- self.assertFalse(hasattr(arr4, 'get_lock'))
- self.assertFalse(hasattr(arr4, 'get_obj'))
-
-
-class _TestArray(BaseTestCase):
-
- def f(self, seq):
- for i in range(1, len(seq)):
- seq[i] += seq[i-1]
-
- def test_array(self, raw=False):
- if self.TYPE != 'processes':
- return
-
- seq = [680, 626, 934, 821, 150, 233, 548, 982, 714, 831]
- if raw:
- arr = self.RawArray('i', seq)
- else:
- arr = self.Array('i', seq)
- 
- self.assertEqual(len(arr), len(seq))
- self.assertEqual(arr[3], seq[3])
- self.assertEqual(list(arr[2:7]), list(seq[2:7]))
- 
- arr[4:8] = seq[4:8] = array.array('i', [1, 2, 3, 4])
- 
- self.assertEqual(list(arr[:]), seq)
- 
- self.f(seq)
- 
- p = self.Process(target=self.f, args=(arr,))
- p.start()
- p.join()
- 
- self.assertEqual(list(arr[:]), seq)
- 
- def test_rawarray(self):
- self.test_array(raw=True)
- 
- def test_getobj_getlock_obj(self):
- if self.TYPE != 'processes':
- return
-
- arr1 = self.Array('i', range(10))
- lock1 = arr1.get_lock()
- obj1 = arr1.get_obj()
-
- arr2 = self.Array('i', range(10), lock=None)
- lock2 = arr2.get_lock()
- obj2 = arr2.get_obj()
-
- lock = self.Lock()
- arr3 = self.Array('i', range(10), lock=lock)
- lock3 = arr3.get_lock()
- obj3 = arr3.get_obj()
- self.assertEqual(lock, lock3)
- 
- arr4 = self.RawArray('i', range(10))
- self.assertFalse(hasattr(arr4, 'get_lock'))
- self.assertFalse(hasattr(arr4, 'get_obj'))
-
-#
-#
-#
-
-class _TestContainers(BaseTestCase):
-
- ALLOWED_TYPES = ('manager',)
-
- def test_list(self):
- a = self.list(range(10))
- self.assertEqual(a[:], range(10))
- 
- b = self.list()
- self.assertEqual(b[:], [])
- 
- b.extend(range(5))
- self.assertEqual(b[:], range(5))
- 
- self.assertEqual(b[2], 2)
- self.assertEqual(b[2:10], [2,3,4])
-
- b *= 2
- self.assertEqual(b[:], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4])
-
- self.assertEqual(b + [5, 6], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4, 5, 6])
-
- self.assertEqual(a[:], range(10))
-
- d = [a, b]
- e = self.list(d)
- self.assertEqual(
- e[:],
- [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4]]
- )
- 
- f = self.list([a])
- a.append('hello')
- self.assertEqual(f[:], [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 'hello']])
-
- def test_dict(self):
- d = self.dict()
- indices = range(65, 70)
- for i in indices:
- d[i] = chr(i)
- self.assertEqual(d.copy(), dict((i, chr(i)) for i in indices))
- self.assertEqual(sorted(d.keys()), indices)
- self.assertEqual(sorted(d.values()), [chr(i) for i in indices])
- self.assertEqual(sorted(d.items()), [(i, chr(i)) for i in indices])
- 
- def test_namespace(self):
- n = self.Namespace()
- n.name = 'Bob'
- n.job = 'Builder'
- n._hidden = 'hidden'
- self.assertEqual((n.name, n.job), ('Bob', 'Builder'))
- del n.job
- self.assertEqual(str(n), "Namespace(name='Bob')")
- self.assertTrue(hasattr(n, 'name'))
- self.assertTrue(not hasattr(n, 'job'))
-
-#
-#
-#
-
-def sqr(x, wait=0.0):
- time.sleep(wait)
- return x*x
-
-class _TestPool(BaseTestCase):
-
- def test_apply(self):
- papply = self.pool.apply
- self.assertEqual(papply(sqr, (5,)), sqr(5))
- self.assertEqual(papply(sqr, (), {'x':3}), sqr(x=3))
-
- def test_map(self):
- pmap = self.pool.map
- self.assertEqual(pmap(sqr, range(10)), map(sqr, range(10)))
- self.assertEqual(pmap(sqr, range(100), chunksize=20),
- map(sqr, range(100)))
- 
- def test_async(self):
- res = self.pool.apply_async(sqr, (7, TIMEOUT1,))
- get = TimingWrapper(res.get)
- self.assertEqual(get(), 49)
- self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1)
-
- def test_async_timeout(self):
- res = self.pool.apply_async(sqr, (6, TIMEOUT2 + 0.2))
- get = TimingWrapper(res.get)
- self.assertRaises(multiprocessing.TimeoutError, get, timeout=TIMEOUT2)
- self.assertTimingAlmostEqual(get.elapsed, TIMEOUT2)
-
- def test_imap(self):
- it = self.pool.imap(sqr, range(10))
- self.assertEqual(list(it), map(sqr, range(10)))
-
- it = self.pool.imap(sqr, range(10))
- for i in range(10):
- self.assertEqual(it.next(), i*i)
- self.assertRaises(StopIteration, it.next)
-
- it = self.pool.imap(sqr, range(1000), chunksize=100)
- for i in range(1000):
- self.assertEqual(it.next(), i*i)
- self.assertRaises(StopIteration, it.next)
-
- def test_imap_unordered(self):
- it = self.pool.imap_unordered(sqr, range(1000))
- self.assertEqual(sorted(it), map(sqr, range(1000)))
-
- it = self.pool.imap_unordered(sqr, range(1000), chunksize=53)
- self.assertEqual(sorted(it), map(sqr, range(1000)))
-
- def test_make_pool(self):
- p = multiprocessing.Pool(3)
- self.assertEqual(3, len(p._pool))
- p.close()
- p.join()
-
- def test_terminate(self):
- if self.TYPE == 'manager':
- # On Unix a forked process increfs each shared object to
- # which its parent process held a reference. If the
- # forked process gets terminated then there is likely to
- # be a reference leak. So to prevent
- # _TestZZZNumberOfObjects from failing we skip this test
- # when using a manager.
- return
-
- result = self.pool.map_async(
- time.sleep, [0.1 for i in range(10000)], chunksize=1
- )
- self.pool.terminate()
- join = TimingWrapper(self.pool.join)
- join()
- self.assertTrue(join.elapsed < 0.2)
-
-#
-# Test that manager has expected number of shared objects left
-#
-
-class _TestZZZNumberOfObjects(BaseTestCase):
- # Because test cases are sorted alphabetically, this one will get
- # run after all the other tests for the manager. It tests that
- # there have been no "reference leaks" for the manager's shared
- # objects. Note the comment in _TestPool.test_terminate().
- ALLOWED_TYPES = ('manager',)
-
- def test_number_of_objects(self):
- EXPECTED_NUMBER = 1 # the pool object is still alive
- multiprocessing.active_children() # discard dead process objs
- gc.collect() # do garbage collection
- refs = self.manager._number_of_objects()
- if refs != EXPECTED_NUMBER:
- print self.manager._debugInfo()
-
- self.assertEqual(refs, EXPECTED_NUMBER)
-
-#
-# Test of creating a customized manager class
-#
-
-from multiprocessing.managers import BaseManager, BaseProxy, RemoteError
- 
-class FooBar(object):
- def f(self):
- return 'f()'
- def g(self):
- raise ValueError
- def _h(self):
- return '_h()'
- 
-def baz():
- for i in xrange(10):
- yield i*i
-
-class IteratorProxy(BaseProxy):
- _exposed_ = ('next', '__next__')
- def __iter__(self):
- return self
- def next(self):
- return self._callmethod('next')
- def __next__(self):
- return self._callmethod('__next__')
-
-class MyManager(BaseManager):
- pass
-
-MyManager.register('Foo', callable=FooBar)
-MyManager.register('Bar', callable=FooBar, exposed=('f', '_h'))
-MyManager.register('baz', callable=baz, proxytype=IteratorProxy)
-
-
-class _TestMyManager(BaseTestCase):
- 
- ALLOWED_TYPES = ('manager',)
-
- def test_mymanager(self):
- manager = MyManager()
- manager.start()
- 
- foo = manager.Foo()
- bar = manager.Bar()
- baz = manager.baz()
- 
- foo_methods = [name for name in ('f', 'g', '_h') if hasattr(foo, name)]
- bar_methods = [name for name in ('f', 'g', '_h') if hasattr(bar, name)]
- 
- self.assertEqual(foo_methods, ['f', 'g'])
- self.assertEqual(bar_methods, ['f', '_h'])
- 
- self.assertEqual(foo.f(), 'f()')
- self.assertRaises(ValueError, foo.g)
- self.assertEqual(foo._callmethod('f'), 'f()')
- self.assertRaises(RemoteError, foo._callmethod, '_h')
- 
- self.assertEqual(bar.f(), 'f()')
- self.assertEqual(bar._h(), '_h()')
- self.assertEqual(bar._callmethod('f'), 'f()')
- self.assertEqual(bar._callmethod('_h'), '_h()')
- 
- self.assertEqual(list(baz), [i*i for i in range(10)])
- 
- manager.shutdown()
- 
-#
-# Test of connecting to a remote server and using xmlrpclib for serialization
-#
-
-_queue = Queue.Queue()
-def get_queue():
- return _queue
-
-class QueueManager(BaseManager):
- '''manager class used by server process'''
-QueueManager.register('get_queue', callable=get_queue)
-
-class QueueManager2(BaseManager):
- '''manager class which specifies the same interface as QueueManager'''
-QueueManager2.register('get_queue')
-
-
-SERIALIZER = 'xmlrpclib'
-
-class _TestRemoteManager(BaseTestCase):
-
- ALLOWED_TYPES = ('manager',)
- 
- def _putter(self, address, authkey):
- manager = QueueManager2(
- address=address, authkey=authkey, serializer=SERIALIZER
- )
- manager.connect()
- queue = manager.get_queue()
- queue.put(('hello world', None, True, 2.25))
-
- def test_remote(self):
- authkey = os.urandom(32)
-
- manager = QueueManager(
- address=('localhost', 0), authkey=authkey, serializer=SERIALIZER
- )
- manager.start()
-
- p = self.Process(target=self._putter, args=(manager.address, authkey))
- p.start()
- 
- manager2 = QueueManager2(
- address=manager.address, authkey=authkey, serializer=SERIALIZER
- )
- manager2.connect()
- queue = manager2.get_queue()
- 
- # Note that xmlrpclib will deserialize object as a list not a tuple
- self.assertEqual(queue.get(), ['hello world', None, True, 2.25])
-
- # Because we are using xmlrpclib for serialization instead of
- # pickle this will cause a serialization error.
- self.assertRaises(Exception, queue.put, time.sleep)
-
- # Make queue finalizer run before the server is stopped
- del queue
- manager.shutdown()
-
-#
-#
-#
-
-SENTINEL = latin('')
-
-class _TestConnection(BaseTestCase):
-
- ALLOWED_TYPES = ('processes', 'threads')
-
- def _echo(self, conn):
- for msg in iter(conn.recv_bytes, SENTINEL):
- conn.send_bytes(msg)
- conn.close()
-
- def test_connection(self):
- conn, child_conn = self.Pipe()
- 
- p = self.Process(target=self._echo, args=(child_conn,))
- p.set_daemon(True)
- p.start()
-
- seq = [1, 2.25, None]
- msg = latin('hello world')
- longmsg = msg * 10
- arr = array.array('i', range(4))
-
- if self.TYPE == 'processes':
- self.assertEqual(type(conn.fileno()), int)
-
- self.assertEqual(conn.send(seq), None)
- self.assertEqual(conn.recv(), seq)
-
- self.assertEqual(conn.send_bytes(msg), None)
- self.assertEqual(conn.recv_bytes(), msg)
-
- if self.TYPE == 'processes':
- buffer = array.array('i', [0]*10)
- expected = list(arr) + [0] * (10 - len(arr))
- self.assertEqual(conn.send_bytes(arr), None)
- self.assertEqual(conn.recv_bytes_into(buffer),
- len(arr) * buffer.itemsize)
- self.assertEqual(list(buffer), expected)
-
- buffer = array.array('i', [0]*10)
- expected = [0] * 3 + list(arr) + [0] * (10 - 3 - len(arr))
- self.assertEqual(conn.send_bytes(arr), None)
- self.assertEqual(conn.recv_bytes_into(buffer, 3 * buffer.itemsize),
- len(arr) * buffer.itemsize)
- self.assertEqual(list(buffer), expected)
-
- buffer = bytearray(latin(' ' * 40))
- self.assertEqual(conn.send_bytes(longmsg), None)
- try:
- res = conn.recv_bytes_into(buffer)
- except multiprocessing.BufferTooShort, e:
- self.assertEqual(e.args, (longmsg,))
- else:
- self.fail('expected BufferTooShort, got %s' % res)
-
- poll = TimingWrapper(conn.poll)
-
- self.assertEqual(poll(), False)
- self.assertTimingAlmostEqual(poll.elapsed, 0)
-
- self.assertEqual(poll(TIMEOUT1), False)
- self.assertTimingAlmostEqual(poll.elapsed, TIMEOUT1)
-
- conn.send(None)
-
- self.assertEqual(poll(TIMEOUT1), True)
- self.assertTimingAlmostEqual(poll.elapsed, 0)
- 
- self.assertEqual(conn.recv(), None)
-
- really_big_msg = latin('X') * (1024 * 1024 * 16) # 16Mb
- conn.send_bytes(really_big_msg)
- self.assertEqual(conn.recv_bytes(), really_big_msg)
- 
- conn.send_bytes(SENTINEL) # tell child to quit
- child_conn.close()
-
- if self.TYPE == 'processes':
- self.assertEqual(conn.readable, True)
- self.assertEqual(conn.writable, True)
- self.assertRaises(EOFError, conn.recv)
- self.assertRaises(EOFError, conn.recv_bytes)
-
- p.join()
- 
- def test_duplex_false(self):
- reader, writer = self.Pipe(duplex=False)
- self.assertEqual(writer.send(1), None)
- self.assertEqual(reader.recv(), 1)
- if self.TYPE == 'processes':
- self.assertEqual(reader.readable, True)
- self.assertEqual(reader.writable, False)
- self.assertEqual(writer.readable, False)
- self.assertEqual(writer.writable, True)
- self.assertRaises(IOError, reader.send, 2)
- self.assertRaises(IOError, writer.recv)
- self.assertRaises(IOError, writer.poll)
-
- def test_spawn_close(self):
- # We test that a pipe connection can be closed by parent
- # process immediately after child is spawned. On Windows this
- # would have sometimes failed on old versions because
- # child_conn would be closed before the child got a chance to
- # duplicate it.
- conn, child_conn = self.Pipe()
- 
- p = self.Process(target=self._echo, args=(child_conn,))
- p.start()
- child_conn.close() # this might complete before child initializes
-
- msg = latin('hello')
- conn.send_bytes(msg)
- self.assertEqual(conn.recv_bytes(), msg)
-
- conn.send_bytes(SENTINEL)
- conn.close()
- p.join()
-
- def test_sendbytes(self):
- if self.TYPE != 'processes':
- return
-
- msg = latin('abcdefghijklmnopqrstuvwxyz')
- a, b = self.Pipe()
- 
- a.send_bytes(msg)
- self.assertEqual(b.recv_bytes(), msg)
-
- a.send_bytes(msg, 5)
- self.assertEqual(b.recv_bytes(), msg[5:])
-
- a.send_bytes(msg, 7, 8)
- self.assertEqual(b.recv_bytes(), msg[7:7+8])
-
- a.send_bytes(msg, 26)
- self.assertEqual(b.recv_bytes(), latin(''))
-
- a.send_bytes(msg, 26, 0)
- self.assertEqual(b.recv_bytes(), latin(''))
-
- self.assertRaises(ValueError, a.send_bytes, msg, 27)
- 
- self.assertRaises(ValueError, a.send_bytes, msg, 22, 5)
- 
- self.assertRaises(ValueError, a.send_bytes, msg, 26, 1)
-
- self.assertRaises(ValueError, a.send_bytes, msg, -1)
-
- self.assertRaises(ValueError, a.send_bytes, msg, 4, -1)
- 
-
-class _TestListenerClient(BaseTestCase):
-
- ALLOWED_TYPES = ('processes', 'threads')
-
- def _test(self, address):
- conn = self.connection.Client(address)
- conn.send('hello')
- conn.close()
-
- def test_listener_client(self): 
- for family in self.connection.families:
- l = self.connection.Listener(family=family)
- p = self.Process(target=self._test, args=(l.address,))
- p.set_daemon(True)
- p.start()
- conn = l.accept()
- self.assertEqual(conn.recv(), 'hello')
- p.join()
- l.close()
-
-#
-# Test of sending connection and socket objects between processes
-#
-
-class _TestPicklingConnections(BaseTestCase):
-
- ALLOWED_TYPES = ('processes',)
-
- def _listener(self, conn, families):
- for fam in families:
- l = self.connection.Listener(family=fam)
- conn.send(l.address)
- new_conn = l.accept()
- conn.send(new_conn)
-
- if self.TYPE == 'processes':
- l = socket.socket()
- l.bind(('localhost', 0))
- conn.send(l.getsockname())
- l.listen(1)
- new_conn, addr = l.accept()
- conn.send(new_conn)
- 
- conn.recv()
-
- def _remote(self, conn):
- for (address, msg) in iter(conn.recv, None):
- client = self.connection.Client(address)
- client.send(msg.upper())
- client.close()
-
- if self.TYPE == 'processes':
- address, msg = conn.recv()
- client = socket.socket()
- client.connect(address)
- client.sendall(msg.upper())
- client.close()
-
- conn.close()
-
- def test_pickling(self):
- try:
- multiprocessing.allow_connection_pickling()
- except ImportError:
- return
- 
- families = self.connection.families
-
- lconn, lconn0 = self.Pipe()
- lp = self.Process(target=self._listener, args=(lconn0, families))
- lp.start()
- lconn0.close()
-
- rconn, rconn0 = self.Pipe()
- rp = self.Process(target=self._remote, args=(rconn0,))
- rp.start()
- rconn0.close()
-
- for fam in families:
- msg = ('This connection uses family %s' % fam).encode('ascii')
- address = lconn.recv()
- rconn.send((address, msg))
- new_conn = lconn.recv()
- self.assertEqual(new_conn.recv(), msg.upper())
- 
- rconn.send(None)
-
- if self.TYPE == 'processes':
- msg = latin('This connection uses a normal socket')
- address = lconn.recv()
- rconn.send((address, msg))
- if hasattr(socket, 'fromfd'):
- new_conn = lconn.recv()
- self.assertEqual(new_conn.recv(100), msg.upper())
- else:
- # XXX On Windows with Py2.6 need to backport fromfd()
- discard = lconn.recv_bytes()
- 
- lconn.send(None)
- 
- rconn.close()
- lconn.close()
- 
- lp.join()
- rp.join()
-
-#
-#
-#
-
-class _TestHeap(BaseTestCase):
-
- ALLOWED_TYPES = ('processes',)
-
- def test_heap(self):
- iterations = 5000
- maxblocks = 50
- blocks = []
-
- # create and destroy lots of blocks of different sizes
- for i in xrange(iterations):
- size = int(random.lognormvariate(0, 1) * 1000)
- b = multiprocessing.heap.BufferWrapper(size)
- blocks.append(b)
- if len(blocks) > maxblocks:
- i = random.randrange(maxblocks)
- del blocks[i]
-
- # get the heap object
- heap = multiprocessing.heap.BufferWrapper._heap
-
- # verify the state of the heap
- all = []
- occupied = 0
- for L in heap._len_to_seq.values():
- for arena, start, stop in L:
- all.append((heap._arenas.index(arena), start, stop,
- stop-start, 'free'))
- for arena, start, stop in heap._allocated_blocks:
- all.append((heap._arenas.index(arena), start, stop,
- stop-start, 'occupied'))
- occupied += (stop-start)
-
- all.sort()
-
- for i in range(len(all)-1):
- (arena, start, stop) = all[i][:3]
- (narena, nstart, nstop) = all[i+1][:3]
- self.assertTrue((arena != narena and nstart == 0) or
- (stop == nstart))
- 
-#
-#
-#
-
-try:
- from ctypes import Structure, Value, copy, c_int, c_double
-except ImportError:
- Structure = object
- c_int = c_double = None
-
-class _Foo(Structure):
- _fields_ = [
- ('x', c_int),
- ('y', c_double)
- ]
-
-class _TestSharedCTypes(BaseTestCase):
-
- ALLOWED_TYPES = ('processes',)
-
- def _double(self, x, y, foo, arr, string):
- x.value *= 2
- y.value *= 2
- foo.x *= 2
- foo.y *= 2
- string.value *= 2
- for i in range(len(arr)):
- arr[i] *= 2
-
- def test_sharedctypes(self, lock=False):
- if c_int is None:
- return
- 
- x = Value('i', 7, lock=lock)
- y = Value(ctypes.c_double, 1.0/3.0, lock=lock)
- foo = Value(_Foo, 3, 2, lock=lock)
- arr = Array('d', range(10), lock=lock)
- string = Array('c', 20, lock=lock)
- string.value = 'hello'
-
- p = self.Process(target=self._double, args=(x, y, foo, arr, string))
- p.start()
- p.join()
-
- self.assertEqual(x.value, 14)
- self.assertAlmostEqual(y.value, 2.0/3.0)
- self.assertEqual(foo.x, 6)
- self.assertAlmostEqual(foo.y, 4.0)
- for i in range(10):
- self.assertAlmostEqual(arr[i], i*2)
- self.assertEqual(string.value, latin('hellohello'))
-
- def test_synchronize(self):
- self.test_sharedctypes(lock=True)
-
- def test_copy(self):
- if c_int is None:
- return
-
- foo = _Foo(2, 5.0)
- bar = copy(foo)
- foo.x = 0
- foo.y = 0
- self.assertEqual(bar.x, 2)
- self.assertAlmostEqual(bar.y, 5.0)
-
-#
-#
-#
-
-class _TestFinalize(BaseTestCase):
-
- ALLOWED_TYPES = ('processes',)
-
- def _test_finalize(self, conn):
- class Foo(object):
- pass
-
- a = Foo()
- util.Finalize(a, conn.send, args=('a',))
- del a # triggers callback for a
-
- b = Foo()
- close_b = util.Finalize(b, conn.send, args=('b',)) 
- close_b() # triggers callback for b
- close_b() # does nothing because callback has already been called
- del b # does nothing because callback has already been called
-
- c = Foo()
- util.Finalize(c, conn.send, args=('c',))
-
- d10 = Foo()
- util.Finalize(d10, conn.send, args=('d10',), exitpriority=1)
-
- d01 = Foo()
- util.Finalize(d01, conn.send, args=('d01',), exitpriority=0)
- d02 = Foo()
- util.Finalize(d02, conn.send, args=('d02',), exitpriority=0)
- d03 = Foo()
- util.Finalize(d03, conn.send, args=('d03',), exitpriority=0)
-
- util.Finalize(None, conn.send, args=('e',), exitpriority=-10)
-
- util.Finalize(None, conn.send, args=('STOP',), exitpriority=-100)
-
- # call mutliprocessing's cleanup function then exit process without
- # garbage collecting locals
- util._exit_function()
- conn.close()
- os._exit(0)
-
- def test_finalize(self):
- conn, child_conn = self.Pipe()
- 
- p = self.Process(target=self._test_finalize, args=(child_conn,))
- p.start()
- p.join()
-
- result = [obj for obj in iter(conn.recv, 'STOP')]
- self.assertEqual(result, ['a', 'b', 'd10', 'd03', 'd02', 'd01', 'e'])
-
-#
-# Test that from ... import * works for each module
-#
-
-class _TestImportStar(BaseTestCase):
-
- ALLOWED_TYPES = ('processes',)
-
- def test_import(self):
- modules = (
- 'multiprocessing', 'multiprocessing.connection',
- 'multiprocessing.heap', 'multiprocessing.managers',
- 'multiprocessing.pool', 'multiprocessing.process',
- 'multiprocessing.reduction', 'multiprocessing.sharedctypes',
- 'multiprocessing.synchronize', 'multiprocessing.util'
- )
- 
- for name in modules:
- __import__(name)
- mod = sys.modules[name]
- 
- for attr in getattr(mod, '__all__', ()):
- self.assertTrue(
- hasattr(mod, attr),
- '%r does not have attribute %r' % (mod, attr)
- )
-
-#
-# Quick test that logging works -- does not test logging output
-#
-
-class _TestLogging(BaseTestCase):
-
- ALLOWED_TYPES = ('processes',)
-
- def test_enable_logging(self):
- logger = multiprocessing.get_logger()
- logger.setLevel(util.SUBWARNING)
- self.assertTrue(logger is not None)
- logger.debug('this will not be printed')
- logger.info('nor will this')
- logger.setLevel(LOG_LEVEL)
-
- def _test_level(self, conn):
- logger = multiprocessing.get_logger()
- conn.send(logger.getEffectiveLevel())
-
- def test_level(self):
- LEVEL1 = 32
- LEVEL2 = 37
- 
- logger = multiprocessing.get_logger()
- root_logger = logging.getLogger()
- root_level = root_logger.level
-
- reader, writer = multiprocessing.Pipe(duplex=False)
-
- logger.setLevel(LEVEL1)
- self.Process(target=self._test_level, args=(writer,)).start()
- self.assertEqual(LEVEL1, reader.recv())
-
- logger.setLevel(logging.NOTSET)
- root_logger.setLevel(LEVEL2)
- self.Process(target=self._test_level, args=(writer,)).start()
- self.assertEqual(LEVEL2, reader.recv())
-
- root_logger.setLevel(root_level)
- logger.setLevel(level=LOG_LEVEL)
-
-#
-# Functions used to create test cases from the base ones in this module
-#
-
-def get_attributes(Source, names):
- d = {}
- for name in names:
- obj = getattr(Source, name)
- if type(obj) == type(get_attributes):
- obj = staticmethod(obj)
- d[name] = obj
- return d
-
-def create_test_cases(Mixin, type):
- result = {}
- glob = globals()
- Type = type[0].upper() + type[1:]
-
- for name in glob.keys():
- if name.startswith('_Test'):
- base = glob[name]
- if type in base.ALLOWED_TYPES:
- newname = 'With' + Type + name[1:]
- class Temp(base, unittest.TestCase, Mixin):
- pass
- result[newname] = Temp
- Temp.__name__ = newname
- Temp.__module__ = Mixin.__module__
- return result
-
-#
-# Create test cases
-#
-
-class ProcessesMixin(object):
- TYPE = 'processes'
- Process = multiprocessing.Process
- locals().update(get_attributes(multiprocessing, (
- 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore',
- 'Condition', 'Event', 'Value', 'Array', 'RawValue',
- 'RawArray', 'current_process', 'active_children', 'Pipe',
- 'connection', 'JoinableQueue'
- )))
-
-testcases_processes = create_test_cases(ProcessesMixin, type='processes')
-globals().update(testcases_processes)
-
-
-class ManagerMixin(object):
- TYPE = 'manager'
- Process = multiprocessing.Process
- manager = object.__new__(multiprocessing.managers.SyncManager)
- locals().update(get_attributes(manager, (
- 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 
- 'Condition', 'Event', 'Value', 'Array', 'list', 'dict',
- 'Namespace', 'JoinableQueue'
- )))
-
-testcases_manager = create_test_cases(ManagerMixin, type='manager')
-globals().update(testcases_manager)
-
-
-class ThreadsMixin(object):
- TYPE = 'threads'
- Process = multiprocessing.dummy.Process
- locals().update(get_attributes(multiprocessing.dummy, (
- 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore',
- 'Condition', 'Event', 'Value', 'Array', 'current_process',
- 'active_children', 'Pipe', 'connection', 'dict', 'list',
- 'Namespace', 'JoinableQueue'
- )))
-
-testcases_threads = create_test_cases(ThreadsMixin, type='threads')
-globals().update(testcases_threads)
-
-#
-#
-#
-
-def test_main(run=None):
- if run is None:
- from test.test_support import run_unittest as run
-
- util.get_temp_dir() # creates temp directory for use by all processes
- 
- multiprocessing.get_logger().setLevel(LOG_LEVEL)
-
- ProcessesMixin.pool = multiprocessing.Pool(4)
- ThreadsMixin.pool = multiprocessing.dummy.Pool(4)
- ManagerMixin.manager.__init__()
- ManagerMixin.manager.start()
- ManagerMixin.pool = ManagerMixin.manager.Pool(4)
-
- testcases = (
- sorted(testcases_processes.values(), key=lambda tc:tc.__name__) +
- sorted(testcases_threads.values(), key=lambda tc:tc.__name__) +
- sorted(testcases_manager.values(), key=lambda tc:tc.__name__)
- )
-
- loadTestsFromTestCase = unittest.defaultTestLoader.loadTestsFromTestCase
- suite = unittest.TestSuite(loadTestsFromTestCase(tc) for tc in testcases)
- run(suite)
-
- ThreadsMixin.pool.terminate()
- ProcessesMixin.pool.terminate()
- ManagerMixin.pool.terminate()
- ManagerMixin.manager.shutdown()
- 
- del ProcessesMixin.pool, ThreadsMixin.pool, ManagerMixin.pool
-
-def main():
- test_main(unittest.TextTestRunner(verbosity=2).run)
-
-if __name__ == '__main__':
- main()
+#
+# Unit tests for the multiprocessing package
+#
+
+import unittest
+import threading
+import Queue
+import time
+import sys
+import os
+import gc
+import signal
+import array
+import copy
+import socket
+import random
+import logging
+
+import multiprocessing.dummy
+import multiprocessing.connection
+import multiprocessing.managers
+import multiprocessing.heap
+import multiprocessing.managers
+import multiprocessing.pool
+import _multiprocessing
+
+from multiprocessing import util
+
+#
+#
+#
+
+if sys.version_info >= (3, 0):
+ def latin(s):
+ return s.encode('latin')
+else:
+ latin = str
+
+try:
+ bytes
+except NameError:
+ bytes = str
+ def bytearray(seq):
+ return array.array('c', seq)
+
+#
+# Constants
+#
+
+LOG_LEVEL = util.SUBWARNING
+#LOG_LEVEL = logging.WARNING
+
+DELTA = 0.1
+CHECK_TIMINGS = False # making true makes tests take a lot longer
+ # and can sometimes cause some non-serious
+ # failures because some calls block a bit
+ # longer than expected
+if CHECK_TIMINGS:
+ TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.82, 0.35, 1.4
+else:
+ TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.1, 0.1, 0.1
+
+HAVE_GETVALUE = not getattr(_multiprocessing,
+ 'HAVE_BROKEN_SEM_GETVALUE', False)
+
+#
+# Creates a wrapper for a function which records the time it takes to finish
+#
+
+class TimingWrapper(object):
+
+ def __init__(self, func):
+ self.func = func
+ self.elapsed = None
+
+ def __call__(self, *args, **kwds):
+ t = time.time()
+ try:
+ return self.func(*args, **kwds)
+ finally:
+ self.elapsed = time.time() - t
+
+#
+# Base class for test cases
+#
+
+class BaseTestCase(object):
+
+ ALLOWED_TYPES = ('processes', 'manager', 'threads')
+
+ def assertTimingAlmostEqual(self, a, b):
+ if CHECK_TIMINGS:
+ self.assertAlmostEqual(a, b, 1)
+
+ def assertReturnsIfImplemented(self, value, func, *args):
+ try:
+ res = func(*args)
+ except NotImplementedError:
+ pass
+ else:
+ return self.assertEqual(value, res)
+
+#
+# Return the value of a semaphore
+#
+
+def get_value(self):
+ try:
+ return self.get_value()
+ except AttributeError:
+ try:
+ return self._Semaphore__value
+ except AttributeError:
+ try:
+ return self._value
+ except AttributeError:
+ raise NotImplementedError
+
+#
+# Testcases
+#
+
+class _TestProcess(BaseTestCase):
+
+ ALLOWED_TYPES = ('processes', 'threads')
+
+ def test_current(self):
+ if self.TYPE == 'threads':
+ return
+
+ current = self.current_process()
+ authkey = current.get_authkey()
+
+ self.assertTrue(current.is_alive())
+ self.assertTrue(not current.is_daemon())
+ self.assertTrue(isinstance(authkey, bytes))
+ self.assertTrue(len(authkey) > 0)
+ self.assertEqual(current.get_ident(), os.getpid())
+ self.assertEqual(current.get_exitcode(), None)
+
+ def _test(self, q, *args, **kwds):
+ current = self.current_process()
+ q.put(args)
+ q.put(kwds)
+ q.put(current.get_name())
+ if self.TYPE != 'threads':
+ q.put(bytes(current.get_authkey()))
+ q.put(current.pid)
+
+ def test_process(self):
+ q = self.Queue(1)
+ e = self.Event()
+ args = (q, 1, 2)
+ kwargs = {'hello':23, 'bye':2.54}
+ name = 'SomeProcess'
+ p = self.Process(
+ target=self._test, args=args, kwargs=kwargs, name=name
+ )
+ p.set_daemon(True)
+ current = self.current_process()
+
+ if self.TYPE != 'threads':
+ self.assertEquals(p.get_authkey(), current.get_authkey())
+ self.assertEquals(p.is_alive(), False)
+ self.assertEquals(p.is_daemon(), True)
+ self.assertTrue(p not in self.active_children())
+ self.assertTrue(type(self.active_children()) is list)
+ self.assertEqual(p.get_exitcode(), None)
+
+ p.start()
+
+ self.assertEquals(p.get_exitcode(), None)
+ self.assertEquals(p.is_alive(), True)
+ self.assertTrue(p in self.active_children())
+
+ self.assertEquals(q.get(), args[1:])
+ self.assertEquals(q.get(), kwargs)
+ self.assertEquals(q.get(), p.get_name())
+ if self.TYPE != 'threads':
+ self.assertEquals(q.get(), current.get_authkey())
+ self.assertEquals(q.get(), p.pid)
+
+ p.join()
+
+ self.assertEquals(p.get_exitcode(), 0)
+ self.assertEquals(p.is_alive(), False)
+ self.assertTrue(p not in self.active_children())
+
+ def _test_terminate(self):
+ time.sleep(1000)
+
+ def test_terminate(self):
+ if self.TYPE == 'threads':
+ return
+
+ p = self.Process(target=self._test_terminate)
+ p.set_daemon(True)
+ p.start()
+
+ self.assertEqual(p.is_alive(), True)
+ self.assertTrue(p in self.active_children())
+ self.assertEqual(p.get_exitcode(), None)
+
+ p.terminate()
+
+ join = TimingWrapper(p.join)
+ self.assertEqual(join(), None)
+ self.assertTimingAlmostEqual(join.elapsed, 0.0)
+
+ self.assertEqual(p.is_alive(), False)
+ self.assertTrue(p not in self.active_children())
+
+ p.join()
+
+ # XXX sometimes get p.get_exitcode() == 0 on Windows ...
+ #self.assertEqual(p.get_exitcode(), -signal.SIGTERM)
+
+ def test_cpu_count(self):
+ try:
+ cpus = multiprocessing.cpu_count()
+ except NotImplementedError:
+ cpus = 1
+ self.assertTrue(type(cpus) is int)
+ self.assertTrue(cpus >= 1)
+
+ def test_active_children(self):
+ self.assertEqual(type(self.active_children()), list)
+
+ p = self.Process(target=time.sleep, args=(DELTA,))
+ self.assertTrue(p not in self.active_children())
+
+ p.start()
+ self.assertTrue(p in self.active_children())
+
+ p.join()
+ self.assertTrue(p not in self.active_children())
+
+ def _test_recursion(self, wconn, id):
+ from multiprocessing import forking
+ wconn.send(id)
+ if len(id) < 2:
+ for i in range(2):
+ p = self.Process(
+ target=self._test_recursion, args=(wconn, id+[i])
+ )
+ p.start()
+ p.join()
+
+ def test_recursion(self):
+ rconn, wconn = self.Pipe(duplex=False)
+ self._test_recursion(wconn, [])
+
+ time.sleep(DELTA)
+ result = []
+ while rconn.poll():
+ result.append(rconn.recv())
+
+ expected = [
+ [],
+ [0],
+ [0, 0],
+ [0, 1],
+ [1],
+ [1, 0],
+ [1, 1]
+ ]
+ self.assertEqual(result, expected)
+
+#
+#
+#
+
+class _UpperCaser(multiprocessing.Process):
+
+ def __init__(self):
+ multiprocessing.Process.__init__(self)
+ self.child_conn, self.parent_conn = multiprocessing.Pipe()
+
+ def run(self):
+ self.parent_conn.close()
+ for s in iter(self.child_conn.recv, None):
+ self.child_conn.send(s.upper())
+ self.child_conn.close()
+
+ def submit(self, s):
+ assert type(s) is str
+ self.parent_conn.send(s)
+ return self.parent_conn.recv()
+
+ def stop(self):
+ self.parent_conn.send(None)
+ self.parent_conn.close()
+ self.child_conn.close()
+
+class _TestSubclassingProcess(BaseTestCase):
+
+ ALLOWED_TYPES = ('processes',)
+
+ def test_subclassing(self):
+ uppercaser = _UpperCaser()
+ uppercaser.start()
+ self.assertEqual(uppercaser.submit('hello'), 'HELLO')
+ self.assertEqual(uppercaser.submit('world'), 'WORLD')
+ uppercaser.stop()
+ uppercaser.join()
+
+#
+#
+#
+
+def queue_empty(q):
+ if hasattr(q, 'empty'):
+ return q.empty()
+ else:
+ return q.qsize() == 0
+
+def queue_full(q, maxsize):
+ if hasattr(q, 'full'):
+ return q.full()
+ else:
+ return q.qsize() == maxsize
+
+
+class _TestQueue(BaseTestCase):
+
+
+ def _test_put(self, queue, child_can_start, parent_can_continue):
+ child_can_start.wait()
+ for i in range(6):
+ queue.get()
+ parent_can_continue.set()
+
+ def test_put(self):
+ MAXSIZE = 6
+ queue = self.Queue(maxsize=MAXSIZE)
+ child_can_start = self.Event()
+ parent_can_continue = self.Event()
+
+ proc = self.Process(
+ target=self._test_put,
+ args=(queue, child_can_start, parent_can_continue)
+ )
+ proc.set_daemon(True)
+ proc.start()
+
+ self.assertEqual(queue_empty(queue), True)
+ self.assertEqual(queue_full(queue, MAXSIZE), False)
+
+ queue.put(1)
+ queue.put(2, True)
+ queue.put(3, True, None)
+ queue.put(4, False)
+ queue.put(5, False, None)
+ queue.put_nowait(6)
+
+ # the values may be in buffer but not yet in pipe so sleep a bit
+ time.sleep(DELTA)
+
+ self.assertEqual(queue_empty(queue), False)
+ self.assertEqual(queue_full(queue, MAXSIZE), True)
+
+ put = TimingWrapper(queue.put)
+ put_nowait = TimingWrapper(queue.put_nowait)
+
+ self.assertRaises(Queue.Full, put, 7, False)
+ self.assertTimingAlmostEqual(put.elapsed, 0)
+
+ self.assertRaises(Queue.Full, put, 7, False, None)
+ self.assertTimingAlmostEqual(put.elapsed, 0)
+
+ self.assertRaises(Queue.Full, put_nowait, 7)
+ self.assertTimingAlmostEqual(put_nowait.elapsed, 0)
+
+ self.assertRaises(Queue.Full, put, 7, True, TIMEOUT1)
+ self.assertTimingAlmostEqual(put.elapsed, TIMEOUT1)
+
+ self.assertRaises(Queue.Full, put, 7, False, TIMEOUT2)
+ self.assertTimingAlmostEqual(put.elapsed, 0)
+
+ self.assertRaises(Queue.Full, put, 7, True, timeout=TIMEOUT3)
+ self.assertTimingAlmostEqual(put.elapsed, TIMEOUT3)
+
+ child_can_start.set()
+ parent_can_continue.wait()
+
+ self.assertEqual(queue_empty(queue), True)
+ self.assertEqual(queue_full(queue, MAXSIZE), False)
+
+ proc.join()
+
+ def _test_get(self, queue, child_can_start, parent_can_continue):
+ child_can_start.wait()
+ queue.put(1)
+ queue.put(2)
+ queue.put(3)
+ queue.put(4)
+ queue.put(5)
+ parent_can_continue.set()
+
+ def test_get(self):
+ queue = self.Queue()
+ child_can_start = self.Event()
+ parent_can_continue = self.Event()
+
+ proc = self.Process(
+ target=self._test_get,
+ args=(queue, child_can_start, parent_can_continue)
+ )
+ proc.set_daemon(True)
+ proc.start()
+
+ self.assertEqual(queue_empty(queue), True)
+
+ child_can_start.set()
+ parent_can_continue.wait()
+
+ time.sleep(DELTA)
+ self.assertEqual(queue_empty(queue), False)
+
+ self.assertEqual(queue.get(), 1)
+ self.assertEqual(queue.get(True, None), 2)
+ self.assertEqual(queue.get(True), 3)
+ self.assertEqual(queue.get(timeout=1), 4)
+ self.assertEqual(queue.get_nowait(), 5)
+
+ self.assertEqual(queue_empty(queue), True)
+
+ get = TimingWrapper(queue.get)
+ get_nowait = TimingWrapper(queue.get_nowait)
+
+ self.assertRaises(Queue.Empty, get, False)
+ self.assertTimingAlmostEqual(get.elapsed, 0)
+
+ self.assertRaises(Queue.Empty, get, False, None)
+ self.assertTimingAlmostEqual(get.elapsed, 0)
+
+ self.assertRaises(Queue.Empty, get_nowait)
+ self.assertTimingAlmostEqual(get_nowait.elapsed, 0)
+
+ self.assertRaises(Queue.Empty, get, True, TIMEOUT1)
+ self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1)
+
+ self.assertRaises(Queue.Empty, get, False, TIMEOUT2)
+ self.assertTimingAlmostEqual(get.elapsed, 0)
+
+ self.assertRaises(Queue.Empty, get, timeout=TIMEOUT3)
+ self.assertTimingAlmostEqual(get.elapsed, TIMEOUT3)
+
+ proc.join()
+
+ def _test_fork(self, queue):
+ for i in range(10, 20):
+ queue.put(i)
+ # note that at this point the items may only be buffered, so the
+ # process cannot shutdown until the feeder thread has finished
+ # pushing items onto the pipe.
+
+ def test_fork(self):
+ # Old versions of Queue would fail to create a new feeder
+ # thread for a forked process if the original process had its
+ # own feeder thread. This test checks that this no longer
+ # happens.
+
+ queue = self.Queue()
+
+ # put items on queue so that main process starts a feeder thread
+ for i in range(10):
+ queue.put(i)
+
+ # wait to make sure thread starts before we fork a new process
+ time.sleep(DELTA)
+
+ # fork process
+ p = self.Process(target=self._test_fork, args=(queue,))
+ p.start()
+
+ # check that all expected items are in the queue
+ for i in range(20):
+ self.assertEqual(queue.get(), i)
+ self.assertRaises(Queue.Empty, queue.get, False)
+
+ p.join()
+
+ def test_qsize(self):
+ q = self.Queue()
+ try:
+ self.assertEqual(q.qsize(), 0)
+ except NotImplementedError:
+ return
+ q.put(1)
+ self.assertEqual(q.qsize(), 1)
+ q.put(5)
+ self.assertEqual(q.qsize(), 2)
+ q.get()
+ self.assertEqual(q.qsize(), 1)
+ q.get()
+ self.assertEqual(q.qsize(), 0)
+
+ def _test_task_done(self, q):
+ for obj in iter(q.get, None):
+ time.sleep(DELTA)
+ q.task_done()
+
+ def test_task_done(self):
+ queue = self.JoinableQueue()
+
+ if sys.version_info < (2, 5) and not hasattr(queue, 'task_done'):
+ return
+
+ workers = [self.Process(target=self._test_task_done, args=(queue,))
+ for i in xrange(4)]
+
+ for p in workers:
+ p.start()
+
+ for i in xrange(10):
+ queue.put(i)
+
+ queue.join()
+
+ for p in workers:
+ queue.put(None)
+
+ for p in workers:
+ p.join()
+
+#
+#
+#
+
+class _TestLock(BaseTestCase):
+
+ def test_lock(self):
+ lock = self.Lock()
+ self.assertEqual(lock.acquire(), True)
+ self.assertEqual(lock.acquire(False), False)
+ self.assertEqual(lock.release(), None)
+ self.assertRaises((ValueError, threading.ThreadError), lock.release)
+
+ def test_rlock(self):
+ lock = self.RLock()
+ self.assertEqual(lock.acquire(), True)
+ self.assertEqual(lock.acquire(), True)
+ self.assertEqual(lock.acquire(), True)
+ self.assertEqual(lock.release(), None)
+ self.assertEqual(lock.release(), None)
+ self.assertEqual(lock.release(), None)
+ self.assertRaises((AssertionError, RuntimeError), lock.release)
+
+
+class _TestSemaphore(BaseTestCase):
+
+ def _test_semaphore(self, sem):
+ self.assertReturnsIfImplemented(2, get_value, sem)
+ self.assertEqual(sem.acquire(), True)
+ self.assertReturnsIfImplemented(1, get_value, sem)
+ self.assertEqual(sem.acquire(), True)
+ self.assertReturnsIfImplemented(0, get_value, sem)
+ self.assertEqual(sem.acquire(False), False)
+ self.assertReturnsIfImplemented(0, get_value, sem)
+ self.assertEqual(sem.release(), None)
+ self.assertReturnsIfImplemented(1, get_value, sem)
+ self.assertEqual(sem.release(), None)
+ self.assertReturnsIfImplemented(2, get_value, sem)
+
+ def test_semaphore(self):
+ sem = self.Semaphore(2)
+ self._test_semaphore(sem)
+ self.assertEqual(sem.release(), None)
+ self.assertReturnsIfImplemented(3, get_value, sem)
+ self.assertEqual(sem.release(), None)
+ self.assertReturnsIfImplemented(4, get_value, sem)
+
+ def test_bounded_semaphore(self):
+ sem = self.BoundedSemaphore(2)
+ self._test_semaphore(sem)
+ # Currently fails on OS/X
+ #if HAVE_GETVALUE:
+ # self.assertRaises(ValueError, sem.release)
+ # self.assertReturnsIfImplemented(2, get_value, sem)
+
+ def test_timeout(self):
+ if self.TYPE != 'processes':
+ return
+
+ sem = self.Semaphore(0)
+ acquire = TimingWrapper(sem.acquire)
+
+ self.assertEqual(acquire(False), False)
+ self.assertTimingAlmostEqual(acquire.elapsed, 0.0)
+
+ self.assertEqual(acquire(False, None), False)
+ self.assertTimingAlmostEqual(acquire.elapsed, 0.0)
+
+ self.assertEqual(acquire(False, TIMEOUT1), False)
+ self.assertTimingAlmostEqual(acquire.elapsed, 0)
+
+ self.assertEqual(acquire(True, TIMEOUT2), False)
+ self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT2)
+
+ self.assertEqual(acquire(timeout=TIMEOUT3), False)
+ self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT3)
+
+
+class _TestCondition(BaseTestCase):
+
+ def f(self, cond, sleeping, woken, timeout=None):
+ cond.acquire()
+ sleeping.release()
+ cond.wait(timeout)
+ woken.release()
+ cond.release()
+
+ def check_invariant(self, cond):
+ # this is only supposed to succeed when there are no sleepers
+ if self.TYPE == 'processes':
+ try:
+ sleepers = (cond._sleeping_count.get_value() -
+ cond._woken_count.get_value())
+ self.assertEqual(sleepers, 0)
+ self.assertEqual(cond._wait_semaphore.get_value(), 0)
+ except NotImplementedError:
+ pass
+
+ def test_notify(self):
+ cond = self.Condition()
+ sleeping = self.Semaphore(0)
+ woken = self.Semaphore(0)
+
+ p = self.Process(target=self.f, args=(cond, sleeping, woken))
+ p.set_daemon(True)
+ p.start()
+
+ p = threading.Thread(target=self.f, args=(cond, sleeping, woken))
+ p.set_daemon(True)
+ p.start()
+
+ # wait for both children to start sleeping
+ sleeping.acquire()
+ sleeping.acquire()
+
+ # check no process/thread has woken up
+ time.sleep(DELTA)
+ self.assertReturnsIfImplemented(0, get_value, woken)
+
+ # wake up one process/thread
+ cond.acquire()
+ cond.notify()
+ cond.release()
+
+ # check one process/thread has woken up
+ time.sleep(DELTA)
+ self.assertReturnsIfImplemented(1, get_value, woken)
+
+ # wake up another
+ cond.acquire()
+ cond.notify()
+ cond.release()
+
+ # check other has woken up
+ time.sleep(DELTA)
+ self.assertReturnsIfImplemented(2, get_value, woken)
+
+ # check state is not mucked up
+ self.check_invariant(cond)
+ p.join()
+
+ def test_notify_all(self):
+ cond = self.Condition()
+ sleeping = self.Semaphore(0)
+ woken = self.Semaphore(0)
+
+ # start some threads/processes which will timeout
+ for i in range(3):
+ p = self.Process(target=self.f,
+ args=(cond, sleeping, woken, TIMEOUT1))
+ p.set_daemon(True)
+ p.start()
+
+ t = threading.Thread(target=self.f,
+ args=(cond, sleeping, woken, TIMEOUT1))
+ t.set_daemon(True)
+ t.start()
+
+ # wait for them all to sleep
+ for i in xrange(6):
+ sleeping.acquire()
+
+ # check they have all timed out
+ for i in xrange(6):
+ woken.acquire()
+ self.assertReturnsIfImplemented(0, get_value, woken)
+
+ # check state is not mucked up
+ self.check_invariant(cond)
+
+ # start some more threads/processes
+ for i in range(3):
+ p = self.Process(target=self.f, args=(cond, sleeping, woken))
+ p.set_daemon(True)
+ p.start()
+
+ t = threading.Thread(target=self.f, args=(cond, sleeping, woken))
+ t.set_daemon(True)
+ t.start()
+
+ # wait for them to all sleep
+ for i in xrange(6):
+ sleeping.acquire()
+
+ # check no process/thread has woken up
+ time.sleep(DELTA)
+ self.assertReturnsIfImplemented(0, get_value, woken)
+
+ # wake them all up
+ cond.acquire()
+ cond.notify_all()
+ cond.release()
+
+ # check they have all woken
+ time.sleep(DELTA)
+ self.assertReturnsIfImplemented(6, get_value, woken)
+
+ # check state is not mucked up
+ self.check_invariant(cond)
+
+ def test_timeout(self):
+ cond = self.Condition()
+ wait = TimingWrapper(cond.wait)
+ cond.acquire()
+ res = wait(TIMEOUT1)
+ cond.release()
+ self.assertEqual(res, None)
+ self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1)
+
+
+class _TestEvent(BaseTestCase):
+
+ def _test_event(self, event):
+ time.sleep(TIMEOUT2)
+ event.set()
+
+ def test_event(self):
+ event = self.Event()
+ wait = TimingWrapper(event.wait)
+
+ # Removed temporaily, due to API shear, this does not
+ # work with threading._Event objects. is_set == isSet
+ #self.assertEqual(event.is_set(), False)
+
+ self.assertEqual(wait(0.0), None)
+ self.assertTimingAlmostEqual(wait.elapsed, 0.0)
+ self.assertEqual(wait(TIMEOUT1), None)
+ self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1)
+
+ event.set()
+
+ # See note above on the API differences
+ # self.assertEqual(event.is_set(), True)
+ self.assertEqual(wait(), None)
+ self.assertTimingAlmostEqual(wait.elapsed, 0.0)
+ self.assertEqual(wait(TIMEOUT1), None)
+ self.assertTimingAlmostEqual(wait.elapsed, 0.0)
+ # self.assertEqual(event.is_set(), True)
+
+ event.clear()
+
+ #self.assertEqual(event.is_set(), False)
+
+ self.Process(target=self._test_event, args=(event,)).start()
+ self.assertEqual(wait(), None)
+
+#
+#
+#
+
+class _TestValue(BaseTestCase):
+
+ codes_values = [
+ ('i', 4343, 24234),
+ ('d', 3.625, -4.25),
+ ('h', -232, 234),
+ ('c', latin('x'), latin('y'))
+ ]
+
+ def _test(self, values):
+ for sv, cv in zip(values, self.codes_values):
+ sv.value = cv[2]
+
+
+ def test_value(self, raw=False):
+ if self.TYPE != 'processes':
+ return
+
+ if raw:
+ values = [self.RawValue(code, value)
+ for code, value, _ in self.codes_values]
+ else:
+ values = [self.Value(code, value)
+ for code, value, _ in self.codes_values]
+
+ for sv, cv in zip(values, self.codes_values):
+ self.assertEqual(sv.value, cv[1])
+
+ proc = self.Process(target=self._test, args=(values,))
+ proc.start()
+ proc.join()
+
+ for sv, cv in zip(values, self.codes_values):
+ self.assertEqual(sv.value, cv[2])
+
+ def test_rawvalue(self):
+ self.test_value(raw=True)
+
+ def test_getobj_getlock(self):
+ if self.TYPE != 'processes':
+ return
+
+ val1 = self.Value('i', 5)
+ lock1 = val1.get_lock()
+ obj1 = val1.get_obj()
+
+ val2 = self.Value('i', 5, lock=None)
+ lock2 = val2.get_lock()
+ obj2 = val2.get_obj()
+
+ lock = self.Lock()
+ val3 = self.Value('i', 5, lock=lock)
+ lock3 = val3.get_lock()
+ obj3 = val3.get_obj()
+ self.assertEqual(lock, lock3)
+
+ arr4 = self.RawValue('i', 5)
+ self.assertFalse(hasattr(arr4, 'get_lock'))
+ self.assertFalse(hasattr(arr4, 'get_obj'))
+
+
+class _TestArray(BaseTestCase):
+
+ def f(self, seq):
+ for i in range(1, len(seq)):
+ seq[i] += seq[i-1]
+
+ def test_array(self, raw=False):
+ if self.TYPE != 'processes':
+ return
+
+ seq = [680, 626, 934, 821, 150, 233, 548, 982, 714, 831]
+ if raw:
+ arr = self.RawArray('i', seq)
+ else:
+ arr = self.Array('i', seq)
+
+ self.assertEqual(len(arr), len(seq))
+ self.assertEqual(arr[3], seq[3])
+ self.assertEqual(list(arr[2:7]), list(seq[2:7]))
+
+ arr[4:8] = seq[4:8] = array.array('i', [1, 2, 3, 4])
+
+ self.assertEqual(list(arr[:]), seq)
+
+ self.f(seq)
+
+ p = self.Process(target=self.f, args=(arr,))
+ p.start()
+ p.join()
+
+ self.assertEqual(list(arr[:]), seq)
+
+ def test_rawarray(self):
+ self.test_array(raw=True)
+
+ def test_getobj_getlock_obj(self):
+ if self.TYPE != 'processes':
+ return
+
+ arr1 = self.Array('i', range(10))
+ lock1 = arr1.get_lock()
+ obj1 = arr1.get_obj()
+
+ arr2 = self.Array('i', range(10), lock=None)
+ lock2 = arr2.get_lock()
+ obj2 = arr2.get_obj()
+
+ lock = self.Lock()
+ arr3 = self.Array('i', range(10), lock=lock)
+ lock3 = arr3.get_lock()
+ obj3 = arr3.get_obj()
+ self.assertEqual(lock, lock3)
+
+ arr4 = self.RawArray('i', range(10))
+ self.assertFalse(hasattr(arr4, 'get_lock'))
+ self.assertFalse(hasattr(arr4, 'get_obj'))
+
+#
+#
+#
+
+class _TestContainers(BaseTestCase):
+
+ ALLOWED_TYPES = ('manager',)
+
+ def test_list(self):
+ a = self.list(range(10))
+ self.assertEqual(a[:], range(10))
+
+ b = self.list()
+ self.assertEqual(b[:], [])
+
+ b.extend(range(5))
+ self.assertEqual(b[:], range(5))
+
+ self.assertEqual(b[2], 2)
+ self.assertEqual(b[2:10], [2,3,4])
+
+ b *= 2
+ self.assertEqual(b[:], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4])
+
+ self.assertEqual(b + [5, 6], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4, 5, 6])
+
+ self.assertEqual(a[:], range(10))
+
+ d = [a, b]
+ e = self.list(d)
+ self.assertEqual(
+ e[:],
+ [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4]]
+ )
+
+ f = self.list([a])
+ a.append('hello')
+ self.assertEqual(f[:], [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 'hello']])
+
+ def test_dict(self):
+ d = self.dict()
+ indices = range(65, 70)
+ for i in indices:
+ d[i] = chr(i)
+ self.assertEqual(d.copy(), dict((i, chr(i)) for i in indices))
+ self.assertEqual(sorted(d.keys()), indices)
+ self.assertEqual(sorted(d.values()), [chr(i) for i in indices])
+ self.assertEqual(sorted(d.items()), [(i, chr(i)) for i in indices])
+
+ def test_namespace(self):
+ n = self.Namespace()
+ n.name = 'Bob'
+ n.job = 'Builder'
+ n._hidden = 'hidden'
+ self.assertEqual((n.name, n.job), ('Bob', 'Builder'))
+ del n.job
+ self.assertEqual(str(n), "Namespace(name='Bob')")
+ self.assertTrue(hasattr(n, 'name'))
+ self.assertTrue(not hasattr(n, 'job'))
+
+#
+#
+#
+
+def sqr(x, wait=0.0):
+ time.sleep(wait)
+ return x*x
+
+class _TestPool(BaseTestCase):
+
+ def test_apply(self):
+ papply = self.pool.apply
+ self.assertEqual(papply(sqr, (5,)), sqr(5))
+ self.assertEqual(papply(sqr, (), {'x':3}), sqr(x=3))
+
+ def test_map(self):
+ pmap = self.pool.map
+ self.assertEqual(pmap(sqr, range(10)), map(sqr, range(10)))
+ self.assertEqual(pmap(sqr, range(100), chunksize=20),
+ map(sqr, range(100)))
+
+ def test_async(self):
+ res = self.pool.apply_async(sqr, (7, TIMEOUT1,))
+ get = TimingWrapper(res.get)
+ self.assertEqual(get(), 49)
+ self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1)
+
+ def test_async_timeout(self):
+ res = self.pool.apply_async(sqr, (6, TIMEOUT2 + 0.2))
+ get = TimingWrapper(res.get)
+ self.assertRaises(multiprocessing.TimeoutError, get, timeout=TIMEOUT2)
+ self.assertTimingAlmostEqual(get.elapsed, TIMEOUT2)
+
+ def test_imap(self):
+ it = self.pool.imap(sqr, range(10))
+ self.assertEqual(list(it), map(sqr, range(10)))
+
+ it = self.pool.imap(sqr, range(10))
+ for i in range(10):
+ self.assertEqual(it.next(), i*i)
+ self.assertRaises(StopIteration, it.next)
+
+ it = self.pool.imap(sqr, range(1000), chunksize=100)
+ for i in range(1000):
+ self.assertEqual(it.next(), i*i)
+ self.assertRaises(StopIteration, it.next)
+
+ def test_imap_unordered(self):
+ it = self.pool.imap_unordered(sqr, range(1000))
+ self.assertEqual(sorted(it), map(sqr, range(1000)))
+
+ it = self.pool.imap_unordered(sqr, range(1000), chunksize=53)
+ self.assertEqual(sorted(it), map(sqr, range(1000)))
+
+ def test_make_pool(self):
+ p = multiprocessing.Pool(3)
+ self.assertEqual(3, len(p._pool))
+ p.close()
+ p.join()
+
+ def test_terminate(self):
+ if self.TYPE == 'manager':
+ # On Unix a forked process increfs each shared object to
+ # which its parent process held a reference. If the
+ # forked process gets terminated then there is likely to
+ # be a reference leak. So to prevent
+ # _TestZZZNumberOfObjects from failing we skip this test
+ # when using a manager.
+ return
+
+ result = self.pool.map_async(
+ time.sleep, [0.1 for i in range(10000)], chunksize=1
+ )
+ self.pool.terminate()
+ join = TimingWrapper(self.pool.join)
+ join()
+ self.assertTrue(join.elapsed < 0.2)
+
+#
+# Test that manager has expected number of shared objects left
+#
+
+class _TestZZZNumberOfObjects(BaseTestCase):
+ # Because test cases are sorted alphabetically, this one will get
+ # run after all the other tests for the manager. It tests that
+ # there have been no "reference leaks" for the manager's shared
+ # objects. Note the comment in _TestPool.test_terminate().
+ ALLOWED_TYPES = ('manager',)
+
+ def test_number_of_objects(self):
+ EXPECTED_NUMBER = 1 # the pool object is still alive
+ multiprocessing.active_children() # discard dead process objs
+ gc.collect() # do garbage collection
+ refs = self.manager._number_of_objects()
+ if refs != EXPECTED_NUMBER:
+ print self.manager._debugInfo()
+
+ self.assertEqual(refs, EXPECTED_NUMBER)
+
+#
+# Test of creating a customized manager class
+#
+
+from multiprocessing.managers import BaseManager, BaseProxy, RemoteError
+
+class FooBar(object):
+ def f(self):
+ return 'f()'
+ def g(self):
+ raise ValueError
+ def _h(self):
+ return '_h()'
+
+def baz():
+ for i in xrange(10):
+ yield i*i
+
+class IteratorProxy(BaseProxy):
+ _exposed_ = ('next', '__next__')
+ def __iter__(self):
+ return self
+ def next(self):
+ return self._callmethod('next')
+ def __next__(self):
+ return self._callmethod('__next__')
+
+class MyManager(BaseManager):
+ pass
+
+MyManager.register('Foo', callable=FooBar)
+MyManager.register('Bar', callable=FooBar, exposed=('f', '_h'))
+MyManager.register('baz', callable=baz, proxytype=IteratorProxy)
+
+
+class _TestMyManager(BaseTestCase):
+
+ ALLOWED_TYPES = ('manager',)
+
+ def test_mymanager(self):
+ manager = MyManager()
+ manager.start()
+
+ foo = manager.Foo()
+ bar = manager.Bar()
+ baz = manager.baz()
+
+ foo_methods = [name for name in ('f', 'g', '_h') if hasattr(foo, name)]
+ bar_methods = [name for name in ('f', 'g', '_h') if hasattr(bar, name)]
+
+ self.assertEqual(foo_methods, ['f', 'g'])
+ self.assertEqual(bar_methods, ['f', '_h'])
+
+ self.assertEqual(foo.f(), 'f()')
+ self.assertRaises(ValueError, foo.g)
+ self.assertEqual(foo._callmethod('f'), 'f()')
+ self.assertRaises(RemoteError, foo._callmethod, '_h')
+
+ self.assertEqual(bar.f(), 'f()')
+ self.assertEqual(bar._h(), '_h()')
+ self.assertEqual(bar._callmethod('f'), 'f()')
+ self.assertEqual(bar._callmethod('_h'), '_h()')
+
+ self.assertEqual(list(baz), [i*i for i in range(10)])
+
+ manager.shutdown()
+
+#
+# Test of connecting to a remote server and using xmlrpclib for serialization
+#
+
+_queue = Queue.Queue()
+def get_queue():
+ return _queue
+
+class QueueManager(BaseManager):
+ '''manager class used by server process'''
+QueueManager.register('get_queue', callable=get_queue)
+
+class QueueManager2(BaseManager):
+ '''manager class which specifies the same interface as QueueManager'''
+QueueManager2.register('get_queue')
+
+
+SERIALIZER = 'xmlrpclib'
+
+class _TestRemoteManager(BaseTestCase):
+
+ ALLOWED_TYPES = ('manager',)
+
+ def _putter(self, address, authkey):
+ manager = QueueManager2(
+ address=address, authkey=authkey, serializer=SERIALIZER
+ )
+ manager.connect()
+ queue = manager.get_queue()
+ queue.put(('hello world', None, True, 2.25))
+
+ def test_remote(self):
+ authkey = os.urandom(32)
+
+ manager = QueueManager(
+ address=('localhost', 0), authkey=authkey, serializer=SERIALIZER
+ )
+ manager.start()
+
+ p = self.Process(target=self._putter, args=(manager.address, authkey))
+ p.start()
+
+ manager2 = QueueManager2(
+ address=manager.address, authkey=authkey, serializer=SERIALIZER
+ )
+ manager2.connect()
+ queue = manager2.get_queue()
+
+ # Note that xmlrpclib will deserialize object as a list not a tuple
+ self.assertEqual(queue.get(), ['hello world', None, True, 2.25])
+
+ # Because we are using xmlrpclib for serialization instead of
+ # pickle this will cause a serialization error.
+ self.assertRaises(Exception, queue.put, time.sleep)
+
+ # Make queue finalizer run before the server is stopped
+ del queue
+ manager.shutdown()
+
+#
+#
+#
+
+SENTINEL = latin('')
+
+class _TestConnection(BaseTestCase):
+
+ ALLOWED_TYPES = ('processes', 'threads')
+
+ def _echo(self, conn):
+ for msg in iter(conn.recv_bytes, SENTINEL):
+ conn.send_bytes(msg)
+ conn.close()
+
+ def test_connection(self):
+ conn, child_conn = self.Pipe()
+
+ p = self.Process(target=self._echo, args=(child_conn,))
+ p.set_daemon(True)
+ p.start()
+
+ seq = [1, 2.25, None]
+ msg = latin('hello world')
+ longmsg = msg * 10
+ arr = array.array('i', range(4))
+
+ if self.TYPE == 'processes':
+ self.assertEqual(type(conn.fileno()), int)
+
+ self.assertEqual(conn.send(seq), None)
+ self.assertEqual(conn.recv(), seq)
+
+ self.assertEqual(conn.send_bytes(msg), None)
+ self.assertEqual(conn.recv_bytes(), msg)
+
+ if self.TYPE == 'processes':
+ buffer = array.array('i', [0]*10)
+ expected = list(arr) + [0] * (10 - len(arr))
+ self.assertEqual(conn.send_bytes(arr), None)
+ self.assertEqual(conn.recv_bytes_into(buffer),
+ len(arr) * buffer.itemsize)
+ self.assertEqual(list(buffer), expected)
+
+ buffer = array.array('i', [0]*10)
+ expected = [0] * 3 + list(arr) + [0] * (10 - 3 - len(arr))
+ self.assertEqual(conn.send_bytes(arr), None)
+ self.assertEqual(conn.recv_bytes_into(buffer, 3 * buffer.itemsize),
+ len(arr) * buffer.itemsize)
+ self.assertEqual(list(buffer), expected)
+
+ buffer = bytearray(latin(' ' * 40))
+ self.assertEqual(conn.send_bytes(longmsg), None)
+ try:
+ res = conn.recv_bytes_into(buffer)
+ except multiprocessing.BufferTooShort, e:
+ self.assertEqual(e.args, (longmsg,))
+ else:
+ self.fail('expected BufferTooShort, got %s' % res)
+
+ poll = TimingWrapper(conn.poll)
+
+ self.assertEqual(poll(), False)
+ self.assertTimingAlmostEqual(poll.elapsed, 0)
+
+ self.assertEqual(poll(TIMEOUT1), False)
+ self.assertTimingAlmostEqual(poll.elapsed, TIMEOUT1)
+
+ conn.send(None)
+
+ self.assertEqual(poll(TIMEOUT1), True)
+ self.assertTimingAlmostEqual(poll.elapsed, 0)
+
+ self.assertEqual(conn.recv(), None)
+
+ really_big_msg = latin('X') * (1024 * 1024 * 16) # 16Mb
+ conn.send_bytes(really_big_msg)
+ self.assertEqual(conn.recv_bytes(), really_big_msg)
+
+ conn.send_bytes(SENTINEL) # tell child to quit
+ child_conn.close()
+
+ if self.TYPE == 'processes':
+ self.assertEqual(conn.readable, True)
+ self.assertEqual(conn.writable, True)
+ self.assertRaises(EOFError, conn.recv)
+ self.assertRaises(EOFError, conn.recv_bytes)
+
+ p.join()
+
+ def test_duplex_false(self):
+ reader, writer = self.Pipe(duplex=False)
+ self.assertEqual(writer.send(1), None)
+ self.assertEqual(reader.recv(), 1)
+ if self.TYPE == 'processes':
+ self.assertEqual(reader.readable, True)
+ self.assertEqual(reader.writable, False)
+ self.assertEqual(writer.readable, False)
+ self.assertEqual(writer.writable, True)
+ self.assertRaises(IOError, reader.send, 2)
+ self.assertRaises(IOError, writer.recv)
+ self.assertRaises(IOError, writer.poll)
+
+ def test_spawn_close(self):
+ # We test that a pipe connection can be closed by parent
+ # process immediately after child is spawned. On Windows this
+ # would have sometimes failed on old versions because
+ # child_conn would be closed before the child got a chance to
+ # duplicate it.
+ conn, child_conn = self.Pipe()
+
+ p = self.Process(target=self._echo, args=(child_conn,))
+ p.start()
+ child_conn.close() # this might complete before child initializes
+
+ msg = latin('hello')
+ conn.send_bytes(msg)
+ self.assertEqual(conn.recv_bytes(), msg)
+
+ conn.send_bytes(SENTINEL)
+ conn.close()
+ p.join()
+
+ def test_sendbytes(self):
+ if self.TYPE != 'processes':
+ return
+
+ msg = latin('abcdefghijklmnopqrstuvwxyz')
+ a, b = self.Pipe()
+
+ a.send_bytes(msg)
+ self.assertEqual(b.recv_bytes(), msg)
+
+ a.send_bytes(msg, 5)
+ self.assertEqual(b.recv_bytes(), msg[5:])
+
+ a.send_bytes(msg, 7, 8)
+ self.assertEqual(b.recv_bytes(), msg[7:7+8])
+
+ a.send_bytes(msg, 26)
+ self.assertEqual(b.recv_bytes(), latin(''))
+
+ a.send_bytes(msg, 26, 0)
+ self.assertEqual(b.recv_bytes(), latin(''))
+
+ self.assertRaises(ValueError, a.send_bytes, msg, 27)
+
+ self.assertRaises(ValueError, a.send_bytes, msg, 22, 5)
+
+ self.assertRaises(ValueError, a.send_bytes, msg, 26, 1)
+
+ self.assertRaises(ValueError, a.send_bytes, msg, -1)
+
+ self.assertRaises(ValueError, a.send_bytes, msg, 4, -1)
+
+
+class _TestListenerClient(BaseTestCase):
+
+ ALLOWED_TYPES = ('processes', 'threads')
+
+ def _test(self, address):
+ conn = self.connection.Client(address)
+ conn.send('hello')
+ conn.close()
+
+ def test_listener_client(self):
+ for family in self.connection.families:
+ l = self.connection.Listener(family=family)
+ p = self.Process(target=self._test, args=(l.address,))
+ p.set_daemon(True)
+ p.start()
+ conn = l.accept()
+ self.assertEqual(conn.recv(), 'hello')
+ p.join()
+ l.close()
+
+#
+# Test of sending connection and socket objects between processes
+#
+
+class _TestPicklingConnections(BaseTestCase):
+
+ ALLOWED_TYPES = ('processes',)
+
+ def _listener(self, conn, families):
+ for fam in families:
+ l = self.connection.Listener(family=fam)
+ conn.send(l.address)
+ new_conn = l.accept()
+ conn.send(new_conn)
+
+ if self.TYPE == 'processes':
+ l = socket.socket()
+ l.bind(('localhost', 0))
+ conn.send(l.getsockname())
+ l.listen(1)
+ new_conn, addr = l.accept()
+ conn.send(new_conn)
+
+ conn.recv()
+
+ def _remote(self, conn):
+ for (address, msg) in iter(conn.recv, None):
+ client = self.connection.Client(address)
+ client.send(msg.upper())
+ client.close()
+
+ if self.TYPE == 'processes':
+ address, msg = conn.recv()
+ client = socket.socket()
+ client.connect(address)
+ client.sendall(msg.upper())
+ client.close()
+
+ conn.close()
+
+ def test_pickling(self):
+ try:
+ multiprocessing.allow_connection_pickling()
+ except ImportError:
+ return
+
+ families = self.connection.families
+
+ lconn, lconn0 = self.Pipe()
+ lp = self.Process(target=self._listener, args=(lconn0, families))
+ lp.start()
+ lconn0.close()
+
+ rconn, rconn0 = self.Pipe()
+ rp = self.Process(target=self._remote, args=(rconn0,))
+ rp.start()
+ rconn0.close()
+
+ for fam in families:
+ msg = ('This connection uses family %s' % fam).encode('ascii')
+ address = lconn.recv()
+ rconn.send((address, msg))
+ new_conn = lconn.recv()
+ self.assertEqual(new_conn.recv(), msg.upper())
+
+ rconn.send(None)
+
+ if self.TYPE == 'processes':
+ msg = latin('This connection uses a normal socket')
+ address = lconn.recv()
+ rconn.send((address, msg))
+ if hasattr(socket, 'fromfd'):
+ new_conn = lconn.recv()
+ self.assertEqual(new_conn.recv(100), msg.upper())
+ else:
+ # XXX On Windows with Py2.6 need to backport fromfd()
+ discard = lconn.recv_bytes()
+
+ lconn.send(None)
+
+ rconn.close()
+ lconn.close()
+
+ lp.join()
+ rp.join()
+
+#
+#
+#
+
+class _TestHeap(BaseTestCase):
+
+ ALLOWED_TYPES = ('processes',)
+
+ def test_heap(self):
+ iterations = 5000
+ maxblocks = 50
+ blocks = []
+
+ # create and destroy lots of blocks of different sizes
+ for i in xrange(iterations):
+ size = int(random.lognormvariate(0, 1) * 1000)
+ b = multiprocessing.heap.BufferWrapper(size)
+ blocks.append(b)
+ if len(blocks) > maxblocks:
+ i = random.randrange(maxblocks)
+ del blocks[i]
+
+ # get the heap object
+ heap = multiprocessing.heap.BufferWrapper._heap
+
+ # verify the state of the heap
+ all = []
+ occupied = 0
+ for L in heap._len_to_seq.values():
+ for arena, start, stop in L:
+ all.append((heap._arenas.index(arena), start, stop,
+ stop-start, 'free'))
+ for arena, start, stop in heap._allocated_blocks:
+ all.append((heap._arenas.index(arena), start, stop,
+ stop-start, 'occupied'))
+ occupied += (stop-start)
+
+ all.sort()
+
+ for i in range(len(all)-1):
+ (arena, start, stop) = all[i][:3]
+ (narena, nstart, nstop) = all[i+1][:3]
+ self.assertTrue((arena != narena and nstart == 0) or
+ (stop == nstart))
+
+#
+#
+#
+
+try:
+ from ctypes import Structure, Value, copy, c_int, c_double
+except ImportError:
+ Structure = object
+ c_int = c_double = None
+
+class _Foo(Structure):
+ _fields_ = [
+ ('x', c_int),
+ ('y', c_double)
+ ]
+
+class _TestSharedCTypes(BaseTestCase):
+
+ ALLOWED_TYPES = ('processes',)
+
+ def _double(self, x, y, foo, arr, string):
+ x.value *= 2
+ y.value *= 2
+ foo.x *= 2
+ foo.y *= 2
+ string.value *= 2
+ for i in range(len(arr)):
+ arr[i] *= 2
+
+ def test_sharedctypes(self, lock=False):
+ if c_int is None:
+ return
+
+ x = Value('i', 7, lock=lock)
+ y = Value(ctypes.c_double, 1.0/3.0, lock=lock)
+ foo = Value(_Foo, 3, 2, lock=lock)
+ arr = Array('d', range(10), lock=lock)
+ string = Array('c', 20, lock=lock)
+ string.value = 'hello'
+
+ p = self.Process(target=self._double, args=(x, y, foo, arr, string))
+ p.start()
+ p.join()
+
+ self.assertEqual(x.value, 14)
+ self.assertAlmostEqual(y.value, 2.0/3.0)
+ self.assertEqual(foo.x, 6)
+ self.assertAlmostEqual(foo.y, 4.0)
+ for i in range(10):
+ self.assertAlmostEqual(arr[i], i*2)
+ self.assertEqual(string.value, latin('hellohello'))
+
+ def test_synchronize(self):
+ self.test_sharedctypes(lock=True)
+
+ def test_copy(self):
+ if c_int is None:
+ return
+
+ foo = _Foo(2, 5.0)
+ bar = copy(foo)
+ foo.x = 0
+ foo.y = 0
+ self.assertEqual(bar.x, 2)
+ self.assertAlmostEqual(bar.y, 5.0)
+
+#
+#
+#
+
+class _TestFinalize(BaseTestCase):
+
+ ALLOWED_TYPES = ('processes',)
+
+ def _test_finalize(self, conn):
+ class Foo(object):
+ pass
+
+ a = Foo()
+ util.Finalize(a, conn.send, args=('a',))
+ del a # triggers callback for a
+
+ b = Foo()
+ close_b = util.Finalize(b, conn.send, args=('b',))
+ close_b() # triggers callback for b
+ close_b() # does nothing because callback has already been called
+ del b # does nothing because callback has already been called
+
+ c = Foo()
+ util.Finalize(c, conn.send, args=('c',))
+
+ d10 = Foo()
+ util.Finalize(d10, conn.send, args=('d10',), exitpriority=1)
+
+ d01 = Foo()
+ util.Finalize(d01, conn.send, args=('d01',), exitpriority=0)
+ d02 = Foo()
+ util.Finalize(d02, conn.send, args=('d02',), exitpriority=0)
+ d03 = Foo()
+ util.Finalize(d03, conn.send, args=('d03',), exitpriority=0)
+
+ util.Finalize(None, conn.send, args=('e',), exitpriority=-10)
+
+ util.Finalize(None, conn.send, args=('STOP',), exitpriority=-100)
+
+ # call mutliprocessing's cleanup function then exit process without
+ # garbage collecting locals
+ util._exit_function()
+ conn.close()
+ os._exit(0)
+
+ def test_finalize(self):
+ conn, child_conn = self.Pipe()
+
+ p = self.Process(target=self._test_finalize, args=(child_conn,))
+ p.start()
+ p.join()
+
+ result = [obj for obj in iter(conn.recv, 'STOP')]
+ self.assertEqual(result, ['a', 'b', 'd10', 'd03', 'd02', 'd01', 'e'])
+
+#
+# Test that from ... import * works for each module
+#
+
+class _TestImportStar(BaseTestCase):
+
+ ALLOWED_TYPES = ('processes',)
+
+ def test_import(self):
+ modules = (
+ 'multiprocessing', 'multiprocessing.connection',
+ 'multiprocessing.heap', 'multiprocessing.managers',
+ 'multiprocessing.pool', 'multiprocessing.process',
+ 'multiprocessing.reduction', 'multiprocessing.sharedctypes',
+ 'multiprocessing.synchronize', 'multiprocessing.util'
+ )
+
+ for name in modules:
+ __import__(name)
+ mod = sys.modules[name]
+
+ for attr in getattr(mod, '__all__', ()):
+ self.assertTrue(
+ hasattr(mod, attr),
+ '%r does not have attribute %r' % (mod, attr)
+ )
+
+#
+# Quick test that logging works -- does not test logging output
+#
+
+class _TestLogging(BaseTestCase):
+
+ ALLOWED_TYPES = ('processes',)
+
+ def test_enable_logging(self):
+ logger = multiprocessing.get_logger()
+ logger.setLevel(util.SUBWARNING)
+ self.assertTrue(logger is not None)
+ logger.debug('this will not be printed')
+ logger.info('nor will this')
+ logger.setLevel(LOG_LEVEL)
+
+ def _test_level(self, conn):
+ logger = multiprocessing.get_logger()
+ conn.send(logger.getEffectiveLevel())
+
+ def test_level(self):
+ LEVEL1 = 32
+ LEVEL2 = 37
+
+ logger = multiprocessing.get_logger()
+ root_logger = logging.getLogger()
+ root_level = root_logger.level
+
+ reader, writer = multiprocessing.Pipe(duplex=False)
+
+ logger.setLevel(LEVEL1)
+ self.Process(target=self._test_level, args=(writer,)).start()
+ self.assertEqual(LEVEL1, reader.recv())
+
+ logger.setLevel(logging.NOTSET)
+ root_logger.setLevel(LEVEL2)
+ self.Process(target=self._test_level, args=(writer,)).start()
+ self.assertEqual(LEVEL2, reader.recv())
+
+ root_logger.setLevel(root_level)
+ logger.setLevel(level=LOG_LEVEL)
+
+#
+# Functions used to create test cases from the base ones in this module
+#
+
+def get_attributes(Source, names):
+ d = {}
+ for name in names:
+ obj = getattr(Source, name)
+ if type(obj) == type(get_attributes):
+ obj = staticmethod(obj)
+ d[name] = obj
+ return d
+
+def create_test_cases(Mixin, type):
+ result = {}
+ glob = globals()
+ Type = type[0].upper() + type[1:]
+
+ for name in glob.keys():
+ if name.startswith('_Test'):
+ base = glob[name]
+ if type in base.ALLOWED_TYPES:
+ newname = 'With' + Type + name[1:]
+ class Temp(base, unittest.TestCase, Mixin):
+ pass
+ result[newname] = Temp
+ Temp.__name__ = newname
+ Temp.__module__ = Mixin.__module__
+ return result
+
+#
+# Create test cases
+#
+
+class ProcessesMixin(object):
+ TYPE = 'processes'
+ Process = multiprocessing.Process
+ locals().update(get_attributes(multiprocessing, (
+ 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore',
+ 'Condition', 'Event', 'Value', 'Array', 'RawValue',
+ 'RawArray', 'current_process', 'active_children', 'Pipe',
+ 'connection', 'JoinableQueue'
+ )))
+
+testcases_processes = create_test_cases(ProcessesMixin, type='processes')
+globals().update(testcases_processes)
+
+
+class ManagerMixin(object):
+ TYPE = 'manager'
+ Process = multiprocessing.Process
+ manager = object.__new__(multiprocessing.managers.SyncManager)
+ locals().update(get_attributes(manager, (
+ 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore',
+ 'Condition', 'Event', 'Value', 'Array', 'list', 'dict',
+ 'Namespace', 'JoinableQueue'
+ )))
+
+testcases_manager = create_test_cases(ManagerMixin, type='manager')
+globals().update(testcases_manager)
+
+
+class ThreadsMixin(object):
+ TYPE = 'threads'
+ Process = multiprocessing.dummy.Process
+ locals().update(get_attributes(multiprocessing.dummy, (
+ 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore',
+ 'Condition', 'Event', 'Value', 'Array', 'current_process',
+ 'active_children', 'Pipe', 'connection', 'dict', 'list',
+ 'Namespace', 'JoinableQueue'
+ )))
+
+testcases_threads = create_test_cases(ThreadsMixin, type='threads')
+globals().update(testcases_threads)
+
+#
+#
+#
+
+def test_main(run=None):
+ if run is None:
+ from test.test_support import run_unittest as run
+
+ util.get_temp_dir() # creates temp directory for use by all processes
+
+ multiprocessing.get_logger().setLevel(LOG_LEVEL)
+
+ ProcessesMixin.pool = multiprocessing.Pool(4)
+ ThreadsMixin.pool = multiprocessing.dummy.Pool(4)
+ ManagerMixin.manager.__init__()
+ ManagerMixin.manager.start()
+ ManagerMixin.pool = ManagerMixin.manager.Pool(4)
+
+ testcases = (
+ sorted(testcases_processes.values(), key=lambda tc:tc.__name__) +
+ sorted(testcases_threads.values(), key=lambda tc:tc.__name__) +
+ sorted(testcases_manager.values(), key=lambda tc:tc.__name__)
+ )
+
+ loadTestsFromTestCase = unittest.defaultTestLoader.loadTestsFromTestCase
+ suite = unittest.TestSuite(loadTestsFromTestCase(tc) for tc in testcases)
+ run(suite)
+
+ ThreadsMixin.pool.terminate()
+ ProcessesMixin.pool.terminate()
+ ManagerMixin.pool.terminate()
+ ManagerMixin.manager.shutdown()
+
+ del ProcessesMixin.pool, ThreadsMixin.pool, ManagerMixin.pool
+
+def main():
+ test_main(unittest.TextTestRunner(verbosity=2).run)
+
+if __name__ == '__main__':
+ main()
Modified: python/trunk/Modules/_multiprocessing/multiprocessing.c
==============================================================================
--- python/trunk/Modules/_multiprocessing/multiprocessing.c	(original)
+++ python/trunk/Modules/_multiprocessing/multiprocessing.c	Fri Jun 13 21:13:39 2008
@@ -1,311 +1,311 @@
-/*
- * Extension module used by multiprocessing package
- *
- * multiprocessing.c
- *
- * Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
- */
-
-#include "multiprocessing.h"
-
-PyObject *create_win32_namespace(void);
-
-PyObject *pickle_dumps, *pickle_loads, *pickle_protocol;
-PyObject *ProcessError, *BufferTooShort;
-
-/*
- * Function which raises exceptions based on error codes
- */
-
-PyObject *
-mp_SetError(PyObject *Type, int num)
-{
-	switch (num) {
-#ifdef MS_WINDOWS
-	case MP_STANDARD_ERROR: 
-		if (Type == NULL)
-			Type = PyExc_WindowsError;
-		PyErr_SetExcFromWindowsErr(Type, 0);
-		break;
-	case MP_SOCKET_ERROR:
-		if (Type == NULL)
-			Type = PyExc_WindowsError;
-		PyErr_SetExcFromWindowsErr(Type, WSAGetLastError());
-		break;
-#else /* !MS_WINDOWS */
-	case MP_STANDARD_ERROR:
-	case MP_SOCKET_ERROR:
-		if (Type == NULL)
-			Type = PyExc_OSError;
-		PyErr_SetFromErrno(Type);
-		break;
-#endif /* !MS_WINDOWS */
-	case MP_MEMORY_ERROR:
-		PyErr_NoMemory();
-		break;
-	case MP_END_OF_FILE:
-		PyErr_SetNone(PyExc_EOFError);
-		break;
-	case MP_EARLY_END_OF_FILE:
-		PyErr_SetString(PyExc_IOError,
-				"got end of file during message");
-		break;
-	case MP_BAD_MESSAGE_LENGTH:
-		PyErr_SetString(PyExc_IOError, "bad message length");
-		break;
-	case MP_EXCEPTION_HAS_BEEN_SET:
-		break;
-	default:
-		PyErr_Format(PyExc_RuntimeError,
-			 "unkown error number %d", num);
-	}
-	return NULL;
-}
-
-
-/*
- * Windows only
- */
-
-#ifdef MS_WINDOWS
-
-/* On Windows we set an event to signal Ctrl-C; compare with timemodule.c */
-
-HANDLE sigint_event = NULL;
-
-static BOOL WINAPI
-ProcessingCtrlHandler(DWORD dwCtrlType)
-{
-	SetEvent(sigint_event);
-	return FALSE;
-}
-
-/*
- * Unix only
- */
-
-#else /* !MS_WINDOWS */
-
-#if HAVE_FD_TRANSFER
-
-/* Functions for transferring file descriptors between processes.
- Reimplements some of the functionality of the fdcred
- module at http://www.mca-ltd.com/resources/fdcred_1.tgz. */
-
-static PyObject *
-multiprocessing_sendfd(PyObject *self, PyObject *args)
-{
-	int conn, fd, res;
-	char dummy_char;
-	char buf[CMSG_SPACE(sizeof(int))];
-	struct msghdr msg = {0};
-	struct iovec dummy_iov;
-	struct cmsghdr *cmsg;
-
-	if (!PyArg_ParseTuple(args, "ii", &conn, &fd))
-		return NULL;
-
-	dummy_iov.iov_base = &dummy_char;
-	dummy_iov.iov_len = 1;
-	msg.msg_control = buf;
-	msg.msg_controllen = sizeof(buf);
-	msg.msg_iov = &dummy_iov;
-	msg.msg_iovlen = 1;
-	cmsg = CMSG_FIRSTHDR(&msg);
-	cmsg->cmsg_level = SOL_SOCKET;
-	cmsg->cmsg_type = SCM_RIGHTS;
-	cmsg->cmsg_len = CMSG_LEN(sizeof(int));
-	msg.msg_controllen = cmsg->cmsg_len;
-	*(int*)CMSG_DATA(cmsg) = fd;
-
-	Py_BEGIN_ALLOW_THREADS
-	res = sendmsg(conn, &msg, 0);
-	Py_END_ALLOW_THREADS
-
-	if (res < 0)
-		return PyErr_SetFromErrno(PyExc_OSError);
-	Py_RETURN_NONE;
-}
-
-static PyObject *
-multiprocessing_recvfd(PyObject *self, PyObject *args)
-{
-	int conn, fd, res;
-	char dummy_char;
-	char buf[CMSG_SPACE(sizeof(int))];
-	struct msghdr msg = {0};
-	struct iovec dummy_iov;
-	struct cmsghdr *cmsg;
-
-	if (!PyArg_ParseTuple(args, "i", &conn))
-		return NULL;
-
-	dummy_iov.iov_base = &dummy_char;
-	dummy_iov.iov_len = 1;
-	msg.msg_control = buf;
-	msg.msg_controllen = sizeof(buf);
-	msg.msg_iov = &dummy_iov;
-	msg.msg_iovlen = 1;
-	cmsg = CMSG_FIRSTHDR(&msg);
-	cmsg->cmsg_level = SOL_SOCKET;
-	cmsg->cmsg_type = SCM_RIGHTS;
-	cmsg->cmsg_len = CMSG_LEN(sizeof(int));
-	msg.msg_controllen = cmsg->cmsg_len;
-
-	Py_BEGIN_ALLOW_THREADS
-	res = recvmsg(conn, &msg, 0);
-	Py_END_ALLOW_THREADS
-
-	if (res < 0)
-		return PyErr_SetFromErrno(PyExc_OSError);
-
-	fd = *(int*)CMSG_DATA(cmsg);
-	return Py_BuildValue("i", fd);
-}
-
-#endif /* HAVE_FD_TRANSFER */
-
-#endif /* !MS_WINDOWS */
-
-
-/*
- * All platforms
- */
-
-static PyObject*
-multiprocessing_address_of_buffer(PyObject *self, PyObject *obj)
-{
-	void *buffer;
-	Py_ssize_t buffer_len;
-
-	if (PyObject_AsWriteBuffer(obj, &buffer, &buffer_len) < 0)
-		return NULL;
-
-	return Py_BuildValue("N" F_PY_SSIZE_T, 
-			 PyLong_FromVoidPtr(buffer), buffer_len);
-}
-
-
-/*
- * Function table
- */
-
-static PyMethodDef module_methods[] = {
-	{"address_of_buffer", multiprocessing_address_of_buffer, METH_O, 
-	 "address_of_buffer(obj) -> int\n" 
-	 "Return address of obj assuming obj supports buffer inteface"},
-#if HAVE_FD_TRANSFER
-	{"sendfd", multiprocessing_sendfd, METH_VARARGS, 
-	 "sendfd(sockfd, fd) -> None\n"
-	 "Send file descriptor given by fd over the unix domain socket\n"
-	 "whose file decriptor is sockfd"},
-	{"recvfd", multiprocessing_recvfd, METH_VARARGS,
-	 "recvfd(sockfd) -> fd\n"
-	 "Receive a file descriptor over a unix domain socket\n"
-	 "whose file decriptor is sockfd"},
-#endif
-	{NULL}
-};
-
-
-/*
- * Initialize
- */
-
-PyMODINIT_FUNC
-init_multiprocessing(void)
-{
-	PyObject *module, *temp, *value;
-
-	/* Initialize module */
-	module = Py_InitModule("_multiprocessing", module_methods);
-	if (!module)
-		return;
-
-	/* Get copy of objects from pickle */
-	temp = PyImport_ImportModule(PICKLE_MODULE);
-	if (!temp)
-		return;
-	pickle_dumps = PyObject_GetAttrString(temp, "dumps");
-	pickle_loads = PyObject_GetAttrString(temp, "loads");
-	pickle_protocol = PyObject_GetAttrString(temp, "HIGHEST_PROTOCOL");
-	Py_XDECREF(temp);
-
-	/* Get copy of BufferTooShort */
-	temp = PyImport_ImportModule("multiprocessing");
-	if (!temp)
-		return;
-	BufferTooShort = PyObject_GetAttrString(temp, "BufferTooShort");
-	Py_XDECREF(temp);
-
-	/* Add connection type to module */
-	if (PyType_Ready(&ConnectionType) < 0)
-		return;
-	Py_INCREF(&ConnectionType);	
-	PyModule_AddObject(module, "Connection", (PyObject*)&ConnectionType);
-
-#if defined(MS_WINDOWS) || HAVE_SEM_OPEN
-	/* Add SemLock type to module */
-	if (PyType_Ready(&SemLockType) < 0)
-		return;
-	Py_INCREF(&SemLockType);
-	PyDict_SetItemString(SemLockType.tp_dict, "SEM_VALUE_MAX", 
-			 Py_BuildValue("i", SEM_VALUE_MAX));
-	PyModule_AddObject(module, "SemLock", (PyObject*)&SemLockType); 
-#endif
-
-#ifdef MS_WINDOWS
-	/* Add PipeConnection to module */
-	if (PyType_Ready(&PipeConnectionType) < 0)
-		return;
-	Py_INCREF(&PipeConnectionType);
-	PyModule_AddObject(module, "PipeConnection",
-			 (PyObject*)&PipeConnectionType);
-
-	/* Initialize win32 class and add to multiprocessing */
-	temp = create_win32_namespace();
-	if (!temp)
-		return;
-	PyModule_AddObject(module, "win32", temp);
-
-	/* Initialize the event handle used to signal Ctrl-C */
-	sigint_event = CreateEvent(NULL, TRUE, FALSE, NULL);
-	if (!sigint_event) {
-		PyErr_SetFromWindowsErr(0);
-		return;
-	}
-	if (!SetConsoleCtrlHandler(ProcessingCtrlHandler, TRUE)) {
-		PyErr_SetFromWindowsErr(0);
-		return;
-	}
-#endif
-
-	/* Add configuration macros */
-	temp = PyDict_New();
-	if (!temp)
-		return;
-#define ADD_FLAG(name)						 \
-	value = Py_BuildValue("i", name);			 \
-	if (value == NULL) { Py_DECREF(temp); return; }		 \
-	if (PyDict_SetItemString(temp, #name, value) < 0) {	 \
-		Py_DECREF(temp); Py_DECREF(value); return; }	 \
-	Py_DECREF(value)
-	
-#ifdef HAVE_SEM_OPEN
-	ADD_FLAG(HAVE_SEM_OPEN);
-#endif
-#ifdef HAVE_SEM_TIMEDWAIT
-	ADD_FLAG(HAVE_SEM_TIMEDWAIT);
-#endif
-#ifdef HAVE_FD_TRANSFER
-	ADD_FLAG(HAVE_FD_TRANSFER);
-#endif
-#ifdef HAVE_BROKEN_SEM_GETVALUE
-	ADD_FLAG(HAVE_BROKEN_SEM_GETVALUE);
-#endif
-#ifdef HAVE_BROKEN_SEM_UNLINK
-	ADD_FLAG(HAVE_BROKEN_SEM_UNLINK);
-#endif
-	if (PyModule_AddObject(module, "flags", temp) < 0)
-		return;
-}
+/*
+ * Extension module used by multiprocessing package
+ *
+ * multiprocessing.c
+ *
+ * Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
+ */
+
+#include "multiprocessing.h"
+
+PyObject *create_win32_namespace(void);
+
+PyObject *pickle_dumps, *pickle_loads, *pickle_protocol;
+PyObject *ProcessError, *BufferTooShort;
+
+/*
+ * Function which raises exceptions based on error codes
+ */
+
+PyObject *
+mp_SetError(PyObject *Type, int num)
+{
+	switch (num) {
+#ifdef MS_WINDOWS
+	case MP_STANDARD_ERROR: 
+		if (Type == NULL)
+			Type = PyExc_WindowsError;
+		PyErr_SetExcFromWindowsErr(Type, 0);
+		break;
+	case MP_SOCKET_ERROR:
+		if (Type == NULL)
+			Type = PyExc_WindowsError;
+		PyErr_SetExcFromWindowsErr(Type, WSAGetLastError());
+		break;
+#else /* !MS_WINDOWS */
+	case MP_STANDARD_ERROR:
+	case MP_SOCKET_ERROR:
+		if (Type == NULL)
+			Type = PyExc_OSError;
+		PyErr_SetFromErrno(Type);
+		break;
+#endif /* !MS_WINDOWS */
+	case MP_MEMORY_ERROR:
+		PyErr_NoMemory();
+		break;
+	case MP_END_OF_FILE:
+		PyErr_SetNone(PyExc_EOFError);
+		break;
+	case MP_EARLY_END_OF_FILE:
+		PyErr_SetString(PyExc_IOError,
+				"got end of file during message");
+		break;
+	case MP_BAD_MESSAGE_LENGTH:
+		PyErr_SetString(PyExc_IOError, "bad message length");
+		break;
+	case MP_EXCEPTION_HAS_BEEN_SET:
+		break;
+	default:
+		PyErr_Format(PyExc_RuntimeError,
+			 "unkown error number %d", num);
+	}
+	return NULL;
+}
+
+
+/*
+ * Windows only
+ */
+
+#ifdef MS_WINDOWS
+
+/* On Windows we set an event to signal Ctrl-C; compare with timemodule.c */
+
+HANDLE sigint_event = NULL;
+
+static BOOL WINAPI
+ProcessingCtrlHandler(DWORD dwCtrlType)
+{
+	SetEvent(sigint_event);
+	return FALSE;
+}
+
+/*
+ * Unix only
+ */
+
+#else /* !MS_WINDOWS */
+
+#if HAVE_FD_TRANSFER
+
+/* Functions for transferring file descriptors between processes.
+ Reimplements some of the functionality of the fdcred
+ module at http://www.mca-ltd.com/resources/fdcred_1.tgz. */
+
+static PyObject *
+multiprocessing_sendfd(PyObject *self, PyObject *args)
+{
+	int conn, fd, res;
+	char dummy_char;
+	char buf[CMSG_SPACE(sizeof(int))];
+	struct msghdr msg = {0};
+	struct iovec dummy_iov;
+	struct cmsghdr *cmsg;
+
+	if (!PyArg_ParseTuple(args, "ii", &conn, &fd))
+		return NULL;
+
+	dummy_iov.iov_base = &dummy_char;
+	dummy_iov.iov_len = 1;
+	msg.msg_control = buf;
+	msg.msg_controllen = sizeof(buf);
+	msg.msg_iov = &dummy_iov;
+	msg.msg_iovlen = 1;
+	cmsg = CMSG_FIRSTHDR(&msg);
+	cmsg->cmsg_level = SOL_SOCKET;
+	cmsg->cmsg_type = SCM_RIGHTS;
+	cmsg->cmsg_len = CMSG_LEN(sizeof(int));
+	msg.msg_controllen = cmsg->cmsg_len;
+	*(int*)CMSG_DATA(cmsg) = fd;
+
+	Py_BEGIN_ALLOW_THREADS
+	res = sendmsg(conn, &msg, 0);
+	Py_END_ALLOW_THREADS
+
+	if (res < 0)
+		return PyErr_SetFromErrno(PyExc_OSError);
+	Py_RETURN_NONE;
+}
+
+static PyObject *
+multiprocessing_recvfd(PyObject *self, PyObject *args)
+{
+	int conn, fd, res;
+	char dummy_char;
+	char buf[CMSG_SPACE(sizeof(int))];
+	struct msghdr msg = {0};
+	struct iovec dummy_iov;
+	struct cmsghdr *cmsg;
+
+	if (!PyArg_ParseTuple(args, "i", &conn))
+		return NULL;
+
+	dummy_iov.iov_base = &dummy_char;
+	dummy_iov.iov_len = 1;
+	msg.msg_control = buf;
+	msg.msg_controllen = sizeof(buf);
+	msg.msg_iov = &dummy_iov;
+	msg.msg_iovlen = 1;
+	cmsg = CMSG_FIRSTHDR(&msg);
+	cmsg->cmsg_level = SOL_SOCKET;
+	cmsg->cmsg_type = SCM_RIGHTS;
+	cmsg->cmsg_len = CMSG_LEN(sizeof(int));
+	msg.msg_controllen = cmsg->cmsg_len;
+
+	Py_BEGIN_ALLOW_THREADS
+	res = recvmsg(conn, &msg, 0);
+	Py_END_ALLOW_THREADS
+
+	if (res < 0)
+		return PyErr_SetFromErrno(PyExc_OSError);
+
+	fd = *(int*)CMSG_DATA(cmsg);
+	return Py_BuildValue("i", fd);
+}
+
+#endif /* HAVE_FD_TRANSFER */
+
+#endif /* !MS_WINDOWS */
+
+
+/*
+ * All platforms
+ */
+
+static PyObject*
+multiprocessing_address_of_buffer(PyObject *self, PyObject *obj)
+{
+	void *buffer;
+	Py_ssize_t buffer_len;
+
+	if (PyObject_AsWriteBuffer(obj, &buffer, &buffer_len) < 0)
+		return NULL;
+
+	return Py_BuildValue("N" F_PY_SSIZE_T, 
+			 PyLong_FromVoidPtr(buffer), buffer_len);
+}
+
+
+/*
+ * Function table
+ */
+
+static PyMethodDef module_methods[] = {
+	{"address_of_buffer", multiprocessing_address_of_buffer, METH_O, 
+	 "address_of_buffer(obj) -> int\n" 
+	 "Return address of obj assuming obj supports buffer inteface"},
+#if HAVE_FD_TRANSFER
+	{"sendfd", multiprocessing_sendfd, METH_VARARGS, 
+	 "sendfd(sockfd, fd) -> None\n"
+	 "Send file descriptor given by fd over the unix domain socket\n"
+	 "whose file decriptor is sockfd"},
+	{"recvfd", multiprocessing_recvfd, METH_VARARGS,
+	 "recvfd(sockfd) -> fd\n"
+	 "Receive a file descriptor over a unix domain socket\n"
+	 "whose file decriptor is sockfd"},
+#endif
+	{NULL}
+};
+
+
+/*
+ * Initialize
+ */
+
+PyMODINIT_FUNC
+init_multiprocessing(void)
+{
+	PyObject *module, *temp, *value;
+
+	/* Initialize module */
+	module = Py_InitModule("_multiprocessing", module_methods);
+	if (!module)
+		return;
+
+	/* Get copy of objects from pickle */
+	temp = PyImport_ImportModule(PICKLE_MODULE);
+	if (!temp)
+		return;
+	pickle_dumps = PyObject_GetAttrString(temp, "dumps");
+	pickle_loads = PyObject_GetAttrString(temp, "loads");
+	pickle_protocol = PyObject_GetAttrString(temp, "HIGHEST_PROTOCOL");
+	Py_XDECREF(temp);
+
+	/* Get copy of BufferTooShort */
+	temp = PyImport_ImportModule("multiprocessing");
+	if (!temp)
+		return;
+	BufferTooShort = PyObject_GetAttrString(temp, "BufferTooShort");
+	Py_XDECREF(temp);
+
+	/* Add connection type to module */
+	if (PyType_Ready(&ConnectionType) < 0)
+		return;
+	Py_INCREF(&ConnectionType);	
+	PyModule_AddObject(module, "Connection", (PyObject*)&ConnectionType);
+
+#if defined(MS_WINDOWS) || HAVE_SEM_OPEN
+	/* Add SemLock type to module */
+	if (PyType_Ready(&SemLockType) < 0)
+		return;
+	Py_INCREF(&SemLockType);
+	PyDict_SetItemString(SemLockType.tp_dict, "SEM_VALUE_MAX", 
+			 Py_BuildValue("i", SEM_VALUE_MAX));
+	PyModule_AddObject(module, "SemLock", (PyObject*)&SemLockType); 
+#endif
+
+#ifdef MS_WINDOWS
+	/* Add PipeConnection to module */
+	if (PyType_Ready(&PipeConnectionType) < 0)
+		return;
+	Py_INCREF(&PipeConnectionType);
+	PyModule_AddObject(module, "PipeConnection",
+			 (PyObject*)&PipeConnectionType);
+
+	/* Initialize win32 class and add to multiprocessing */
+	temp = create_win32_namespace();
+	if (!temp)
+		return;
+	PyModule_AddObject(module, "win32", temp);
+
+	/* Initialize the event handle used to signal Ctrl-C */
+	sigint_event = CreateEvent(NULL, TRUE, FALSE, NULL);
+	if (!sigint_event) {
+		PyErr_SetFromWindowsErr(0);
+		return;
+	}
+	if (!SetConsoleCtrlHandler(ProcessingCtrlHandler, TRUE)) {
+		PyErr_SetFromWindowsErr(0);
+		return;
+	}
+#endif
+
+	/* Add configuration macros */
+	temp = PyDict_New();
+	if (!temp)
+		return;
+#define ADD_FLAG(name)						 \
+	value = Py_BuildValue("i", name);			 \
+	if (value == NULL) { Py_DECREF(temp); return; }		 \
+	if (PyDict_SetItemString(temp, #name, value) < 0) {	 \
+		Py_DECREF(temp); Py_DECREF(value); return; }	 \
+	Py_DECREF(value)
+	
+#ifdef HAVE_SEM_OPEN
+	ADD_FLAG(HAVE_SEM_OPEN);
+#endif
+#ifdef HAVE_SEM_TIMEDWAIT
+	ADD_FLAG(HAVE_SEM_TIMEDWAIT);
+#endif
+#ifdef HAVE_FD_TRANSFER
+	ADD_FLAG(HAVE_FD_TRANSFER);
+#endif
+#ifdef HAVE_BROKEN_SEM_GETVALUE
+	ADD_FLAG(HAVE_BROKEN_SEM_GETVALUE);
+#endif
+#ifdef HAVE_BROKEN_SEM_UNLINK
+	ADD_FLAG(HAVE_BROKEN_SEM_UNLINK);
+#endif
+	if (PyModule_AddObject(module, "flags", temp) < 0)
+		return;
+}
Modified: python/trunk/Modules/_multiprocessing/multiprocessing.h
==============================================================================
--- python/trunk/Modules/_multiprocessing/multiprocessing.h	(original)
+++ python/trunk/Modules/_multiprocessing/multiprocessing.h	Fri Jun 13 21:13:39 2008
@@ -1,163 +1,163 @@
-#ifndef MULTIPROCESSING_H
-#define MULTIPROCESSING_H
-
-#define PY_SSIZE_T_CLEAN
-
-#include "Python.h"
-#include "structmember.h"
-#include "pythread.h"
-
-/*
- * Platform includes and definitions
- */
-
-#ifdef MS_WINDOWS
-# define WIN32_LEAN_AND_MEAN
-# include <windows.h>
-# include <winsock2.h>
-# include <process.h>		 /* getpid() */
-# define SEM_HANDLE HANDLE
-# define SEM_VALUE_MAX LONG_MAX
-#else
-# include <fcntl.h> /* O_CREAT and O_EXCL */
-# include <sys/socket.h>
-# include <arpa/inet.h> /* htonl() and ntohl() */
-# if HAVE_SEM_OPEN
-# include <semaphore.h>
- typedef sem_t *SEM_HANDLE;
-# endif
-# define HANDLE int
-# define SOCKET int
-# define BOOL int
-# define UINT32 uint32_t
-# define INT32 int32_t
-# define TRUE 1
-# define FALSE 0
-# define INVALID_HANDLE_VALUE (-1)
-#endif
-
-/*
- * Make sure Py_ssize_t available
- */
-
-#if PY_VERSION_HEX < 0x02050000 && !defined(PY_SSIZE_T_MIN)
- typedef int Py_ssize_t;
-# define PY_SSIZE_T_MAX INT_MAX
-# define PY_SSIZE_T_MIN INT_MIN
-# define F_PY_SSIZE_T "i"
-# define PY_FORMAT_SIZE_T ""
-# define PyInt_FromSsize_t(n) PyInt_FromLong((long)n)
-#else
-# define F_PY_SSIZE_T "n"
-#endif
-
-/*
- * Format codes
- */
-
-#if SIZEOF_VOID_P == SIZEOF_LONG
-# define F_POINTER "k"
-# define T_POINTER T_ULONG
-#elif defined(HAVE_LONG_LONG) && (SIZEOF_VOID_P == SIZEOF_LONG_LONG)
-# define F_POINTER "K"
-# define T_POINTER T_ULONGLONG
-#else
-# error "can't find format code for unsigned integer of same size as void*"
-#endif
-
-#ifdef MS_WINDOWS
-# define F_HANDLE F_POINTER
-# define T_HANDLE T_POINTER
-# define F_SEM_HANDLE F_HANDLE
-# define T_SEM_HANDLE T_HANDLE
-# define F_DWORD "k"
-# define T_DWORD T_ULONG
-#else
-# define F_HANDLE "i"
-# define T_HANDLE T_INT
-# define F_SEM_HANDLE F_POINTER
-# define T_SEM_HANDLE T_POINTER
-#endif
-
-#if PY_VERSION_HEX >= 0x03000000
-# define F_RBUFFER "y"
-#else
-# define F_RBUFFER "s"
-#endif
-
-/*
- * Error codes which can be returned by functions called without GIL
- */
-
-#define MP_SUCCESS (0)
-#define MP_STANDARD_ERROR (-1)
-#define MP_MEMORY_ERROR (-1001)
-#define MP_END_OF_FILE (-1002)
-#define MP_EARLY_END_OF_FILE (-1003)
-#define MP_BAD_MESSAGE_LENGTH (-1004)
-#define MP_SOCKET_ERROR (-1005)
-#define MP_EXCEPTION_HAS_BEEN_SET (-1006)
-
-PyObject *mp_SetError(PyObject *Type, int num);
-
-/*
- * Externs - not all will really exist on all platforms
- */
-
-extern PyObject *pickle_dumps;
-extern PyObject *pickle_loads;
-extern PyObject *pickle_protocol;
-extern PyObject *BufferTooShort;
-extern PyTypeObject SemLockType;
-extern PyTypeObject ConnectionType;
-extern PyTypeObject PipeConnectionType;
-extern HANDLE sigint_event;
-
-/*
- * Py3k compatibility
- */
-
-#if PY_VERSION_HEX >= 0x03000000
-# define PICKLE_MODULE "pickle"
-# define FROM_FORMAT PyUnicode_FromFormat
-# define PyInt_FromLong PyLong_FromLong
-# define PyInt_FromSsize_t PyLong_FromSsize_t
-#else
-# define PICKLE_MODULE "cPickle"
-# define FROM_FORMAT PyString_FromFormat
-#endif
-
-#ifndef PyVarObject_HEAD_INIT
-# define PyVarObject_HEAD_INIT(type, size) PyObject_HEAD_INIT(type) size,
-#endif
-
-#ifndef Py_TPFLAGS_HAVE_WEAKREFS
-# define Py_TPFLAGS_HAVE_WEAKREFS 0
-#endif
-
-/*
- * Connection definition
- */
-
-#define CONNECTION_BUFFER_SIZE 1024
-
-typedef struct {
-	PyObject_HEAD
-	HANDLE handle;
-	int flags;
-	PyObject *weakreflist;
-	char buffer[CONNECTION_BUFFER_SIZE];
-} ConnectionObject;
-
-/*
- * Miscellaneous
- */
-
-#define MAX_MESSAGE_LENGTH 0x7fffffff
-
-#ifndef MIN
-# define MIN(x, y) ((x) < (y) ? x : y)
-# define MAX(x, y) ((x) > (y) ? x : y)
-#endif
-
-#endif /* MULTIPROCESSING_H */
+#ifndef MULTIPROCESSING_H
+#define MULTIPROCESSING_H
+
+#define PY_SSIZE_T_CLEAN
+
+#include "Python.h"
+#include "structmember.h"
+#include "pythread.h"
+
+/*
+ * Platform includes and definitions
+ */
+
+#ifdef MS_WINDOWS
+# define WIN32_LEAN_AND_MEAN
+# include <windows.h>
+# include <winsock2.h>
+# include <process.h>		 /* getpid() */
+# define SEM_HANDLE HANDLE
+# define SEM_VALUE_MAX LONG_MAX
+#else
+# include <fcntl.h> /* O_CREAT and O_EXCL */
+# include <sys/socket.h>
+# include <arpa/inet.h> /* htonl() and ntohl() */
+# if HAVE_SEM_OPEN
+# include <semaphore.h>
+ typedef sem_t *SEM_HANDLE;
+# endif
+# define HANDLE int
+# define SOCKET int
+# define BOOL int
+# define UINT32 uint32_t
+# define INT32 int32_t
+# define TRUE 1
+# define FALSE 0
+# define INVALID_HANDLE_VALUE (-1)
+#endif
+
+/*
+ * Make sure Py_ssize_t available
+ */
+
+#if PY_VERSION_HEX < 0x02050000 && !defined(PY_SSIZE_T_MIN)
+ typedef int Py_ssize_t;
+# define PY_SSIZE_T_MAX INT_MAX
+# define PY_SSIZE_T_MIN INT_MIN
+# define F_PY_SSIZE_T "i"
+# define PY_FORMAT_SIZE_T ""
+# define PyInt_FromSsize_t(n) PyInt_FromLong((long)n)
+#else
+# define F_PY_SSIZE_T "n"
+#endif
+
+/*
+ * Format codes
+ */
+
+#if SIZEOF_VOID_P == SIZEOF_LONG
+# define F_POINTER "k"
+# define T_POINTER T_ULONG
+#elif defined(HAVE_LONG_LONG) && (SIZEOF_VOID_P == SIZEOF_LONG_LONG)
+# define F_POINTER "K"
+# define T_POINTER T_ULONGLONG
+#else
+# error "can't find format code for unsigned integer of same size as void*"
+#endif
+
+#ifdef MS_WINDOWS
+# define F_HANDLE F_POINTER
+# define T_HANDLE T_POINTER
+# define F_SEM_HANDLE F_HANDLE
+# define T_SEM_HANDLE T_HANDLE
+# define F_DWORD "k"
+# define T_DWORD T_ULONG
+#else
+# define F_HANDLE "i"
+# define T_HANDLE T_INT
+# define F_SEM_HANDLE F_POINTER
+# define T_SEM_HANDLE T_POINTER
+#endif
+
+#if PY_VERSION_HEX >= 0x03000000
+# define F_RBUFFER "y"
+#else
+# define F_RBUFFER "s"
+#endif
+
+/*
+ * Error codes which can be returned by functions called without GIL
+ */
+
+#define MP_SUCCESS (0)
+#define MP_STANDARD_ERROR (-1)
+#define MP_MEMORY_ERROR (-1001)
+#define MP_END_OF_FILE (-1002)
+#define MP_EARLY_END_OF_FILE (-1003)
+#define MP_BAD_MESSAGE_LENGTH (-1004)
+#define MP_SOCKET_ERROR (-1005)
+#define MP_EXCEPTION_HAS_BEEN_SET (-1006)
+
+PyObject *mp_SetError(PyObject *Type, int num);
+
+/*
+ * Externs - not all will really exist on all platforms
+ */
+
+extern PyObject *pickle_dumps;
+extern PyObject *pickle_loads;
+extern PyObject *pickle_protocol;
+extern PyObject *BufferTooShort;
+extern PyTypeObject SemLockType;
+extern PyTypeObject ConnectionType;
+extern PyTypeObject PipeConnectionType;
+extern HANDLE sigint_event;
+
+/*
+ * Py3k compatibility
+ */
+
+#if PY_VERSION_HEX >= 0x03000000
+# define PICKLE_MODULE "pickle"
+# define FROM_FORMAT PyUnicode_FromFormat
+# define PyInt_FromLong PyLong_FromLong
+# define PyInt_FromSsize_t PyLong_FromSsize_t
+#else
+# define PICKLE_MODULE "cPickle"
+# define FROM_FORMAT PyString_FromFormat
+#endif
+
+#ifndef PyVarObject_HEAD_INIT
+# define PyVarObject_HEAD_INIT(type, size) PyObject_HEAD_INIT(type) size,
+#endif
+
+#ifndef Py_TPFLAGS_HAVE_WEAKREFS
+# define Py_TPFLAGS_HAVE_WEAKREFS 0
+#endif
+
+/*
+ * Connection definition
+ */
+
+#define CONNECTION_BUFFER_SIZE 1024
+
+typedef struct {
+	PyObject_HEAD
+	HANDLE handle;
+	int flags;
+	PyObject *weakreflist;
+	char buffer[CONNECTION_BUFFER_SIZE];
+} ConnectionObject;
+
+/*
+ * Miscellaneous
+ */
+
+#define MAX_MESSAGE_LENGTH 0x7fffffff
+
+#ifndef MIN
+# define MIN(x, y) ((x) < (y) ? x : y)
+# define MAX(x, y) ((x) > (y) ? x : y)
+#endif
+
+#endif /* MULTIPROCESSING_H */
Modified: python/trunk/Modules/_multiprocessing/pipe_connection.c
==============================================================================
--- python/trunk/Modules/_multiprocessing/pipe_connection.c	(original)
+++ python/trunk/Modules/_multiprocessing/pipe_connection.c	Fri Jun 13 21:13:39 2008
@@ -1,136 +1,136 @@
-/*
- * A type which wraps a pipe handle in message oriented mode
- *
- * pipe_connection.c
- *
- * Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
- */
-
-#include "multiprocessing.h"
-
-#define CLOSE(h) CloseHandle(h)
-
-/*
- * Send string to the pipe; assumes in message oriented mode
- */
-
-static Py_ssize_t
-conn_send_string(ConnectionObject *conn, char *string, size_t length)
-{
-	DWORD amount_written;
-
-	return WriteFile(conn->handle, string, length, &amount_written, NULL)
-		? MP_SUCCESS : MP_STANDARD_ERROR;
-}
-
-/*
- * Attempts to read into buffer, or if buffer too small into *newbuffer.
- *
- * Returns number of bytes read. Assumes in message oriented mode.
- */
-
-static Py_ssize_t
-conn_recv_string(ConnectionObject *conn, char *buffer, 
-		 size_t buflength, char **newbuffer, size_t maxlength)
-{
-	DWORD left, length, full_length, err;
-
-	*newbuffer = NULL;
-
-	if (ReadFile(conn->handle, buffer, MIN(buflength, maxlength), 
-		 &length, NULL))
-		return length;
-
-	err = GetLastError();
-	if (err != ERROR_MORE_DATA) {
-		if (err == ERROR_BROKEN_PIPE)
-			return MP_END_OF_FILE;
-		return MP_STANDARD_ERROR;
-	}
-
-	if (!PeekNamedPipe(conn->handle, NULL, 0, NULL, NULL, &left))
-		return MP_STANDARD_ERROR;
-
-	full_length = length + left;
-	if (full_length > maxlength)
-		return MP_BAD_MESSAGE_LENGTH;
-
-	*newbuffer = PyMem_Malloc(full_length);
-	if (*newbuffer == NULL)
-		return MP_MEMORY_ERROR;
-
-	memcpy(*newbuffer, buffer, length);
-
-	if (ReadFile(conn->handle, *newbuffer+length, left, &length, NULL)) {
-		assert(length == left);
-		return full_length;
-	} else {
-		PyMem_Free(*newbuffer);
-		return MP_STANDARD_ERROR;
-	}
-}
-
-/*
- * Check whether any data is available for reading
- */
-
-#define conn_poll(conn, timeout) conn_poll_save(conn, timeout, _save)
-
-static int
-conn_poll_save(ConnectionObject *conn, double timeout, PyThreadState *_save)
-{
-	DWORD bytes, deadline, delay;
-	int difference, res;
-	BOOL block = FALSE;
-
-	if (!PeekNamedPipe(conn->handle, NULL, 0, NULL, &bytes, NULL))
-		return MP_STANDARD_ERROR;
-
-	if (timeout == 0.0)
-		return bytes > 0;
-
-	if (timeout < 0.0)
-		block = TRUE;
-	else
-		/* XXX does not check for overflow */
-		deadline = GetTickCount() + (DWORD)(1000 * timeout + 0.5);
-
-	Sleep(0);
-
-	for (delay = 1 ; ; delay += 1) {
-		if (!PeekNamedPipe(conn->handle, NULL, 0, NULL, &bytes, NULL))
-			return MP_STANDARD_ERROR;
-		else if (bytes > 0)
-			return TRUE;
-
-		if (!block) {
-			difference = deadline - GetTickCount();
-			if (difference < 0)
-				return FALSE;
-			if ((int)delay > difference)
-				delay = difference;
-		}
-
-		if (delay > 20)
-			delay = 20;
-
-		Sleep(delay);
-
-		/* check for signals */
-		Py_BLOCK_THREADS 
-		res = PyErr_CheckSignals();
-		Py_UNBLOCK_THREADS
-
-		if (res)
-			return MP_EXCEPTION_HAS_BEEN_SET;
-	}
-}
-
-/*
- * "connection.h" defines the PipeConnection type using the definitions above
- */
-
-#define CONNECTION_NAME "PipeConnection"
-#define CONNECTION_TYPE PipeConnectionType
-
-#include "connection.h"
+/*
+ * A type which wraps a pipe handle in message oriented mode
+ *
+ * pipe_connection.c
+ *
+ * Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
+ */
+
+#include "multiprocessing.h"
+
+#define CLOSE(h) CloseHandle(h)
+
+/*
+ * Send string to the pipe; assumes in message oriented mode
+ */
+
+static Py_ssize_t
+conn_send_string(ConnectionObject *conn, char *string, size_t length)
+{
+	DWORD amount_written;
+
+	return WriteFile(conn->handle, string, length, &amount_written, NULL)
+		? MP_SUCCESS : MP_STANDARD_ERROR;
+}
+
+/*
+ * Attempts to read into buffer, or if buffer too small into *newbuffer.
+ *
+ * Returns number of bytes read. Assumes in message oriented mode.
+ */
+
+static Py_ssize_t
+conn_recv_string(ConnectionObject *conn, char *buffer, 
+		 size_t buflength, char **newbuffer, size_t maxlength)
+{
+	DWORD left, length, full_length, err;
+
+	*newbuffer = NULL;
+
+	if (ReadFile(conn->handle, buffer, MIN(buflength, maxlength), 
+		 &length, NULL))
+		return length;
+
+	err = GetLastError();
+	if (err != ERROR_MORE_DATA) {
+		if (err == ERROR_BROKEN_PIPE)
+			return MP_END_OF_FILE;
+		return MP_STANDARD_ERROR;
+	}
+
+	if (!PeekNamedPipe(conn->handle, NULL, 0, NULL, NULL, &left))
+		return MP_STANDARD_ERROR;
+
+	full_length = length + left;
+	if (full_length > maxlength)
+		return MP_BAD_MESSAGE_LENGTH;
+
+	*newbuffer = PyMem_Malloc(full_length);
+	if (*newbuffer == NULL)
+		return MP_MEMORY_ERROR;
+
+	memcpy(*newbuffer, buffer, length);
+
+	if (ReadFile(conn->handle, *newbuffer+length, left, &length, NULL)) {
+		assert(length == left);
+		return full_length;
+	} else {
+		PyMem_Free(*newbuffer);
+		return MP_STANDARD_ERROR;
+	}
+}
+
+/*
+ * Check whether any data is available for reading
+ */
+
+#define conn_poll(conn, timeout) conn_poll_save(conn, timeout, _save)
+
+static int
+conn_poll_save(ConnectionObject *conn, double timeout, PyThreadState *_save)
+{
+	DWORD bytes, deadline, delay;
+	int difference, res;
+	BOOL block = FALSE;
+
+	if (!PeekNamedPipe(conn->handle, NULL, 0, NULL, &bytes, NULL))
+		return MP_STANDARD_ERROR;
+
+	if (timeout == 0.0)
+		return bytes > 0;
+
+	if (timeout < 0.0)
+		block = TRUE;
+	else
+		/* XXX does not check for overflow */
+		deadline = GetTickCount() + (DWORD)(1000 * timeout + 0.5);
+
+	Sleep(0);
+
+	for (delay = 1 ; ; delay += 1) {
+		if (!PeekNamedPipe(conn->handle, NULL, 0, NULL, &bytes, NULL))
+			return MP_STANDARD_ERROR;
+		else if (bytes > 0)
+			return TRUE;
+
+		if (!block) {
+			difference = deadline - GetTickCount();
+			if (difference < 0)
+				return FALSE;
+			if ((int)delay > difference)
+				delay = difference;
+		}
+
+		if (delay > 20)
+			delay = 20;
+
+		Sleep(delay);
+
+		/* check for signals */
+		Py_BLOCK_THREADS 
+		res = PyErr_CheckSignals();
+		Py_UNBLOCK_THREADS
+
+		if (res)
+			return MP_EXCEPTION_HAS_BEEN_SET;
+	}
+}
+
+/*
+ * "connection.h" defines the PipeConnection type using the definitions above
+ */
+
+#define CONNECTION_NAME "PipeConnection"
+#define CONNECTION_TYPE PipeConnectionType
+
+#include "connection.h"
Modified: python/trunk/Modules/_multiprocessing/win32_functions.c
==============================================================================
--- python/trunk/Modules/_multiprocessing/win32_functions.c	(original)
+++ python/trunk/Modules/_multiprocessing/win32_functions.c	Fri Jun 13 21:13:39 2008
@@ -1,260 +1,260 @@
-/*
- * Win32 functions used by multiprocessing package
- *
- * win32_functions.c
- *
- * Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
- */
-
-#include "multiprocessing.h"
-
-
-#define WIN32_FUNCTION(func) \
- {#func, (PyCFunction)win32_ ## func, METH_VARARGS | METH_STATIC, ""}
-
-#define WIN32_CONSTANT(fmt, con) \
- PyDict_SetItemString(Win32Type.tp_dict, #con, Py_BuildValue(fmt, con))
-
-
-static PyObject *
-win32_CloseHandle(PyObject *self, PyObject *args)
-{
-	HANDLE hObject;
-	BOOL success;
-
-	if (!PyArg_ParseTuple(args, F_HANDLE, &hObject))
-		return NULL;
-
-	Py_BEGIN_ALLOW_THREADS
-	success = CloseHandle(hObject); 
-	Py_END_ALLOW_THREADS
-
-	if (!success)
-		return PyErr_SetFromWindowsErr(0);
-
-	Py_RETURN_NONE;
-}
-
-static PyObject *
-win32_ConnectNamedPipe(PyObject *self, PyObject *args)
-{
-	HANDLE hNamedPipe;
-	LPOVERLAPPED lpOverlapped;
-	BOOL success;
-
-	if (!PyArg_ParseTuple(args, F_HANDLE F_POINTER, 
-			 &hNamedPipe, &lpOverlapped))
-		return NULL;
-
-	Py_BEGIN_ALLOW_THREADS
-	success = ConnectNamedPipe(hNamedPipe, lpOverlapped);
-	Py_END_ALLOW_THREADS
-
-	if (!success)
-		return PyErr_SetFromWindowsErr(0);
-
-	Py_RETURN_NONE;
-}
-
-static PyObject *
-win32_CreateFile(PyObject *self, PyObject *args)
-{
-	LPCTSTR lpFileName;
-	DWORD dwDesiredAccess;
-	DWORD dwShareMode;
-	LPSECURITY_ATTRIBUTES lpSecurityAttributes;
-	DWORD dwCreationDisposition;
-	DWORD dwFlagsAndAttributes;
-	HANDLE hTemplateFile;
-	HANDLE handle;
-
-	if (!PyArg_ParseTuple(args, "s" F_DWORD F_DWORD F_POINTER 
-			 F_DWORD F_DWORD F_HANDLE,
-			 &lpFileName, &dwDesiredAccess, &dwShareMode, 
-			 &lpSecurityAttributes, &dwCreationDisposition, 
-			 &dwFlagsAndAttributes, &hTemplateFile))
-		return NULL;
-
-	Py_BEGIN_ALLOW_THREADS
-	handle = CreateFile(lpFileName, dwDesiredAccess, 
-			 dwShareMode, lpSecurityAttributes, 
-			 dwCreationDisposition, 
-			 dwFlagsAndAttributes, hTemplateFile);
-	Py_END_ALLOW_THREADS
-
-	if (handle == INVALID_HANDLE_VALUE)
-		return PyErr_SetFromWindowsErr(0);
-
-	return Py_BuildValue(F_HANDLE, handle);
-}
-
-static PyObject *
-win32_CreateNamedPipe(PyObject *self, PyObject *args)
-{
-	LPCTSTR lpName;
-	DWORD dwOpenMode;
-	DWORD dwPipeMode;
-	DWORD nMaxInstances;
-	DWORD nOutBufferSize;
-	DWORD nInBufferSize;
-	DWORD nDefaultTimeOut;
-	LPSECURITY_ATTRIBUTES lpSecurityAttributes;
-	HANDLE handle;
-
-	if (!PyArg_ParseTuple(args, "s" F_DWORD F_DWORD F_DWORD 
-			 F_DWORD F_DWORD F_DWORD F_POINTER,
-			 &lpName, &dwOpenMode, &dwPipeMode, 
-			 &nMaxInstances, &nOutBufferSize, 
-			 &nInBufferSize, &nDefaultTimeOut,
-			 &lpSecurityAttributes))
-		return NULL;
-
-	Py_BEGIN_ALLOW_THREADS
-	handle = CreateNamedPipe(lpName, dwOpenMode, dwPipeMode, 
-				 nMaxInstances, nOutBufferSize, 
-				 nInBufferSize, nDefaultTimeOut,
-				 lpSecurityAttributes);
-	Py_END_ALLOW_THREADS
-
-	if (handle == INVALID_HANDLE_VALUE)
-		return PyErr_SetFromWindowsErr(0);
-
-	return Py_BuildValue(F_HANDLE, handle);
-}
-
-static PyObject *
-win32_ExitProcess(PyObject *self, PyObject *args)
-{
-	UINT uExitCode;
-
-	if (!PyArg_ParseTuple(args, "I", &uExitCode))
-		return NULL;
-
-	ExitProcess(uExitCode);
-
-	return NULL;
-}
-
-static PyObject *
-win32_GetLastError(PyObject *self, PyObject *args)
-{
-	return Py_BuildValue(F_DWORD, GetLastError());
-}
-
-static PyObject *
-win32_OpenProcess(PyObject *self, PyObject *args)
-{
-	DWORD dwDesiredAccess;
-	BOOL bInheritHandle;
-	DWORD dwProcessId;
-	HANDLE handle;
-
-	if (!PyArg_ParseTuple(args, F_DWORD "i" F_DWORD, 
-			 &dwDesiredAccess, &bInheritHandle, &dwProcessId))
-		return NULL;
-
-	handle = OpenProcess(dwDesiredAccess, bInheritHandle, dwProcessId); 
-	if (handle == NULL)
-		return PyErr_SetFromWindowsErr(0);
-
-	return Py_BuildValue(F_HANDLE, handle);
-}
-
-static PyObject *
-win32_SetNamedPipeHandleState(PyObject *self, PyObject *args)
-{
-	HANDLE hNamedPipe;
-	PyObject *oArgs[3];
-	DWORD dwArgs[3], *pArgs[3] = {NULL, NULL, NULL};
-	int i;
-
-	if (!PyArg_ParseTuple(args, F_HANDLE "OOO", 
-			 &hNamedPipe, &oArgs[0], &oArgs[1], &oArgs[2]))
-		return NULL;
-
-	PyErr_Clear();
-
-	for (i = 0 ; i < 3 ; i++) {
-		if (oArgs[i] != Py_None) {
-			dwArgs[i] = PyInt_AsUnsignedLongMask(oArgs[i]);
-			if (PyErr_Occurred())
-				return NULL;
-			pArgs[i] = &dwArgs[i];
-		}
-	}
-
-	if (!SetNamedPipeHandleState(hNamedPipe, pArgs[0], pArgs[1], pArgs[2]))
-		return PyErr_SetFromWindowsErr(0);
-
-	Py_RETURN_NONE;
-}
-
-static PyObject *
-win32_WaitNamedPipe(PyObject *self, PyObject *args)
-{
-	LPCTSTR lpNamedPipeName;
-	DWORD nTimeOut;
-	BOOL success;
-
-	if (!PyArg_ParseTuple(args, "s" F_DWORD, &lpNamedPipeName, &nTimeOut))
-		return NULL;
-
-	Py_BEGIN_ALLOW_THREADS
-	success = WaitNamedPipe(lpNamedPipeName, nTimeOut);
-	Py_END_ALLOW_THREADS
-
-	if (!success)
-		return PyErr_SetFromWindowsErr(0);
-
-	Py_RETURN_NONE;
-}
-
-static PyMethodDef win32_methods[] = {
-	WIN32_FUNCTION(CloseHandle),
-	WIN32_FUNCTION(GetLastError),
-	WIN32_FUNCTION(OpenProcess),
-	WIN32_FUNCTION(ExitProcess),
-	WIN32_FUNCTION(ConnectNamedPipe),
-	WIN32_FUNCTION(CreateFile),
-	WIN32_FUNCTION(CreateNamedPipe),
-	WIN32_FUNCTION(SetNamedPipeHandleState),
-	WIN32_FUNCTION(WaitNamedPipe),
-	{NULL}
-};
-
-
-PyTypeObject Win32Type = {
-	PyVarObject_HEAD_INIT(NULL, 0)
-};
-
-
-PyObject *
-create_win32_namespace(void)
-{
-	Win32Type.tp_name = "_multiprocessing.win32";
-	Win32Type.tp_methods = win32_methods;
-	if (PyType_Ready(&Win32Type) < 0)
-		return NULL;
-	Py_INCREF(&Win32Type);
-
-	WIN32_CONSTANT(F_DWORD, ERROR_ALREADY_EXISTS);
-	WIN32_CONSTANT(F_DWORD, ERROR_PIPE_BUSY);
-	WIN32_CONSTANT(F_DWORD, ERROR_PIPE_CONNECTED);
-	WIN32_CONSTANT(F_DWORD, ERROR_SEM_TIMEOUT);
-	WIN32_CONSTANT(F_DWORD, GENERIC_READ);
-	WIN32_CONSTANT(F_DWORD, GENERIC_WRITE);
-	WIN32_CONSTANT(F_DWORD, INFINITE);
-	WIN32_CONSTANT(F_DWORD, NMPWAIT_WAIT_FOREVER);
-	WIN32_CONSTANT(F_DWORD, OPEN_EXISTING);
-	WIN32_CONSTANT(F_DWORD, PIPE_ACCESS_DUPLEX);
-	WIN32_CONSTANT(F_DWORD, PIPE_ACCESS_INBOUND);
-	WIN32_CONSTANT(F_DWORD, PIPE_READMODE_MESSAGE);
-	WIN32_CONSTANT(F_DWORD, PIPE_TYPE_MESSAGE);
-	WIN32_CONSTANT(F_DWORD, PIPE_UNLIMITED_INSTANCES);
-	WIN32_CONSTANT(F_DWORD, PIPE_WAIT);
-	WIN32_CONSTANT(F_DWORD, PROCESS_ALL_ACCESS);
-
-	WIN32_CONSTANT("i", NULL);
-
-	return (PyObject*)&Win32Type;
-}
+/*
+ * Win32 functions used by multiprocessing package
+ *
+ * win32_functions.c
+ *
+ * Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
+ */
+
+#include "multiprocessing.h"
+
+
+#define WIN32_FUNCTION(func) \
+ {#func, (PyCFunction)win32_ ## func, METH_VARARGS | METH_STATIC, ""}
+
+#define WIN32_CONSTANT(fmt, con) \
+ PyDict_SetItemString(Win32Type.tp_dict, #con, Py_BuildValue(fmt, con))
+
+
+static PyObject *
+win32_CloseHandle(PyObject *self, PyObject *args)
+{
+	HANDLE hObject;
+	BOOL success;
+
+	if (!PyArg_ParseTuple(args, F_HANDLE, &hObject))
+		return NULL;
+
+	Py_BEGIN_ALLOW_THREADS
+	success = CloseHandle(hObject); 
+	Py_END_ALLOW_THREADS
+
+	if (!success)
+		return PyErr_SetFromWindowsErr(0);
+
+	Py_RETURN_NONE;
+}
+
+static PyObject *
+win32_ConnectNamedPipe(PyObject *self, PyObject *args)
+{
+	HANDLE hNamedPipe;
+	LPOVERLAPPED lpOverlapped;
+	BOOL success;
+
+	if (!PyArg_ParseTuple(args, F_HANDLE F_POINTER, 
+			 &hNamedPipe, &lpOverlapped))
+		return NULL;
+
+	Py_BEGIN_ALLOW_THREADS
+	success = ConnectNamedPipe(hNamedPipe, lpOverlapped);
+	Py_END_ALLOW_THREADS
+
+	if (!success)
+		return PyErr_SetFromWindowsErr(0);
+
+	Py_RETURN_NONE;
+}
+
+static PyObject *
+win32_CreateFile(PyObject *self, PyObject *args)
+{
+	LPCTSTR lpFileName;
+	DWORD dwDesiredAccess;
+	DWORD dwShareMode;
+	LPSECURITY_ATTRIBUTES lpSecurityAttributes;
+	DWORD dwCreationDisposition;
+	DWORD dwFlagsAndAttributes;
+	HANDLE hTemplateFile;
+	HANDLE handle;
+
+	if (!PyArg_ParseTuple(args, "s" F_DWORD F_DWORD F_POINTER 
+			 F_DWORD F_DWORD F_HANDLE,
+			 &lpFileName, &dwDesiredAccess, &dwShareMode, 
+			 &lpSecurityAttributes, &dwCreationDisposition, 
+			 &dwFlagsAndAttributes, &hTemplateFile))
+		return NULL;
+
+	Py_BEGIN_ALLOW_THREADS
+	handle = CreateFile(lpFileName, dwDesiredAccess, 
+			 dwShareMode, lpSecurityAttributes, 
+			 dwCreationDisposition, 
+			 dwFlagsAndAttributes, hTemplateFile);
+	Py_END_ALLOW_THREADS
+
+	if (handle == INVALID_HANDLE_VALUE)
+		return PyErr_SetFromWindowsErr(0);
+
+	return Py_BuildValue(F_HANDLE, handle);
+}
+
+static PyObject *
+win32_CreateNamedPipe(PyObject *self, PyObject *args)
+{
+	LPCTSTR lpName;
+	DWORD dwOpenMode;
+	DWORD dwPipeMode;
+	DWORD nMaxInstances;
+	DWORD nOutBufferSize;
+	DWORD nInBufferSize;
+	DWORD nDefaultTimeOut;
+	LPSECURITY_ATTRIBUTES lpSecurityAttributes;
+	HANDLE handle;
+
+	if (!PyArg_ParseTuple(args, "s" F_DWORD F_DWORD F_DWORD 
+			 F_DWORD F_DWORD F_DWORD F_POINTER,
+			 &lpName, &dwOpenMode, &dwPipeMode, 
+			 &nMaxInstances, &nOutBufferSize, 
+			 &nInBufferSize, &nDefaultTimeOut,
+			 &lpSecurityAttributes))
+		return NULL;
+
+	Py_BEGIN_ALLOW_THREADS
+	handle = CreateNamedPipe(lpName, dwOpenMode, dwPipeMode, 
+				 nMaxInstances, nOutBufferSize, 
+				 nInBufferSize, nDefaultTimeOut,
+				 lpSecurityAttributes);
+	Py_END_ALLOW_THREADS
+
+	if (handle == INVALID_HANDLE_VALUE)
+		return PyErr_SetFromWindowsErr(0);
+
+	return Py_BuildValue(F_HANDLE, handle);
+}
+
+static PyObject *
+win32_ExitProcess(PyObject *self, PyObject *args)
+{
+	UINT uExitCode;
+
+	if (!PyArg_ParseTuple(args, "I", &uExitCode))
+		return NULL;
+
+	ExitProcess(uExitCode);
+
+	return NULL;
+}
+
+static PyObject *
+win32_GetLastError(PyObject *self, PyObject *args)
+{
+	return Py_BuildValue(F_DWORD, GetLastError());
+}
+
+static PyObject *
+win32_OpenProcess(PyObject *self, PyObject *args)
+{
+	DWORD dwDesiredAccess;
+	BOOL bInheritHandle;
+	DWORD dwProcessId;
+	HANDLE handle;
+
+	if (!PyArg_ParseTuple(args, F_DWORD "i" F_DWORD, 
+			 &dwDesiredAccess, &bInheritHandle, &dwProcessId))
+		return NULL;
+
+	handle = OpenProcess(dwDesiredAccess, bInheritHandle, dwProcessId); 
+	if (handle == NULL)
+		return PyErr_SetFromWindowsErr(0);
+
+	return Py_BuildValue(F_HANDLE, handle);
+}
+
+static PyObject *
+win32_SetNamedPipeHandleState(PyObject *self, PyObject *args)
+{
+	HANDLE hNamedPipe;
+	PyObject *oArgs[3];
+	DWORD dwArgs[3], *pArgs[3] = {NULL, NULL, NULL};
+	int i;
+
+	if (!PyArg_ParseTuple(args, F_HANDLE "OOO", 
+			 &hNamedPipe, &oArgs[0], &oArgs[1], &oArgs[2]))
+		return NULL;
+
+	PyErr_Clear();
+
+	for (i = 0 ; i < 3 ; i++) {
+		if (oArgs[i] != Py_None) {
+			dwArgs[i] = PyInt_AsUnsignedLongMask(oArgs[i]);
+			if (PyErr_Occurred())
+				return NULL;
+			pArgs[i] = &dwArgs[i];
+		}
+	}
+
+	if (!SetNamedPipeHandleState(hNamedPipe, pArgs[0], pArgs[1], pArgs[2]))
+		return PyErr_SetFromWindowsErr(0);
+
+	Py_RETURN_NONE;
+}
+
+static PyObject *
+win32_WaitNamedPipe(PyObject *self, PyObject *args)
+{
+	LPCTSTR lpNamedPipeName;
+	DWORD nTimeOut;
+	BOOL success;
+
+	if (!PyArg_ParseTuple(args, "s" F_DWORD, &lpNamedPipeName, &nTimeOut))
+		return NULL;
+
+	Py_BEGIN_ALLOW_THREADS
+	success = WaitNamedPipe(lpNamedPipeName, nTimeOut);
+	Py_END_ALLOW_THREADS
+
+	if (!success)
+		return PyErr_SetFromWindowsErr(0);
+
+	Py_RETURN_NONE;
+}
+
+static PyMethodDef win32_methods[] = {
+	WIN32_FUNCTION(CloseHandle),
+	WIN32_FUNCTION(GetLastError),
+	WIN32_FUNCTION(OpenProcess),
+	WIN32_FUNCTION(ExitProcess),
+	WIN32_FUNCTION(ConnectNamedPipe),
+	WIN32_FUNCTION(CreateFile),
+	WIN32_FUNCTION(CreateNamedPipe),
+	WIN32_FUNCTION(SetNamedPipeHandleState),
+	WIN32_FUNCTION(WaitNamedPipe),
+	{NULL}
+};
+
+
+PyTypeObject Win32Type = {
+	PyVarObject_HEAD_INIT(NULL, 0)
+};
+
+
+PyObject *
+create_win32_namespace(void)
+{
+	Win32Type.tp_name = "_multiprocessing.win32";
+	Win32Type.tp_methods = win32_methods;
+	if (PyType_Ready(&Win32Type) < 0)
+		return NULL;
+	Py_INCREF(&Win32Type);
+
+	WIN32_CONSTANT(F_DWORD, ERROR_ALREADY_EXISTS);
+	WIN32_CONSTANT(F_DWORD, ERROR_PIPE_BUSY);
+	WIN32_CONSTANT(F_DWORD, ERROR_PIPE_CONNECTED);
+	WIN32_CONSTANT(F_DWORD, ERROR_SEM_TIMEOUT);
+	WIN32_CONSTANT(F_DWORD, GENERIC_READ);
+	WIN32_CONSTANT(F_DWORD, GENERIC_WRITE);
+	WIN32_CONSTANT(F_DWORD, INFINITE);
+	WIN32_CONSTANT(F_DWORD, NMPWAIT_WAIT_FOREVER);
+	WIN32_CONSTANT(F_DWORD, OPEN_EXISTING);
+	WIN32_CONSTANT(F_DWORD, PIPE_ACCESS_DUPLEX);
+	WIN32_CONSTANT(F_DWORD, PIPE_ACCESS_INBOUND);
+	WIN32_CONSTANT(F_DWORD, PIPE_READMODE_MESSAGE);
+	WIN32_CONSTANT(F_DWORD, PIPE_TYPE_MESSAGE);
+	WIN32_CONSTANT(F_DWORD, PIPE_UNLIMITED_INSTANCES);
+	WIN32_CONSTANT(F_DWORD, PIPE_WAIT);
+	WIN32_CONSTANT(F_DWORD, PROCESS_ALL_ACCESS);
+
+	WIN32_CONSTANT("i", NULL);
+
+	return (PyObject*)&Win32Type;
+}
