diff --git a/Lib/multiprocessing/__init__.py b/Lib/multiprocessing/__init__.py
index 1cb3222..decb2ad 100644
--- a/Lib/multiprocessing/__init__.py
+++ b/Lib/multiprocessing/__init__.py
@@ -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)
diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py
index 752d9ab..f5a3301 100644
--- a/Lib/multiprocessing/connection.py
+++ b/Lib/multiprocessing/connection.py
@@ -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)
diff --git a/Lib/multiprocessing/dummy/__init__.py b/Lib/multiprocessing/dummy/__init__.py
index cabf580..dd0f07b 100644
--- a/Lib/multiprocessing/dummy/__init__.py
+++ b/Lib/multiprocessing/dummy/__init__.py
@@ -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
diff --git a/Lib/multiprocessing/dummy/connection.py b/Lib/multiprocessing/dummy/connection.py
index dd2bcb9..4f0a680 100644
--- a/Lib/multiprocessing/dummy/connection.py
+++ b/Lib/multiprocessing/dummy/connection.py
@@ -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
diff --git a/Lib/multiprocessing/forking.py b/Lib/multiprocessing/forking.py
index 2c1d3cf..6107f07 100644
--- a/Lib/multiprocessing/forking.py
+++ b/Lib/multiprocessing/forking.py
@@ -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()
 
diff --git a/Lib/multiprocessing/heap.py b/Lib/multiprocessing/heap.py
index 7e596ca..f6b3404 100644
--- a/Lib/multiprocessing/heap.py
+++ b/Lib/multiprocessing/heap.py
@@ -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]
diff --git a/Lib/multiprocessing/managers.py b/Lib/multiprocessing/managers.py
index 908c193..6c1d912 100644
--- a/Lib/multiprocessing/managers.py
+++ b/Lib/multiprocessing/managers.py
@@ -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.
     '''
diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py
index 0255c86..79f0a29 100644
--- a/Lib/multiprocessing/pool.py
+++ b/Lib/multiprocessing/pool.py
@@ -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
diff --git a/Lib/multiprocessing/process.py b/Lib/multiprocessing/process.py
index 915d863..43d8297 100644
--- a/Lib/multiprocessing/process.py
+++ b/Lib/multiprocessing/process.py
@@ -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)
diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py
index 78cb362..ea89090 100644
--- a/Lib/multiprocessing/queues.py
+++ b/Lib/multiprocessing/queues.py
@@ -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:
diff --git a/Lib/multiprocessing/reduction.py b/Lib/multiprocessing/reduction.py
index 0d6cf4f..17778ef 100644
--- a/Lib/multiprocessing/reduction.py
+++ b/Lib/multiprocessing/reduction.py
@@ -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)
diff --git a/Lib/multiprocessing/sharedctypes.py b/Lib/multiprocessing/sharedctypes.py
index 6877249..808d312 100644
--- a/Lib/multiprocessing/sharedctypes.py
+++ b/Lib/multiprocessing/sharedctypes.py
@@ -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')
diff --git a/Lib/multiprocessing/synchronize.py b/Lib/multiprocessing/synchronize.py
index d642032..6a7189a 100644
--- a/Lib/multiprocessing/synchronize.py
+++ b/Lib/multiprocessing/synchronize.py
@@ -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:
diff --git a/Lib/multiprocessing/util.py b/Lib/multiprocessing/util.py
index d1b190c..25ff8bd 100644
--- a/Lib/multiprocessing/util.py
+++ b/Lib/multiprocessing/util.py
@@ -83,7 +83,7 @@
     import logging
     if hasattr(logging, 'multiprocessing'):
         return
-    
+
     logging._acquireLock()
     try:
         OldLoggerClass = logging.getLoggerClass()
diff --git a/Lib/test/test_multiprocessing.py b/Lib/test/test_multiprocessing.py
index d75fd20..4d3527c 100644
--- a/Lib/test/test_multiprocessing.py
+++ b/Lib/test/test_multiprocessing.py
@@ -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()