cpython: 9558e9360afc (original) (raw)

Mercurial > cpython

changeset 86401:9558e9360afc

Try doing a raw test of os.fork()/os.kill().

Richard Oudkerk shibturn@gmail.com
date	Thu, 17 Oct 2013 14:24:06 +0100
parents	9cd88b39ef62
children	5cc9cbfb5cff
files	Lib/test/_test_multiprocessing.py
diffstat	1 files changed, 36 insertions(+), 5 deletions(-)[+] [-] Lib/test/_test_multiprocessing.py 41

line wrap: on

line diff

--- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -273,10 +273,11 @@ class _TestProcess(BaseTestCase): @classmethod def _test_terminate(cls):

   print('signal.getsignal(SIGTERM) =', signal.getsignal(signal.SIGTERM))[](#l1.7)

```
   print('starting sleep')[](#l1.8)
```

   print('signal.getsignal(SIGTERM) =',[](#l1.9)

         signal.getsignal(signal.SIGTERM), file=sys.stderr)[](#l1.10)

   print('starting sleep', file=sys.stderr)[](#l1.11)
   time.sleep(100)[](#l1.12)

```
   print('finished sleep')[](#l1.13)
```

   print('finished sleep', file=sys.stderr)[](#l1.14)

def test_terminate(self): if self.TYPE == 'threads': @@ -314,11 +315,12 @@ class _TestProcess(BaseTestCase): try: signal.alarm(10) self.assertEqual(join(), None)

           signal.alarm(0)[](#l1.22)
       except RuntimeError:[](#l1.23)

           print('os.waitpid() =', os.waitpid(p.pid, os.WNOHANG))[](#l1.24)

           print('os.waitpid() =',[](#l1.25)

                 os.waitpid(p.pid, os.WNOHANG), file=sys.stderr)[](#l1.26)
           raise[](#l1.27)
       finally:[](#l1.28)

           signal.alarm(0)[](#l1.29)
           signal.signal(signal.SIGALRM, old_handler)[](#l1.30)
   else:[](#l1.31)
       self.assertEqual(join(), None)[](#l1.32)

@@ -333,6 +335,35 @@ class _TestProcess(BaseTestCase): # XXX sometimes get p.exitcode == 0 on Windows ... #self.assertEqual(p.exitcode, -signal.SIGTERM)

@unittest.skipIf(WIN32, 'Unix only')
def test_sigterm(self):

   # A test for the Gentoo build bot which does not directly use[](#l1.39)

   # multiprocessing.  Start and terminate child processes.[](#l1.40)

   if self.TYPE != 'processes':[](#l1.41)

```
       return[](#l1.42)
```
```
   for i in range(10):[](#l1.43)
```
```
       pid = os.fork()[](#l1.44)
```
```
       if pid == 0:[](#l1.45)
```
```
           try:[](#l1.46)
```

               print('sleeping', file=sys.stderr)[](#l1.47)

               time.sleep(100)[](#l1.48)

               print('waking', file=sys.stderr)[](#l1.49)

```
           finally:[](#l1.50)
```

               sys.stderr.flush()[](#l1.51)

```
               os._exit(0)[](#l1.52)
```
```
       else:[](#l1.53)
```

           os.kill(pid, signal.SIGTERM)[](#l1.54)

           def handler(*args):[](#l1.55)

               raise RuntimeError('join took too long: %s' % p)[](#l1.56)

           old_handler = signal.signal(signal.SIGALRM, handler)[](#l1.57)

```
           try:[](#l1.58)
```

               signal.alarm(10)[](#l1.59)

               pid_status = os.waitpid(pid, 0)[](#l1.60)

               self.assertEqual(pid_status[0], pid)[](#l1.61)

```
           finally:[](#l1.62)
```

               signal.alarm(0)[](#l1.63)

               signal.signal(signal.SIGALRM, old_handler)[](#l1.64)

+ def test_cpu_count(self): try: cpus = multiprocessing.cpu_count()