[Python-checkins] cpython (2.7): Followup to issue #11867: Use socketpair(), since FreeBSD < 8 doesn't really

charles-francois.natali python-checkins at python.org
Tue Dec 20 11:50:02 CET 2011


http://hg.python.org/cpython/rev/c9facd251725
changeset:   74093:c9facd251725
branch:      2.7
parent:      74089:355466216029
user:        Charles-François Natali <neologix at free.fr>
date:        Tue Dec 20 11:47:23 2011 +0100
summary:
  Followup to issue #11867: Use socketpair(), since FreeBSD < 8 doesn't really
support multiprocessing.Event.

files:
  Lib/test/test_mailbox.py |  19 ++++++++-----------
  1 files changed, 8 insertions(+), 11 deletions(-)


diff --git a/Lib/test/test_mailbox.py b/Lib/test/test_mailbox.py
--- a/Lib/test/test_mailbox.py
+++ b/Lib/test/test_mailbox.py
@@ -15,10 +15,6 @@
     import fcntl
 except ImportError:
     pass
-try:
-    import multiprocessing
-except ImportError:
-    multiprocessing = None
 
 # Silence Py3k warning
 rfc822 = test_support.import_module('rfc822', deprecated=True)
@@ -870,12 +866,13 @@
         self._box = self._factory(self._path)
 
     @unittest.skipUnless(hasattr(os, 'fork'), "Test needs fork().")
-    @unittest.skipUnless(multiprocessing, "Test needs multiprocessing.")
+    @unittest.skipUnless(hasattr(socket, 'socketpair'), "Test needs socketpair().")
     def test_lock_conflict(self):
         # Fork off a child process that will lock the mailbox temporarily,
         # unlock it and exit.
-        ready = multiprocessing.Event()
-        done = multiprocessing.Event()
+        c, p = socket.socketpair()
+        self.addCleanup(c.close)
+        self.addCleanup(p.close)
 
         pid = os.fork()
         if pid == 0:
@@ -883,22 +880,22 @@
             try:
                 # lock the mailbox, and signal the parent it can proceed
                 self._box.lock()
-                ready.set()
+                c.send(b'c')
 
                 # wait until the parent is done, and unlock the mailbox
-                done.wait(5)
+                c.recv(1)
                 self._box.unlock()
             finally:
                 os._exit(0)
 
         # In the parent, wait until the child signals it locked the mailbox.
-        ready.wait(5)
+        p.recv(1)
         try:
             self.assertRaises(mailbox.ExternalClashError,
                               self._box.lock)
         finally:
             # Signal the child it can now release the lock and exit.
-            done.set()
+            p.send(b'p')
             # Wait for child to exit.  Locking should now succeed.
             exited_pid, status = os.waitpid(pid, 0)
 

-- 
Repository URL: http://hg.python.org/cpython


More information about the Python-checkins mailing list