in src/lighteval/utils/timeout.py [0:0]
def timeout(timeout_seconds: int = 10): # noqa: C901
"""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":
# Unix-like approach: signal.alarm
import signal
def decorator(func):
def handler(signum, frame):
raise TimeoutError("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:
# Cancel the alarm and restore previous handler
signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
return wrapper
return decorator
else:
# Windows approach: use multiprocessing
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():
# Timeout: Terminate the process
p.terminate()
p.join()
raise TimeoutError("Operation timed out!")
# If we got here, the process completed in time.
success, value = q.get()
if success:
return value
else:
# The child raised an exception; re-raise it here
raise value
return wrapper
return decorator