Conversation
demos/multi_guess_server.py
Outdated
|
|
||
|
|
||
| class MultiQueue: | ||
| def __init__(self, queues: dict[str, Any]): |
There was a problem hiding this comment.
Find the real type (not Any).
demos/multi_guess_server.py
Outdated
| class MultiQueue: | ||
| def __init__(self, queues: dict[str, Any]): | ||
| self.queues = queues | ||
| self.gets = {} # asyncio tasks -> identity |
There was a problem hiding this comment.
self.gets: dict[asyncio.Task, str]
demos/multi_guess_server.py
Outdated
| continue # Skip - already removed | ||
|
|
||
| result = await task | ||
| return ident, result |
demos/multi_guess_server.py
Outdated
| # Done waiting for this identity -> remove | ||
| self._active.discard(ident) | ||
|
|
||
| async def __anext__(self): |
demos/multi_guess_server.py
Outdated
| for ident in players | ||
| }) as guess_queues | ||
| these({ident: queue('guess', ident) for ident in players}) as guess_queues, | ||
| MultiQueue(guess_queues) as mq |
There was a problem hiding this comment.
async with MultiQueue('guess', ['id1', 'id2', 'id3']) as guess_queues:
demos/multi_guess_server.py
Outdated
| self.gets: dict[asyncio.Task, str] = {} | ||
| self.reverse: dict[str, asyncio.Task] = {} | ||
|
|
There was a problem hiding this comment.
I would name these something like
task_to_ident
ident_to_task
demos/multi_guess_server.py
Outdated
| async def __aenter__(self): | ||
| # Listen on all queues -> create a task for each queue.get() | ||
| for ident, q in self.queues.items(): | ||
| task = asyncio.create_task(q.get()) |
There was a problem hiding this comment.
I think we need to use the quest version of create_task so the task get's recorded.
demos/multi_guess_server.py
Outdated
| for task in done: | ||
| ident = self.gets.pop(task) | ||
| # Stop listening to this identity | ||
| self.reverse.pop(ident, None) |
There was a problem hiding this comment.
try del self.reverse[ident] instead
There was a problem hiding this comment.
That might fix the await warning you are getting.
demos/multi_guess_server.py
Outdated
| self.gets.pop(task) | ||
| task.cancel() | ||
|
|
||
| async def __aiter__(self): |
There was a problem hiding this comment.
I think this logic already only supports one response per identity.
We need some tests demonstrating how the MultiQueue should work.
demos/multi_guess_server.py
Outdated
| for guess_get in asyncio.as_completed(guess_gets): | ||
| guess = await guess_get | ||
| ident = guess_gets[guess] | ||
| async with SingleResponseMultiQueue(queues) as mq: |
There was a problem hiding this comment.
async with queues('guess', players) as qs:
demos/multi_guess_server.py
Outdated
| continue | ||
|
|
||
|
|
||
| class SingleResponseMultiQueue: |
demos/test_multiqueue.py
Outdated
| results = [] | ||
|
|
||
| async def dummy_workflow(): | ||
| async with SingleResponseMultiQueue(queues) as mq: |
There was a problem hiding this comment.
MultiQueue('message', players, single_response=True)
| # Iterate guesses one at a time | ||
| async with MultiQueue('guess', players, single_response=True) as mq: | ||
| async for ident, guess in mq: |
There was a problem hiding this comment.
I love how much more concise this is from what it used to be. :)
| async def __aenter__(self): | ||
| # Listen on all queues -> create a task for each queue.get() | ||
| for ident, q in self.queues.items(): | ||
| self._add_task(ident, q) | ||
| return self |
There was a problem hiding this comment.
All the queues need to be __aenter__ed here or they won't appear in workflow history.
src/quest/external.py
Outdated
| def remove(self, ident: str): | ||
| # Stop listening to this identity queue | ||
| if ident not in self.ident_to_task: | ||
| raise KeyError(f"Identity '{ident}' does not exist in MultiQueue.") | ||
|
|
||
| task = self.ident_to_task.pop(ident) | ||
| self.task_to_ident.pop(task) | ||
| task.cancel() |
There was a problem hiding this comment.
We need to __aexit__ each queue as it is removed.
| async def __aexit__(self, exc_type, exc_val, exc_tb): | ||
| # Cancel all pending tasks - context exits | ||
| for task in self.task_to_ident: | ||
| task.cancel() |
There was a problem hiding this comment.
We need to __aexit__ each queue as we exit.
| async def __aiter__(self): | ||
| while self.task_to_ident: | ||
| # Wait until any of the current task is done | ||
| done, _ = await asyncio.wait(self.task_to_ident.keys(), return_when=asyncio.FIRST_COMPLETED) | ||
|
|
||
| for task in done: | ||
| ident = self.task_to_ident.pop(task) | ||
| # Stop listening to this identity | ||
| del self.ident_to_task[ident] | ||
|
|
||
| try: | ||
| result = await task | ||
| yield ident, result | ||
|
|
||
| # Start listening again | ||
| if not self.single_response: | ||
| self._add_task(ident, self.queues[ident]) | ||
|
|
||
| except asyncio.CancelledError: | ||
| continue |
There was a problem hiding this comment.
This is looking cleaner. Nice.
| workflow = historian.run() | ||
|
|
||
| await asyncio.sleep(0.1) | ||
|
|
There was a problem hiding this comment.
get resources for both p1 and p2, and show they both have "chats"
Then p1 says "bye", and show that p1 no longer has chats.
show that p2 has "chats" until the end of the workflow.
| await historian.record_external_event('chat', 'p1', 'put', 'hello') | ||
| await historian.record_external_event('chat', 'p2', 'put', 'hi') | ||
| # Second message from p1 - should be ignored due to single_response = True | ||
| await historian.record_external_event('chat', 'p1', 'put', 'should not be received') |
There was a problem hiding this comment.
This should throw an exception.
| wrapper = self.queues.pop(ident, None) | ||
| if wrapper: | ||
| await wrapper.__aexit__(None, None, None) | ||
|
|
There was a problem hiding this comment.
wrapper = self.queues.pop(ident)
await wrapper.__aexit__(None, None, None)
| if wrapper: | ||
| await wrapper.__aexit__(None, None, None) | ||
|
|
||
| self.active_queues.pop(ident, None) |
2 participants
No description provided.