marcenacp's picture
Deploy (see actual commits on https://github.com/mlcommons/croissant).
edf454b
raw
history blame
873 Bytes
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"))