Spaces:
Running
Running
import multiprocessing | |
import time | |
from typing import TypedDict | |
class _Result(TypedDict): | |
bar: int | |
def bar(result): | |
while True: | |
time.sleep(1) | |
result["bar"] += 1 | |
print(result["bar"]) | |
if result["bar"] > 5: | |
return | |
def foo(): | |
"""Generates the data and waits at most _TIMEOUT_SECONDS.""" | |
with multiprocessing.Manager() as manager: | |
result: _Result = manager.dict(bar=0) | |
process = multiprocessing.Process(target=bar, args=(result,)) | |
process.start() | |
if not process.is_alive(): | |
return result | |
time.sleep(3) | |
if process.is_alive(): | |
process.kill() | |
result["exception"] = TimeoutError( | |
"The generation took too long and was killed." | |
) | |
return _Result(**result) | |
print("FINAL RESULT", foo().get("bar")) | |