#

# Unit tests for the multiprocessing package

#

import unittest

import unittest.mock

import queue as pyqueue

import textwrap

import time

import io

import itertools

import sys

import os

import gc

import importlib

import errno

import functools

import signal

import array

import collections.abc

import socket

import random

import logging

import shutil

import subprocess

import struct

import tempfile

import operator

import pickle

import weakref

import warnings

import test.support

import test.support.script_helper

from test import support

from test.support import hashlib_helper

from test.support import import_helper

from test.support import os_helper

from test.support import script_helper

from test.support import socket_helper

from test.support import threading_helper

from test.support import warnings_helper

from test.support import subTests

from test.support.script_helper import assert_python_failure, assert_python_ok

# Skip tests if _multiprocessing wasn't built.

_multiprocessing = import_helper.import_module('_multiprocessing')

# Skip tests if sem_open implementation is broken.

support.skip_if_broken_multiprocessing_synchronize()

import threading

import multiprocessing.connection

import multiprocessing.dummy

import multiprocessing.heap

import multiprocessing.managers

import multiprocessing.pool

import multiprocessing.queues

from multiprocessing.connection import wait

from multiprocessing import util

try:

from multiprocessing import reduction

HAS_REDUCTION = reduction.HAVE_SEND_HANDLE

except ImportError:

HAS_REDUCTION = False

try:

from multiprocessing.sharedctypes import Value, copy

HAS_SHAREDCTYPES = True

except ImportError:

HAS_SHAREDCTYPES = False

try:

from multiprocessing import shared_memory

HAS_SHMEM = True

except ImportError:

HAS_SHMEM = False

try:

import msvcrt

except ImportError:

msvcrt = None

if support.HAVE_ASAN_FORK_BUG:

# gh-89363: Skip multiprocessing tests if Python is built with ASAN to

# work around a libasan race condition: dead lock in pthread_create().

raise unittest.SkipTest("libasan has a pthread_create() dead lock related to thread+fork")

# gh-110666: Tolerate a difference of 100 ms when comparing timings

# (clock resolution)

CLOCK_RES = 0.100

def latin(s):

return s.encode('latin')

def close_queue(queue):

if isinstance(queue, multiprocessing.queues.Queue):

queue.close()

queue.join_thread()

def join_process(process):

# Since multiprocessing.Process has the same API than threading.Thread

# (join() and is_alive(), the support function can be reused

threading_helper.join_thread(process)

if os.name == "posix":

from multiprocessing import resource_tracker

def _resource_unlink(name, rtype):

resource_tracker._CLEANUP_FUNCS[rtype](name)

#

# Constants

#

LOG_LEVEL = util.SUBWARNING

#LOG_LEVEL = logging.DEBUG

DELTA = 0.1

CHECK_TIMINGS = False # making true makes tests take a lot longer

# and can sometimes cause some non-serious

# failures because some calls block a bit

# longer than expected

if CHECK_TIMINGS:

TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.82, 0.35, 1.4

else:

TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.1, 0.1, 0.1

# BaseManager.shutdown_timeout

SHUTDOWN_TIMEOUT = support.SHORT_TIMEOUT

WAIT_ACTIVE_CHILDREN_TIMEOUT = 5.0

HAVE_GETVALUE = not getattr(_multiprocessing,

'HAVE_BROKEN_SEM_GETVALUE', False)

WIN32 = (sys.platform == "win32")

def wait_for_handle(handle, timeout):

if timeout is not None and timeout < 0.0:

timeout = None

return wait([handle], timeout)

try:

MAXFD = os.sysconf("SC_OPEN_MAX")

except:

MAXFD = 256

# To speed up tests when using the forkserver, we can preload these:

PRELOAD = ['__main__', 'test.test_multiprocessing_forkserver']

#

# Some tests require ctypes

#

try:

from ctypes import Structure, c_int, c_double, c_longlong

except ImportError:

Structure = object

c_int = c_double = c_longlong = None

def check_enough_semaphores():

"""Check that the system supports enough semaphores to run the test."""

# minimum number of semaphores available according to POSIX

nsems_min = 256

try:

nsems = os.sysconf("SC_SEM_NSEMS_MAX")

except (AttributeError, ValueError):

# sysconf not available or setting not available

return

if nsems == -1 or nsems >= nsems_min:

return

raise unittest.SkipTest("The OS doesn't support enough semaphores "

"to run the test (required: %d)." % nsems_min)

def only_run_in_spawn_testsuite(reason):

"""Returns a decorator: raises SkipTest when SM != spawn at test time.

This can be useful to save overall Python test suite execution time.

"spawn" is the universal mode available on all platforms so this limits the

decorated test to only execute within test_multiprocessing_spawn.

This would not be necessary if we refactored our test suite to split things

into other test files when they are not start method specific to be rerun

under all start methods.

"""

def decorator(test_item):

@functools.wraps(test_item)

def spawn_check_wrapper(*args, **kwargs):

if (start_method := multiprocessing.get_start_method()) != "spawn":

raise unittest.SkipTest(f"{start_method=}, not 'spawn'; {reason}")

return test_item(*args, **kwargs)

return spawn_check_wrapper

return decorator

def only_run_in_forkserver_testsuite(reason):

"""Returns a decorator: raises SkipTest unless fork is supported

and the current start method is forkserver.

Combines @support.requires_fork() with the single-run semantics of

only_run_in_spawn_testsuite(), but uses the forkserver testsuite as

the single-run target. Appropriate for tests that exercise

os.fork() directly (raw fork or mp.set_start_method("fork") in a

subprocess) and don't vary by start method, since forkserver is

only available on platforms that support fork.

"""

def decorator(test_item):

@functools.wraps(test_item)

def forkserver_check_wrapper(*args, **kwargs):

if not support.has_fork_support:

raise unittest.SkipTest("requires working os.fork()")

if (start_method := multiprocessing.get_start_method()) != "forkserver":

raise unittest.SkipTest(

f"{start_method=}, not 'forkserver'; {reason}")

return test_item(*args, **kwargs)

return forkserver_check_wrapper

return decorator

class TestInternalDecorators(unittest.TestCase):

"""Logic within a test suite that could errantly skip tests? Test it!"""

@support.requires_fork()

def test_only_run_in_spawn_testsuite(self):

if multiprocessing.get_start_method() != "spawn":

raise unittest.SkipTest("only run in test_multiprocessing_spawn.")

try:

@only_run_in_spawn_testsuite("testing this decorator")

def return_four_if_spawn():

return 4

except Exception as err:

self.fail(f"expected decorated `def` not to raise; caught {err}")

orig_start_method = multiprocessing.get_start_method(allow_none=True)

try:

multiprocessing.set_start_method("spawn", force=True)

self.assertEqual(return_four_if_spawn(), 4)

multiprocessing.set_start_method("fork", force=True)

with self.assertRaises(unittest.SkipTest) as ctx:

return_four_if_spawn()

self.assertIn("testing this decorator", str(ctx.exception))

self.assertIn("start_method=", str(ctx.exception))

finally:

multiprocessing.set_start_method(orig_start_method, force=True)

@support.requires_fork()

def test_only_run_in_forkserver_testsuite(self):

if multiprocessing.get_start_method() != "forkserver":

raise unittest.SkipTest("only run in test_multiprocessing_forkserver.")

try:

@only_run_in_forkserver_testsuite("testing this decorator")

def return_four_if_forkserver():

return 4

except Exception as err:

self.fail(f"expected decorated `def` not to raise; caught {err}")

orig_start_method = multiprocessing.get_start_method(allow_none=True)

try:

multiprocessing.set_start_method("forkserver", force=True)

self.assertEqual(return_four_if_forkserver(), 4)

multiprocessing.set_start_method("spawn", force=True)

with self.assertRaises(unittest.SkipTest) as ctx:

return_four_if_forkserver()

self.assertIn("testing this decorator", str(ctx.exception))

self.assertIn("start_method=", str(ctx.exception))

finally:

multiprocessing.set_start_method(orig_start_method, force=True)

#

# Creates a wrapper for a function which records the time it takes to finish

#

class TimingWrapper(object):

def __init__(self, func):

self.func = func

self.elapsed = None

def __call__(self, *args, **kwds):

t = time.monotonic()

try:

return self.func(*args, **kwds)

finally:

self.elapsed = time.monotonic() - t

#

# Base class for test cases

#

class BaseTestCase(object):

ALLOWED_TYPES = ('processes', 'manager', 'threads')

# If not empty, limit which start method suites run this class.

START_METHODS: set[str] = set()

start_method = None # set by install_tests_in_module_dict()

def assertTimingAlmostEqual(self, a, b):

if CHECK_TIMINGS:

self.assertAlmostEqual(a, b, 1)

def assertReturnsIfImplemented(self, value, func, *args):

try:

res = func(*args)

except NotImplementedError:

pass

else:

return self.assertEqual(value, res)

# For the sanity of Windows users, rather than crashing or freezing in

# multiple ways.

def __reduce__(self, *args):

raise NotImplementedError("shouldn't try to pickle a test case")

__reduce_ex__ = __reduce__

#

# Return the value of a semaphore

#

def get_value(self):

try:

return self.get_value()

except AttributeError:

try:

return self._Semaphore__value

except AttributeError:

try:

return self._value

except AttributeError:

raise NotImplementedError

#

# Testcases

#

class DummyCallable:

def __call__(self, q, c):

assert isinstance(c, DummyCallable)

q.put(5)

class _TestProcess(BaseTestCase):

ALLOWED_TYPES = ('processes', 'threads')

def test_current(self):

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

current = self.current_process()

authkey = current.authkey

self.assertTrue(current.is_alive())

self.assertFalse(current.daemon)

self.assertIsInstance(authkey, bytes)

self.assertTrue(len(authkey) > 0)

self.assertEqual(current.ident, os.getpid())

self.assertEqual(current.exitcode, None)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_set_executable(self):

if self.TYPE == 'threads':

self.skipTest(f'test not appropriate for {self.TYPE}')

paths = [

sys.executable, # str

os.fsencode(sys.executable), # bytes

os_helper.FakePath(sys.executable), # os.PathLike

os_helper.FakePath(os.fsencode(sys.executable)), # os.PathLike bytes

]

for path in paths:

self.set_executable(path)

p = self.Process()

p.start()

p.join()

self.assertEqual(p.exitcode, 0)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

@support.requires_resource('cpu')

def test_args_argument(self):

# bpo-45735: Using list or tuple as *args* in constructor could

# achieve the same effect.

args_cases = (1, "str", [1], (1,))

args_types = (list, tuple)

test_cases = itertools.product(args_cases, args_types)

for args, args_type in test_cases:

with self.subTest(args=args, args_type=args_type):

q = self.Queue(1)

# pass a tuple or list as args

p = self.Process(target=self._test_args, args=args_type((q, args)))

p.daemon = True

p.start()

child_args = q.get()

self.assertEqual(child_args, args)

p.join()

close_queue(q)

@classmethod

def _test_args(cls, q, arg):

q.put(arg)

def test_daemon_argument(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

# By default uses the current process's daemon flag.

proc0 = self.Process(target=self._test)

self.assertEqual(proc0.daemon, self.current_process().daemon)

proc1 = self.Process(target=self._test, daemon=True)

self.assertTrue(proc1.daemon)

proc2 = self.Process(target=self._test, daemon=False)

self.assertFalse(proc2.daemon)

@classmethod

def _test(cls, q, *args, **kwds):

current = cls.current_process()

q.put(args)

q.put(kwds)

q.put(current.name)

if cls.TYPE != 'threads':

q.put(bytes(current.authkey))

q.put(current.pid)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_parent_process_attributes(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

self.assertIsNone(self.parent_process())

rconn, wconn = self.Pipe(duplex=False)

p = self.Process(target=self._test_send_parent_process, args=(wconn,))

p.start()

p.join()

parent_pid, parent_name = rconn.recv()

self.assertEqual(parent_pid, self.current_process().pid)

self.assertEqual(parent_pid, os.getpid())

self.assertEqual(parent_name, self.current_process().name)

@classmethod

def _test_send_parent_process(cls, wconn):

from multiprocessing.process import parent_process

wconn.send([parent_process().pid, parent_process().name])

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_parent_process(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

# Launch a child process. Make it launch a grandchild process. Kill the

# child process and make sure that the grandchild notices the death of

# its parent (a.k.a the child process).

rconn, wconn = self.Pipe(duplex=False)

p = self.Process(

target=self._test_create_grandchild_process, args=(wconn, ))

p.start()

if not rconn.poll(timeout=support.LONG_TIMEOUT):

raise AssertionError("Could not communicate with child process")

parent_process_status = rconn.recv()

self.assertEqual(parent_process_status, "alive")

p.terminate()

p.join()

if not rconn.poll(timeout=support.LONG_TIMEOUT):

raise AssertionError("Could not communicate with child process")

parent_process_status = rconn.recv()

self.assertEqual(parent_process_status, "not alive")

@classmethod

def _test_create_grandchild_process(cls, wconn):

p = cls.Process(target=cls._test_report_parent_status, args=(wconn, ))

p.start()

time.sleep(300)

@classmethod

def _test_report_parent_status(cls, wconn):

from multiprocessing.process import parent_process

wconn.send("alive" if parent_process().is_alive() else "not alive")

parent_process().join(timeout=support.SHORT_TIMEOUT)

wconn.send("alive" if parent_process().is_alive() else "not alive")

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_process(self):

q = self.Queue(1)

e = self.Event()

args = (q, 1, 2)

kwargs = {'hello':23, 'bye':2.54}

name = 'SomeProcess'

p = self.Process(

target=self._test, args=args, kwargs=kwargs, name=name

)

p.daemon = True

current = self.current_process()

if self.TYPE != 'threads':

self.assertEqual(p.authkey, current.authkey)

self.assertEqual(p.is_alive(), False)

self.assertEqual(p.daemon, True)

self.assertNotIn(p, self.active_children())

self.assertIs(type(self.active_children()), list)

self.assertEqual(p.exitcode, None)

p.start()

self.assertEqual(p.exitcode, None)

self.assertEqual(p.is_alive(), True)

self.assertIn(p, self.active_children())

self.assertEqual(q.get(), args[1:])

self.assertEqual(q.get(), kwargs)

self.assertEqual(q.get(), p.name)

if self.TYPE != 'threads':

self.assertEqual(q.get(), current.authkey)

self.assertEqual(q.get(), p.pid)

p.join()

self.assertEqual(p.exitcode, 0)

self.assertEqual(p.is_alive(), False)

self.assertNotIn(p, self.active_children())

close_queue(q)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

@unittest.skipUnless(threading._HAVE_THREAD_NATIVE_ID, "needs native_id")

def test_process_mainthread_native_id(self):

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

current_mainthread_native_id = threading.main_thread().native_id

q = self.Queue(1)

p = self.Process(target=self._test_process_mainthread_native_id, args=(q,))

p.start()

child_mainthread_native_id = q.get()

p.join()

close_queue(q)

self.assertNotEqual(current_mainthread_native_id, child_mainthread_native_id)

@classmethod

def _test_process_mainthread_native_id(cls, q):

mainthread_native_id = threading.main_thread().native_id

q.put(mainthread_native_id)

@classmethod

def _sleep_some(cls):

time.sleep(100)

@classmethod

def _sleep_some_event(cls, event):

event.set()

time.sleep(100)

@classmethod

def _sleep_no_int_handler(cls, event):

signal.signal(signal.SIGINT, signal.SIG_DFL)

cls._sleep_some_event(event)

@classmethod

def _test_sleep(cls, delay):

time.sleep(delay)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def _kill_process(self, meth, target=None):

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

event = self.Event()

if not target:

target = self._sleep_some_event

p = self.Process(target=target, args=(event,))

p.daemon = True

p.start()

self.assertEqual(p.is_alive(), True)

self.assertIn(p, self.active_children())

self.assertEqual(p.exitcode, None)

join = TimingWrapper(p.join)

self.assertEqual(join(0), None)

self.assertTimingAlmostEqual(join.elapsed, 0.0)

self.assertEqual(p.is_alive(), True)

self.assertEqual(join(-1), None)

self.assertTimingAlmostEqual(join.elapsed, 0.0)

self.assertEqual(p.is_alive(), True)

timeout = support.SHORT_TIMEOUT

if not event.wait(timeout):

p.terminate()

p.join()

self.fail(f"event not signaled in {timeout} seconds")

meth(p)

if hasattr(signal, 'alarm'):

# On the Gentoo buildbot waitpid() often seems to block forever.

# We use alarm() to interrupt it if it blocks for too long.

def handler(*args):

raise RuntimeError('join took too long: %s' % p)

old_handler = signal.signal(signal.SIGALRM, handler)

try:

signal.alarm(10)

self.assertEqual(join(), None)

finally:

signal.alarm(0)

signal.signal(signal.SIGALRM, old_handler)

else:

self.assertEqual(join(), None)

self.assertTimingAlmostEqual(join.elapsed, 0.0)

self.assertEqual(p.is_alive(), False)

self.assertNotIn(p, self.active_children())

p.join()

return p.exitcode

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

@unittest.skipIf(os.name == 'nt', "POSIX only")

def test_interrupt(self):

exitcode = self._kill_process(multiprocessing.Process.interrupt)

self.assertEqual(exitcode, 1)

# exit code 1 is hard-coded for uncaught exceptions

# (KeyboardInterrupt in this case)

# in multiprocessing.BaseProcess._bootstrap

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

@unittest.skipIf(os.name == 'nt', "POSIX only")

def test_interrupt_no_handler(self):

exitcode = self._kill_process(multiprocessing.Process.interrupt, target=self._sleep_no_int_handler)

self.assertEqual(exitcode, -signal.SIGINT)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_terminate(self):

exitcode = self._kill_process(multiprocessing.Process.terminate)

self.assertEqual(exitcode, -signal.SIGTERM)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_kill(self):

exitcode = self._kill_process(multiprocessing.Process.kill)

if os.name != 'nt':

self.assertEqual(exitcode, -signal.SIGKILL)

else:

self.assertEqual(exitcode, -signal.SIGTERM)

def test_cpu_count(self):

try:

cpus = multiprocessing.cpu_count()

except NotImplementedError:

cpus = 1

self.assertIsInstance(cpus, int)

self.assertGreaterEqual(cpus, 1)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_active_children(self):

self.assertEqual(type(self.active_children()), list)

event = self.Event()

p = self.Process(target=event.wait, args=())

self.assertNotIn(p, self.active_children())

try:

p.daemon = True

p.start()

self.assertIn(p, self.active_children())

finally:

event.set()

p.join()

self.assertNotIn(p, self.active_children())

@classmethod

def _test_recursion(cls, wconn, id):

wconn.send(id)

if len(id) < 2:

for i in range(2):

p = cls.Process(

target=cls._test_recursion, args=(wconn, id+[i])

)

p.start()

p.join()

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_recursion(self):

rconn, wconn = self.Pipe(duplex=False)

self._test_recursion(wconn, [])

time.sleep(DELTA)

result = []

while rconn.poll():

result.append(rconn.recv())

expected = [

[],

[0],

[0, 0],

[0, 1],

[1],

[1, 0],

[1, 1]

]

self.assertEqual(result, expected)

@classmethod

def _test_sentinel(cls, event):

event.wait(10.0)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_sentinel(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

event = self.Event()

p = self.Process(target=self._test_sentinel, args=(event,))

with self.assertRaises(ValueError):

p.sentinel

p.start()

self.addCleanup(p.join)

sentinel = p.sentinel

self.assertIsInstance(sentinel, int)

self.assertFalse(wait_for_handle(sentinel, timeout=0.0))

event.set()

p.join()

self.assertTrue(wait_for_handle(sentinel, timeout=1))

@classmethod

def _test_close(cls, rc=0, q=None):

if q is not None:

q.get()

sys.exit(rc)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_close(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

q = self.Queue()

p = self.Process(target=self._test_close, kwargs={'q': q})

p.daemon = True

p.start()

self.assertEqual(p.is_alive(), True)

# Child is still alive, cannot close

with self.assertRaises(ValueError):

p.close()

q.put(None)

p.join()

self.assertEqual(p.is_alive(), False)

self.assertEqual(p.exitcode, 0)

p.close()

with self.assertRaises(ValueError):

p.is_alive()

with self.assertRaises(ValueError):

p.join()

with self.assertRaises(ValueError):

p.terminate()

p.close()

wr = weakref.ref(p)

del p

gc.collect()

self.assertIs(wr(), None)

close_queue(q)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

@support.requires_resource('walltime')

def test_many_processes(self):

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm = multiprocessing.get_start_method()

N = 5 if sm == 'spawn' else 100

# Try to overwhelm the forkserver loop with events

procs = [self.Process(target=self._test_sleep, args=(0.01,))

for i in range(N)]

for p in procs:

p.start()

for p in procs:

join_process(p)

for p in procs:

self.assertEqual(p.exitcode, 0)

procs = [self.Process(target=self._sleep_some)

for i in range(N)]

for p in procs:

p.start()

time.sleep(0.001) # let the children start...

for p in procs:

p.terminate()

for p in procs:

join_process(p)

if os.name != 'nt':

exitcodes = [-signal.SIGTERM]

if sys.platform == 'darwin':

# bpo-31510: On macOS, killing a freshly started process with

# SIGTERM sometimes kills the process with SIGKILL.

exitcodes.append(-signal.SIGKILL)

for p in procs:

self.assertIn(p.exitcode, exitcodes)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_lose_target_ref(self):

c = DummyCallable()

wr = weakref.ref(c)

q = self.Queue()

p = self.Process(target=c, args=(q, c))

del c

p.start()

p.join()

gc.collect() # For PyPy or other GCs.

self.assertIs(wr(), None)

self.assertEqual(q.get(), 5)

close_queue(q)

@classmethod

def _test_child_fd_inflation(self, evt, q):

q.put(os_helper.fd_count())

evt.wait()

def test_child_fd_inflation(self):

# Number of fds in child processes should not grow with the

# number of running children.

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm = multiprocessing.get_start_method()

if sm == 'fork':

# The fork method by design inherits all fds from the parent,

# trying to go against it is a lost battle

self.skipTest('test not appropriate for {}'.format(sm))

N = 5

evt = self.Event()

q = self.Queue()

procs = [self.Process(target=self._test_child_fd_inflation, args=(evt, q))

for i in range(N)]

for p in procs:

p.start()

try:

fd_counts = [q.get() for i in range(N)]

self.assertEqual(len(set(fd_counts)), 1, fd_counts)

finally:

evt.set()

for p in procs:

p.join()

close_queue(q)

@classmethod

def _test_wait_for_threads(self, evt):

def func1():

time.sleep(0.5)

evt.set()

def func2():

time.sleep(20)

evt.clear()

threading.Thread(target=func1).start()

threading.Thread(target=func2, daemon=True).start()

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_wait_for_threads(self):

# A child process should wait for non-daemonic threads to end

# before exiting

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

evt = self.Event()

proc = self.Process(target=self._test_wait_for_threads, args=(evt,))

proc.start()

proc.join()

self.assertTrue(evt.is_set())

@classmethod

def _test_error_on_stdio_flush(self, evt, break_std_streams={}):

for stream_name, action in break_std_streams.items():

if action == 'close':

stream = io.StringIO()

stream.close()

else:

assert action == 'remove'

stream = None

setattr(sys, stream_name, None)

evt.set()

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_error_on_stdio_flush_1(self):

# Check that Process works with broken standard streams

streams = [io.StringIO(), None]

streams[0].close()

for stream_name in ('stdout', 'stderr'):

for stream in streams:

old_stream = getattr(sys, stream_name)

setattr(sys, stream_name, stream)

try:

evt = self.Event()

proc = self.Process(target=self._test_error_on_stdio_flush,

args=(evt,))

proc.start()

proc.join()

self.assertTrue(evt.is_set())

self.assertEqual(proc.exitcode, 0)

finally:

setattr(sys, stream_name, old_stream)

@warnings_helper.ignore_fork_in_thread_deprecation_warnings()

def test_error_on_stdio_flush_2(self):

# Same as test_error_on_stdio_flush_1(), but standard streams are

# broken by the child process

for stream_name in ('stdout', 'stderr'):

for action in ('close', 'remove'):

old_stream = getattr(sys, stream_name)

try:

evt = self.Event()

proc = self.Process(target=self._test_error_on_stdio_flush,

args=(evt, {stream_name: action}))

proc.start()

proc.join()

self.assertTrue(evt.is_set())

self.assertEqual(proc.exitcode, 0)

finally:

setattr(sys, stream_name, old_stream)

@staticmethod

def _sleep_and_set_event(evt, delay=0.0):

time.sleep(delay)

evt.set()

def check_forkserver_death(self, signum):

# bpo-31308: if the forkserver process has died, we should still

# be able to create and run new Process instances (the forkserver

# is implicitly restarted).

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm = multiprocessing.get_start_method()

if sm != 'forkserver':

# The fork method by design inherits all fds from the parent,

# trying to go against it is a lost battle

self.skipTest('test not appropriate for {}'.format(sm))

from multiprocessing.forkserver import _forkserver

_forkserver.ensure_running()

# First process sleeps 500 ms

delay = 0.5

evt = self.Event()

proc = self.Process(target=self._sleep_and_set_event, args=(evt, delay))

proc.start()

pid = _forkserver._forkserver_pid

os.kill(pid, signum)

# give time to the fork server to die and time to proc to complete

time.sleep(delay * 2.0)

evt2 = self.Event()

proc2 = self.Process(target=self._sleep_and_set_event, args=(evt2,))

proc2.start()

proc2.join()

self.assertTrue(evt2.is_set())

self.assertEqual(proc2.exitcode, 0)

proc.join()

self.assertTrue(evt.is_set())

self.assertIn(proc.exitcode, (0, 255))

def test_forkserver_sigint(self):

# Catchable signal

self.check_forkserver_death(signal.SIGINT)