|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import os |
|
from math_verify.errors import TimeoutException |
|
|
|
|
|
def timeout(timeout_seconds: int = 10): |
|
"""A decorator that applies a timeout to the decorated function. |
|
|
|
Args: |
|
timeout_seconds (int): Number of seconds before timing out the decorated function. |
|
Defaults to 10 seconds. |
|
|
|
Notes: |
|
On Unix systems, uses a signal-based alarm approach which is more efficient as it doesn't require spawning a new process. |
|
On Windows systems, uses a multiprocessing-based approach since signal.alarm is not available. This will incur a huge performance penalty. |
|
""" |
|
if os.name == "posix": |
|
|
|
import signal |
|
|
|
def decorator(func): |
|
def handler(signum, frame): |
|
raise TimeoutException("Operation timed out!") |
|
|
|
def wrapper(*args, **kwargs): |
|
old_handler = signal.getsignal(signal.SIGALRM) |
|
signal.signal(signal.SIGALRM, handler) |
|
signal.alarm(timeout_seconds) |
|
try: |
|
return func(*args, **kwargs) |
|
finally: |
|
|
|
signal.alarm(0) |
|
signal.signal(signal.SIGALRM, old_handler) |
|
|
|
return wrapper |
|
|
|
return decorator |
|
|
|
else: |
|
|
|
from multiprocessing import Process, Queue |
|
|
|
def decorator(func): |
|
def wrapper(*args, **kwargs): |
|
q = Queue() |
|
|
|
def run_func(q, args, kwargs): |
|
try: |
|
result = func(*args, **kwargs) |
|
q.put((True, result)) |
|
except Exception as e: |
|
q.put((False, e)) |
|
|
|
p = Process(target=run_func, args=(q, args, kwargs)) |
|
p.start() |
|
p.join(timeout_seconds) |
|
|
|
if p.is_alive(): |
|
|
|
p.terminate() |
|
p.join() |
|
raise TimeoutException("Operation timed out!") |
|
|
|
|
|
success, value = q.get() |
|
if success: |
|
return value |
|
else: |
|
|
|
raise value |
|
|
|
return wrapper |
|
|
|
return decorator |
|
|