Scheduler refactor [utils]: multiprocessing encoding and messaging #293
Conversation
This file contains hidden or 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
markurtz
force-pushed
the
feature/refactor/utils-extended
branch
from
d463e77 to
02f97ff
Compare
markurtz
force-pushed
the
feature/refactor/utils-concurrency
branch
from
8831554 to
da5effc
Compare
There was a problem hiding this comment.
Pull Request Overview
This PR introduces comprehensive multiprocessing utilities for the scheduler refactor, focusing on encoding and messaging capabilities to optimize data transfer between processes. The implementation provides configurable serialization strategies, binary encoding options, and high-performance interprocess communication mechanisms.
- Adds robust message encoding system with support for Pydantic models, configurable serialization (dict/sequence), and binary encoding (msgpack/msgspec)
- Implements abstract messaging framework with concrete implementations for queues, manager queues, and pipes
- Provides comprehensive test coverage for all new utility modules including smoke, sanity, and regression scenarios
Reviewed Changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/unit/utils/test_text.py | Fixes spelling error in test docstring (puncutation → punctuation) |
| tests/unit/utils/test_messaging.py | Comprehensive test suite for messaging utilities with multiprocessing scenarios |
| tests/unit/utils/test_encoding.py | Complete test coverage for encoding utilities with various data types and serialization strategies |
| src/guidellm/utils/messaging.py | Core messaging abstractions and implementations for interprocess communication |
| src/guidellm/utils/encoding.py | Message encoding utilities with Pydantic support and configurable serialization/encoding |
| src/guidellm/utils/init.py | Exports new encoding and messaging utilities from utils module |
Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.
markurtz
force-pushed
the
feature/refactor/utils-concurrency
branch
from
7b74f29 to
e8300a9
Compare
markurtz
force-pushed
the
feature/refactor/utils-extended
branch
from
fc47105 to
b70ee98
Compare
Base automatically changed from
feature/refactor/utils-extended
to
feature/refactor/main
August 26, 2025 19:41
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Signed-off-by: Mark Kurtz <mark.j.kurtz@gmail.com>
… for the scheduler refactor
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Signed-off-by: Mark Kurtz <mark.j.kurtz@gmail.com>
markurtz
force-pushed
the
feature/refactor/utils-concurrency
branch
from
1ecc54a to
cff8d91
Compare
sjmonson
requested changes
Aug 26, 2025
Co-authored-by: Samuel Monson <smonson@redhat.com> Signed-off-by: Mark Kurtz <mark.j.kurtz@gmail.com>
4 tasks
jaredoconnell
approved these changes
Aug 28, 2025
Comment on lines
+242
to
+251
| async def get(self, timeout: float | None = None) -> ReceiveMessageT: | ||
| """ | ||
| Retrieve message from receive buffer with optional timeout. | ||
|
|
||
| :param timeout: Maximum time to wait for a message | ||
| :return: Decoded message from the receive buffer | ||
| """ | ||
| return await asyncio.wait_for( | ||
| self.buffer_receive_queue.async_get(), timeout=timeout | ||
| ) |
There was a problem hiding this comment.
Does this function return a TimeoutError when it times out? If so I think it would be beneficial to document that on all of these functions that have a timeout.
|
Resetting the refactor state with new PRs, closing this one out |
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.
Summary
This PR introduces four new utility modules that provide improved capabilities for transferring information across processes to maximize performance. This is done through encoding.py to best serialize the data into a transferrable format and minimize the size of the data as well as through messaging.py which supports high performance multiprocessing queues and pipes for transfer that works with the encoding.
Details
MessageEncoding.pyInterProcessMessaging,InterProcessMessagingManagerQueue,InterProcessMessagingPipe,InterProcessMessagingQueueTest Plan
Related Issues
Use of AI
## WRITTEN BY AI ##)