Lib/test/test_fork1.py

   1 """This test checks for correct fork() behavior.
   2 """
   3
   4 import errno
   5 import imp
   6 import os
   7 import signal
   8 import sys
   9 import time
  10 import threading
  11
  12 from test.fork_wait import ForkWait
  13 from test.test_support import run_unittest, reap_children, get_attribute
  14
  15 #Skip test if fork does not exist.
  16 get_attribute(os, 'fork')
  17
  18
  19 class ForkTest(ForkWait):
  20     def wait_impl(self, cpid):
  21         for i in range(10):
  22             # waitpid() shouldn't hang, but some of the buildbots seem to hang
  23             # in the forking tests.  This is an attempt to fix the problem.
  24             spid, status = os.waitpid(cpid, os.WNOHANG)
  25             if spid == cpid:
  26                 break
  27             time.sleep(1.0)
  28
  29         self.assertEqual(spid, cpid)
  30         self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
  31
  32     def test_import_lock_fork(self):
  33         import_started = threading.Event()
  34         fake_module_name = "fake test module"
  35         partial_module = "partial"
  36         complete_module = "complete"
  37         def importer():
  38             imp.acquire_lock()
  39             sys.modules[fake_module_name] = partial_module
  40             import_started.set()
  41             time.sleep(0.01) # Give the other thread time to try and acquire.
  42             sys.modules[fake_module_name] = complete_module
  43             imp.release_lock()
  44         t = threading.Thread(target=importer)
  45         t.start()
  46         import_started.wait()
  47         pid = os.fork()
  48         try:
  49             if not pid:
  50                 m = __import__(fake_module_name)
  51                 if m == complete_module:
  52                     os._exit(0)
  53                 else:
  54                     os._exit(1)
  55             else:
  56                 t.join()
  57                 # Exitcode 1 means the child got a partial module (bad.) No
  58                 # exitcode (but a hang, which manifests as 'got pid 0')
  59                 # means the child deadlocked (also bad.)
  60                 self.wait_impl(pid)
  61         finally:
  62             try:
  63                 os.kill(pid, signal.SIGKILL)
  64             except OSError:
  65                 pass
  66
  67 def test_main():
  68     run_unittest(ForkTest)
  69     reap_children()
  70
  71 if __name__ == "__main__":
  72     test_main()