如何在出现异常时终止multiprocessing.map的多线程任务？

Great question! You absolutely can stop all running tests as soon as one fails—though the approach varies a bit depending on whether you're using threads (like your current ThreadPool setup) or actual multiprocessing. Let's break down practical, safe solutions:

Solution 1: Thread Pool (Your Current Setup) #

Python doesn't support safe forced termination of threads (it can lead to locked resources or unstable state), so the best approach is to use a thread-safe signal that your test functions can check and respond to. We'll use threading.Event for this:

import threading
import sys
from multiprocessing.dummy import ThreadPool

# Create a shared, thread-safe event to trigger termination
stop_event = threading.Event()

def run_test_wrapper(test_func):
    try:
        # Check if we need to stop before starting the test
        if stop_event.is_set():
            return
        
        # Run the test — for long-running tests, add stop checks inside!
        test_func()
    except Exception as e:
        print(f"Test failed with error: {str(e)}")
        # Set the event to tell all other threads to exit
        stop_event.set()
        # Re-raise the exception to trigger the main error handler
        raise

testlist = [test1, test2, test3]
pool = ThreadPool(len(testlist))

try:
    pool.map(run_test_wrapper, testlist)
    print('Success')
    sys.exit(0)
except Exception:
    print('FAILURE!')
    # Ensure the stop signal is sent to any remaining threads
    stop_event.set()
    # Close the pool and wait for threads to exit gracefully
    pool.close()
    pool.join()
    sys.exit(1)

Important for Long-Running Tests #

If your tests run in loops or take time to complete, add periodic checks for the stop_event inside the test functions themselves:

def test1():
    for test_step in range(1000):
        # Exit early if termination is triggered
        if stop_event.is_set():
            print("test1 received stop signal — exiting")
            return
        
        # Execute your test logic here
        ...

Solution 2: True Multiprocessing (If That's What You Intended) #

If you meant to use separate OS processes instead of threads, we can use multiprocessing.Event and even force-terminate processes if needed (safer here than with threads):

import multiprocessing
import sys

stop_event = multiprocessing.Event()

def run_test_wrapper(test_func):
    try:
        if stop_event.is_set():
            return
        test_func()
    except Exception as e:
        print(f"Test failed with error: {str(e)}")
        stop_event.set()
        raise

testlist = [test1, test2, test3]
pool = multiprocessing.Pool(len(testlist))

try:
    pool.map(run_test_wrapper, testlist)
    print('Success')
    sys.exit(0)
except Exception:
    print('FAILURE!')
    stop_event.set()
    # Force-terminate all running processes immediately
    pool.terminate()
    pool.join()
    sys.exit(1)

pool.terminate() kills all child processes right away, which is more aggressive but works well for isolated tests that don't require cleanup.

内容的提问来源于stack exchange，提问作者Ray Salemi