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")) | |