models - bedrock - daemon thread #593
Conversation
|
|
||
| thread = asyncio.to_thread(self._stream, callback, messages, tool_specs, system_prompt) |
There was a problem hiding this comment.
asyncio.to_thread uses a thread pool executor that is not directly configurable. Consequently, we have to create our own threads in order to control the daemon mode.
|
|
||
| thread = asyncio.to_thread(self._stream, callback, messages, tool_specs, system_prompt) | ||
| task = asyncio.create_task(thread) | ||
| threading.Thread(target=target, daemon=True).start() |
There was a problem hiding this comment.
One down side here is that we create a new thread on every stream invocation rather than pull from a pool of existing threads. In this context however, the impact should be negligible as bedrock.converse_stream is a long running call.
There was a problem hiding this comment.
I'm not as much concerned about the cost (time/space) of creating the new thread. But for highly concurrent use cases this concerns me because of the failure mode.
In the previous case it seems we would have waited until a thread became in the ThreadPoolExecutor allowing the request to succeed. In the current case we are introducing a new thread exhaustion failure mode.
There was a problem hiding this comment.
This is a really good call out and should be thought through carefully. I'll note that I avoided use of a ThreadPoolExecutor here because there is no easy way to configure daemon mode. We would have to derive our own version with some overrides.
|
|
||
| while True: | ||
| event = await queue.get() | ||
| if event is None: | ||
| break | ||
| if isinstance(event, Exception): | ||
| raise event |
There was a problem hiding this comment.
Python threads do not propagate raised exceptions to the main thread automatically. Consequently, we need to explicitly communicate exceptions which we can do so through our existing queue.
| loop.call_soon_threadsafe(queue.put_nowait, event) | ||
| if event is None: | ||
| return | ||
| def target() -> None: |
There was a problem hiding this comment.
nit: should this be private?
There was a problem hiding this comment.
The diff cutoff above but this is a nested function of stream and so is private by that nature.
|
|
||
| thread = asyncio.to_thread(self._stream, callback, messages, tool_specs, system_prompt) | ||
| task = asyncio.create_task(thread) | ||
| threading.Thread(target=target, daemon=True).start() |
There was a problem hiding this comment.
I'm not as much concerned about the cost (time/space) of creating the new thread. But for highly concurrent use cases this concerns me because of the failure mode.
In the previous case it seems we would have waited until a thread became in the ThreadPoolExecutor allowing the request to succeed. In the current case we are introducing a new thread exhaustion failure mode.
Description
Running Bedrock model provider in a daemon thread so that it terminates immediately with the main thread if the main thread goes down. This is particularly useful for
agent-builderwhich has special controls around keyboard interrupts.Related Issues
#560
Type of Change
Bug fix
Testing
How have you tested the change? Verify that the changes do not break functionality or introduce warnings in consuming repositories: agents-docs, agents-tools, agents-cli
hatch run preparehatch run test-integstrands: Ran a local version ofagent-builderthat I am updating with non-blocking threads. Ran against this updated version of the Bedrock model provider and pressedctrl-c. The CLI now exits immediately as before.python test_threading: Wrote the following script for testingctrl-cresponse directlyChecklist
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.