[BUG] Python SDK: Scheduling too much work with `asyncio.to_thread` will exhaust the default executor pool and block hatchet worker
Describe the issue
If you schedule work with asyncio.to_thread or asyncio.get_running_loop().run_in_executor(None, ...) that work will be submitted to the default executor threadpool which is shared with hatchet.
This can lead to the main loop being blocked and lots of warnings that the main loop is blocked with a link to https://hatchet.run/blog/warning-event-loop-blocked.
For me the issue was two-fold:
- (1) Execution was actually blocked
- (2) the blog post indicated that this only happens if you accidently run blocking code inside async functions
I have made a reproducible test example here (will add code further down too)
Environment
- SDK: Python v1.38.1
- Engine: Self-hosted (but doesn't matter)
Expected behavior
(1) I would like it not to be blocked (2) I would like the warning to indicate that this might also be the reason
Code to Reproduce, Logs, or Screenshots
test file:
from subprocess import Popen
from typing import Any
from hatchet_sdk.runnables.types import EmptyModel
import pytest
from hatchet_sdk import Hatchet
from tests.thread_pool_exhaust.worker import my_async_task_that_i_do_control
@pytest.mark.parametrize(
"on_demand_worker",
[
[
"poetry",
"run",
"python",
"tests/thread_pool_exhaust/worker.py",
]
],
indirect=True,
)
@pytest.mark.asyncio(loop_scope="session")
async def test_thread_pool_exhaust(
hatchet: Hatchet, on_demand_worker: Popen[Any]
) -> None:
await my_async_task_that_i_do_control.aio_run_many(
[
my_async_task_that_i_do_control.create_bulk_run_item(EmptyModel())
for _ in range(20)
]
)worker file
from datetime import timedelta
from hatchet_sdk import Context, EmptyModel, Hatchet
from hatchet_sdk import Hatchet
import time
import asyncio
hatchet = Hatchet()
async def lifespan():
import asyncio
from concurrent.futures import ThreadPoolExecutor
# the default default (hehe) executor is instantiated with ThreadPoolExecutor()
# which has max_workers = min(32, (os.cpu_count() or 1) + 4)
# 5 simulates a single core worker
asyncio.get_running_loop().set_default_executor(
ThreadPoolExecutor(max_workers=5, thread_name_prefix="tiny-default-pool")
)
yield
def sync_code_i_dont_control():
time.sleep(5)
@hatchet.task()
async def my_async_task_that_i_do_control(input: EmptyModel, ctx: Context) -> None:
await asyncio.to_thread(sync_code_i_dont_control)
def main() -> None:
worker = hatchet.worker(
"worker_that_will_be_exhausted",
slots=50,
workflows=[
my_async_task_that_i_do_control,
],
lifespan=lifespan,
)
worker.start()
if __name__ == "__main__":
main()Additional context I think it can be possible to determine if the default executor is at capacity by looking at the queue statistics before writing the warning
AI Disclosure
- I acknowledge that an LLM was used in the creation of this Issue, in accordance with Hatchet's AI_POLICY.md.
- Details: No AI
Source: hatchet-dev/hatchet