多进程超时装饰器类遇Pickling错误，求将Windows兼容超时函数改写为装饰器

Great call using multiprocessing.Pool for timeout handling on Windows—this avoids the pitfalls of signal-based or thread-based approaches that often fail there. Converting your existing function into a reusable decorator is straightforward, and we can make it flexible so you can apply it to any function with a specified timeout.

Parameterized Timeout Decorator Implementation #

Here's how to wrap your logic into a clean, reusable decorator that accepts a timeout value:

import multiprocessing
from multiprocessing import Pool, TimeoutError

def timeout(timeout_seconds):
    def decorator(func):
        def wrapper(*args, **kwargs):
            # Create a single-process pool to isolate the target function
            with Pool(processes=1) as pool:
                # Submit the function call with its arguments/keyword arguments
                result = pool.apply_async(func, args=args, kwds=kwargs)
                try:
                    # Wait for the result with the specified timeout
                    return result.get(timeout=timeout_seconds)
                except TimeoutError:
                    # Re-raise the timeout exception so the caller can handle it
                    raise
        return wrapper
    return decorator

How It Works #

• Parameterized structure: First you pass the timeout value (e.g., @timeout(5) for a 5-second limit), then the decorator wraps your target function.
• Isolation via Pool: The single-process pool ensures the target function runs in a separate process, which is critical for reliable timeout handling on Windows.
• Exception consistency: It preserves the original behavior of raising multiprocessing.TimeoutError when the function exceeds the timeout, so you can catch and handle this exactly as before.
• Clean cleanup: The with statement guarantees the pool is properly closed and resources are released after the function completes or times out.

Example Usage #

import time

# Decorate a function with a 3-second timeout
@timeout(3)
def slow_operation(delay):
    time.sleep(delay)
    return f"Success after {delay} seconds"

# Test with a short delay (completes on time)
try:
    print(slow_operation(2))  # Output: Success after 2 seconds
except TimeoutError:
    print("Operation timed out")

# Test with a long delay (triggers timeout)
try:
    print(slow_operation(5))
except TimeoutError:
    print("Operation timed out")  # This will execute

Critical Windows-Specific Note #

When running on Windows, always wrap code that uses the decorated function in if __name__ == '__main__':—this prevents infinite process spawning, which is a requirement for multiprocessing to work correctly:

if __name__ == '__main__':
    try:
        slow_operation(5)
    except TimeoutError:
        print("Timed out!")

Also, ensure the function being decorated (and any arguments passed to it) are pickleable—most standard functions and data types work fine, but custom objects may need explicit pickling support.

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