MCPcopy Index your code
hub / github.com/RustPython/RustPython / tearDownModule

Function tearDownModule

Lib/test/_test_multiprocessing.py:6779–6808  ·  view source on GitHub ↗
()

Source from the content-addressed store, hash-verified

6777 multiprocessing.get_logger().setLevel(LOG_LEVEL)
6778
6779 def tearDownModule():
6780 need_sleep = False
6781
6782 # bpo-26762: Some multiprocessing objects like Pool create reference
6783 # cycles. Trigger a garbage collection to break these cycles.
6784 test.support.gc_collect()
6785
6786 multiprocessing.set_start_method(old_start_method[0], force=True)
6787 # pause a bit so we don't get warning about dangling threads/processes
6788 # TODO: RUSTPYTHON: Filter out stopped processes since gc.collect() is a no-op
6789 processes = {p for p in multiprocessing.process._dangling if p.is_alive()} - {p for p in dangling[0] if p.is_alive()}
6790 if processes:
6791 need_sleep = True
6792 test.support.environment_altered = True
6793 support.print_warning(f'Dangling processes: {processes}')
6794 processes = None
6795
6796 # TODO: RUSTPYTHON: Filter out stopped threads since gc.collect() is a no-op
6797 threads = {t for t in threading._dangling if t.is_alive()} - {t for t in dangling[1] if t.is_alive()}
6798 if threads:
6799 need_sleep = True
6800 test.support.environment_altered = True
6801 support.print_warning(f'Dangling threads: {threads}')
6802 threads = None
6803
6804 # Sleep 500 ms to give time to child processes to complete.
6805 if need_sleep:
6806 time.sleep(0.5)
6807
6808 multiprocessing.util._cleanup_tests()
6809
6810 remote_globs['setUpModule'] = setUpModule
6811 remote_globs['tearDownModule'] = tearDownModule

Callers

nothing calls this directly

Calls 3

set_start_methodMethod · 0.45
is_aliveMethod · 0.45
sleepMethod · 0.45

Tested by

no test coverage detected