[DOW-113] deprecate output queue and manually attach workers to each other #593
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
ajar98
force-pushed
the
ajay/dow-113-deprecate-output-queue-and-manually-attach-workers-to-each
branch
from
c86f666
to
9e28363
Compare
ajar98
force-pushed
the
ajay/dow-113-deprecate-output-queue-and-manually-attach-workers-to-each
branch
from
c787ea0
to
84afa9c
Compare
ajar98
changed the title
adnaans
reviewed
Jul 5, 2024
| @@ -253,7 +259,9 @@ async def test_transcriptions_worker_ignores_utterances_before_initial_message( | |||
| is_final=True, | |||
| ), | |||
| ) | |||
| assert await _consume_worker_output(streaming_conversation.transcriptions_worker) is None | |||
| transcriptions_worker_consumer = QueueConsumer() | |||
There was a problem hiding this comment.
You should remove line 239 right? And attach the consumer earlier? The consumer otherwise doesn't get the consume_nonblocking() results above
There was a problem hiding this comment.
yup, absolutely right!
| @@ -310,7 +318,10 @@ async def test_transcriptions_worker_ignores_associated_ignored_utterance( | |||
| is_final=False, | |||
| ), | |||
| ) | |||
| assert await _consume_worker_output(streaming_conversation.transcriptions_worker) is None | |||
| transcriptions_worker_consumer = QueueConsumer() | |||
There was a problem hiding this comment.
If I'm correct above, likewise here and in all other relevant parts below
…ally-attach-workers-to-each
adnaans
suggested changes
Jul 5, 2024
| @@ -390,6 +383,7 @@ async def test_transcriptions_worker_interrupts_on_interim_transcripts( | |||
|
|
|||
| transcriptions_worker_consumer = QueueConsumer() | |||
There was a problem hiding this comment.
consumer should be added before like 376 again
| @@ -437,6 +428,7 @@ async def test_transcriptions_worker_interrupts_immediately_before_bot_has_begun | |||
| ) | |||
| transcriptions_worker_consumer = QueueConsumer() | |||
| streaming_conversation.transcriptions_worker.consumer = transcriptions_worker_consumer | |||
| streaming_conversation.transcriptions_worker.start() | |||
There was a problem hiding this comment.
consumer should be added before 422
adnaans
approved these changes
Jul 5, 2024
ajar98
added a commit
that referenced
this pull request
Jul 6, 2024
…other (#593) * deprecate output queues * fix quickstarts * fix mypy * fix tests * fix mypy * adds comment * adds back streamingconversation test * make input queue private on AsyncWorker * update tests * resolve PR comments
ajar98
added a commit
that referenced
this pull request
Jul 8, 2024
* remove phantom parameter (#612) * support files in make transcribe (#610) * support files in make transcribe * switch comment * uses .value instead of passing enum for sampling rate (#613) * [hotfix] fix ref in make transcribe (#614) * [DOW-113] deprecate output queue and manually attach workers to each other (#593) * deprecate output queues * fix quickstarts * fix mypy * fix tests * fix mypy * adds comment * adds back streamingconversation test * make input queue private on AsyncWorker * update tests * resolve PR comments * create scripts for vocodehq-public (#615) * add script used to make PR * adds test target for vocodehq-public
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
overall idea here - pushes the notion of queues under the abstraction barrier of workers. Workers do not need to know about the inner workings of how other workers process their events. In order to wire up workers, manually connect them to each other using instance variables and only use
consume_nonblockingto communicate - this also improves the typing of events coming out of workers.Mechanically:
AbstractWorkerwhich just knows how to consume inputs, but places no restrictions on how it createsasyncio.Queues to do the processingAsyncWorkerto only create aninput_queue, and not have it be overrideableStreamingConversationso that the workers are connected to each other as described aboveAlso adds a nice test that asserts that a message can make it through the whole pipeline (
test_streaming_conversation_pipeline)