cpython: c9facd251725 (original) (raw)

Mercurial > cpython

changeset 74093:c9facd251725 2.7

Followup to issue #11867: Use socketpair(), since FreeBSD < 8 doesn't really support multiprocessing.Event. [#11867]

Charles-François Natali neologix@free.fr
date	Tue, 20 Dec 2011 11:47:23 +0100
parents	355466216029
children	5e939912f9f8
files	Lib/test/test_mailbox.py
diffstat	1 files changed, 8 insertions(+), 11 deletions(-)[+] [-] Lib/test/test_mailbox.py 19

line wrap: on

line diff

--- a/Lib/test/test_mailbox.py +++ b/Lib/test/test_mailbox.py @@ -15,10 +15,6 @@ try: 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 @@ class _TestMboxMMDF(TestMailbox): 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()[](#l1.23)

   done = multiprocessing.Event()[](#l1.24)

```
   c, p = socket.socketpair()[](#l1.25)
```
```
   self.addCleanup(c.close)[](#l1.26)
```
```
   self.addCleanup(p.close)[](#l1.27)
```

pid = os.fork() if pid == 0: @@ -883,22 +880,22 @@ class _TestMboxMMDF(TestMailbox): try: # lock the mailbox, and signal the parent it can proceed self._box.lock()

```
           ready.set()[](#l1.35)
```

```
           c.send(b'c')[](#l1.36)
```

# wait until the parent is done, and unlock the mailbox

```
           done.wait(5)[](#l1.39)
```

           c.recv(1)[](#l1.40)
           self._box.unlock()[](#l1.41)
       finally:[](#l1.42)
           os._exit(0)[](#l1.43)

# In the parent, wait until the child signals it locked the mailbox.

```
   ready.wait(5)[](#l1.46)
```

   p.recv(1)[](#l1.47)
   try:[](#l1.48)
       self.assertRaises(mailbox.ExternalClashError,[](#l1.49)
                         self._box.lock)[](#l1.50)
   finally:[](#l1.51)
       # Signal the child it can now release the lock and exit.[](#l1.52)

```
       done.set()[](#l1.53)
```

       p.send(b'p')[](#l1.54)
       # Wait for child to exit.  Locking should now succeed.[](#l1.55)
       exited_pid, status = os.waitpid(pid, 0)[](#l1.56)