Merged revisions 73694,73708,73738 via svnmerge from

svn+ssh://pythondev@svn.python.org/python/trunk

........
  r73694 | jesse.noller | 2009-06-29 14:24:26 -0400 (Mon, 29 Jun 2009) | 1 line

  Issue 5740: multiprocessing.connection.* authkey fixes
........
  r73708 | jesse.noller | 2009-06-30 13:11:52 -0400 (Tue, 30 Jun 2009) | 1 line

  Resolves issues 5155, 5313, 5331 - bad file descriptor error with processes in processes
........
  r73738 | r.david.murray | 2009-06-30 22:49:10 -0400 (Tue, 30 Jun 2009) | 2 lines

  Make punctuation prettier and break up run-on sentence.
........

Lib/test/test_multiprocessing.py

@@ -8,6 +8,7 @@ import unittest
 import threading
 import queue as pyqueue
 import time
+import io
 import sys
 import os
 import gc
@@ -1868,7 +1869,74 @@ class TestInitializers(unittest.TestCase):
         p.join()
         self.assertEqual(self.ns.test, 1)
 
-testcases_other = [OtherTest, TestInvalidHandle, TestInitializers]
+#
+# Issue 5155, 5313, 5331: Test process in processes
+# Verifies os.close(sys.stdin.fileno) vs. sys.stdin.close() behavior
+#
+
+def _ThisSubProcess(q):
+    try:
+        item = q.get(block=False)
+    except pyqueue.Empty:
+        pass
+
+def _TestProcess(q):
+    queue = multiprocessing.Queue()
+    subProc = multiprocessing.Process(target=_ThisSubProcess, args=(queue,))
+    subProc.start()
+    subProc.join()
+
+def _afunc(x):
+    return x*x
+
+def pool_in_process():
+    pool = multiprocessing.Pool(processes=4)
+    x = pool.map(_afunc, [1, 2, 3, 4, 5, 6, 7])
+
+class _file_like(object):
+    def __init__(self, delegate):
+        self._delegate = delegate
+        self._pid = None
+
+    @property
+    def cache(self):
+        pid = os.getpid()
+        # There are no race conditions since fork keeps only the running thread
+        if pid != self._pid:
+            self._pid = pid
+            self._cache = []
+        return self._cache
+
+    def write(self, data):
+        self.cache.append(data)
+
+    def flush(self):
+        self._delegate.write(''.join(self.cache))
+        self._cache = []
+
+class TestStdinBadfiledescriptor(unittest.TestCase):
+
+    def test_queue_in_process(self):
+        queue = multiprocessing.Queue()
+        proc = multiprocessing.Process(target=_TestProcess, args=(queue,))
+        proc.start()
+        proc.join()
+
+    def test_pool_in_process(self):
+        p = multiprocessing.Process(target=pool_in_process)
+        p.start()
+        p.join()
+
+    def test_flushing(self):
+        sio = io.StringIO()
+        flike = _file_like(sio)
+        flike.write('foo')
+        proc = multiprocessing.Process(target=lambda: flike.flush())
+        flike.flush()
+        assert sio.getvalue() == 'foo'
+
+testcases_other = [OtherTest, TestInvalidHandle, TestInitializers,
+                   TestStdinBadfiledescriptor]
 
 #
 #