Skip to content

Instantly share code, notes, and snippets.

@uysalserkan
Created August 13, 2024 17:21
Show Gist options
  • Save uysalserkan/b25731bd680ff05a59935ed520b5f590 to your computer and use it in GitHub Desktop.
Save uysalserkan/b25731bd680ff05a59935ed520b5f590 to your computer and use it in GitHub Desktop.
Timeout decoration with multiprocessing library.
def timeout_decorator(timeout):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
result_queue = multiprocessing.Queue()
def target_func(queue, *args, **kwargs):
try:
result = func(*args, **kwargs)
queue.put(result)
except Exception as e:
queue.put(e)
process = multiprocessing.Process(target=target_func, args=(result_queue, *args), kwargs=kwargs)
process.start()
process.join(timeout)
if process.is_alive():
process.terminate()
process.kill()
process.join() # Ensure the process has terminated
print(f"Response time exceeded {timeout} seconds.")
return JSONResponse(
content={'error': f'Function execution exceeded {timeout} seconds.'},
status_code=408,
)
if not result_queue.empty():
result = result_queue.get()
if isinstance(result, Exception):
print(f"Error in function: {result}")
return JSONResponse(
content={'error': {result}},
status_code=408,
)
return result
else:
print("Function did not return anything.")
return {'error': 'Function did not return a result'}
return wrapper
return decorator
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment