
import os, sys, time, thread

LONGSLEEP = 2

SHORTSLEEP = 0.5

alive = {}

def f(id):
    while 1:
        alive[id] = os.getpid()
        try:
            time.sleep(SHORTSLEEP)
        except IOError:
            pass

def main():
    for i in range(4):
        thread.start_new(f, (i,))

    time.sleep(LONGSLEEP)

    a = alive.keys()
    a.sort()
    assert a == range(4)

    cpid = os.fork()

    if cpid == 0:
        # Child
        time.sleep(LONGSLEEP)
        n = 0
        pid = os.getpid()
        for key in alive.keys():
            if alive[key] == pid:
                n = n+1
        os._exit(n)
    else:
        # Parent
        spid, status = os.waitpid(cpid, 0)
        assert spid == cpid
        assert status == 0, "cause = %d, exit = %d" % (status&0xff, status>>8)

main()


import os, sys, time, thread

LONGSLEEP = 2

SHORTSLEEP = 0.5

NUM_THREADS = 4

alive = {}

def f(id):
    while 1:
        alive[id] = os.getpid()
        try:
            time.sleep(SHORTSLEEP)
        except IOError:
            pass

def main():
    for i in range(NUM_THREADS):
        thread.start_new(f, (i,))

    time.sleep(LONGSLEEP)

    a = alive.keys()
    a.sort()
    assert a == range(NUM_THREADS)

    prefork_lives = alive.copy()

    cpid = os.fork()

    if cpid == 0:
        # Child
        time.sleep(LONGSLEEP)
        n = 0
        for key in alive.keys():
            if alive[key] != prefork_lives[key]:
                n = n+1
        os._exit(n)
    else:
        # Parent
        spid, status = os.waitpid(cpid, 0)
        assert spid == cpid
        assert status == 0, "cause = %d, exit = %d" % (status&0xff, status>>8)

main()

Use a constant to specify the number of child threads to create.
Use a constant to specify the number of child threads to create.

Instead of assuming that the number process ids of the threads is the
same as the process id of the controlling process, use a copy of the
dictionary and check for changes in the process ids of the threads
from the thread's process ids in the parent process.  This makes the
test make more sense on systems which assign a new pid to each thread
(i.e., Linux).

This doesn't fix the other problems evident with this test on Linux.

  
  import os, sys, time, thread
  
  LONGSLEEP = 2
  
  SHORTSLEEP = 0.5
+ 
+ NUM_THREADS = 4
  
  alive = {}
  
  def f(id):
      while 1:
          alive[id] = os.getpid()
          try:
              time.sleep(SHORTSLEEP)
          except IOError:
              pass
  
  def main():
-     for i in range(4):
+     for i in range(NUM_THREADS):
          thread.start_new(f, (i,))
  
      time.sleep(LONGSLEEP)
  
      a = alive.keys()
      a.sort()
-     assert a == range(4)
+     assert a == range(NUM_THREADS)
+ 
+     prefork_lives = alive.copy()
  
      cpid = os.fork()
  
      if cpid == 0:
          # Child
          time.sleep(LONGSLEEP)
          n = 0
-         pid = os.getpid()
          for key in alive.keys():
-             if alive[key] == pid:
+             if alive[key] != prefork_lives[key]:
                  n = n+1
          os._exit(n)
      else:
          # Parent
          spid, status = os.waitpid(cpid, 0)
          assert spid == cpid
          assert status == 0, "cause = %d, exit = %d" % (status&0xff, status>>8)
  
  main()
  
Use a constant to specify the number of child threads to create.
## Code Before:

import os, sys, time, thread

LONGSLEEP = 2

SHORTSLEEP = 0.5

alive = {}

def f(id):
    while 1:
        alive[id] = os.getpid()
        try:
            time.sleep(SHORTSLEEP)
        except IOError:
            pass

def main():
    for i in range(4):
        thread.start_new(f, (i,))

    time.sleep(LONGSLEEP)

    a = alive.keys()
    a.sort()
    assert a == range(4)

    cpid = os.fork()

    if cpid == 0:
        # Child
        time.sleep(LONGSLEEP)
        n = 0
        pid = os.getpid()
        for key in alive.keys():
            if alive[key] == pid:
                n = n+1
        os._exit(n)
    else:
        # Parent
        spid, status = os.waitpid(cpid, 0)
        assert spid == cpid
        assert status == 0, "cause = %d, exit = %d" % (status&0xff, status>>8)

main()

## Instruction:
Use a constant to specify the number of child threads to create.
## Code After:

import os, sys, time, thread

LONGSLEEP = 2

SHORTSLEEP = 0.5

NUM_THREADS = 4

alive = {}

def f(id):
    while 1:
        alive[id] = os.getpid()
        try:
            time.sleep(SHORTSLEEP)
        except IOError:
            pass

def main():
    for i in range(NUM_THREADS):
        thread.start_new(f, (i,))

    time.sleep(LONGSLEEP)

    a = alive.keys()
    a.sort()
    assert a == range(NUM_THREADS)

    prefork_lives = alive.copy()

    cpid = os.fork()

    if cpid == 0:
        # Child
        time.sleep(LONGSLEEP)
        n = 0
        for key in alive.keys():
            if alive[key] != prefork_lives[key]:
                n = n+1
        os._exit(n)
    else:
        # Parent
        spid, status = os.waitpid(cpid, 0)
        assert spid == cpid
        assert status == 0, "cause = %d, exit = %d" % (status&0xff, status>>8)

main()
