Latest state updates for perf fixes for multiprocessing communication

Author: markurtz

src/guidellm/benchmark/entrypoints.py

@@ -42,7 +42,7 @@
     NonDistributedEnvironment,
     StrategyType,
 )
-from guidellm.utils import UNSET, Console, InfoMixin
+from guidellm.utils import Console, InfoMixin
 
 __all__ = [
     "benchmark_generative_text",
@@ -103,8 +103,8 @@ async def benchmark_generative_text(  # noqa: C901
     print_updates: bool = False,
     # Aggregators configuration
     add_aggregators: (
-        dict[str, str | dict[str, Any] | Aggregator | CompilableAggregator]
-    ) = UNSET,
+        dict[str, str | dict[str, Any] | Aggregator | CompilableAggregator] | None
+    ) = None,
     warmup: float | None = None,
     cooldown: float | None = None,
     request_samples: int | None = 20,
@@ -209,7 +209,7 @@ async def benchmark_generative_text(  # noqa: C901
             )
         elif constraints:
             raise ValueError(
-                "Constraints must be empty or unset when providing a Profile instance. "
+                "Constraints must be empty when providing a Profile instance. "
                 f"Provided constraints: {constraints} ; provided profile: {profile}"
             )
         console_step.finish(

src/guidellm/scheduler/worker.py

@@ -31,7 +31,7 @@
     ScheduledRequestInfo,
 )
 from guidellm.scheduler.strategy import ScheduledRequestTimings
-from guidellm.utils import MsgpackEncoding, synchronous_to_exitable_async
+from guidellm.utils import MessageEncoding, synchronous_to_exitable_async
 
 __all__ = ["WorkerProcess"]
 
@@ -492,7 +492,7 @@ def _pull_requests_generator(self) -> Generator:
 
             try:
                 message = self.requests_queue.get(timeout=self.poll_intervals)
-                request_tuple = MsgpackEncoding.decode(message)
+                request_tuple = MessageEncoding.decode_message(message)
                 self.pending_requests_queue.sync_put(request_tuple)
             except QueueEmpty:
                 pass  # No update available, continue polling
@@ -522,7 +522,9 @@ def _push_updates_generator(self) -> Generator:
                     update_tuple[2]
                 )
 
-                message = MsgpackEncoding.encode((response, request, request_info))
+                message = MessageEncoding.encode_message(
+                    (response, request, request_info)
+                )
                 self.updates_queue.put(message)
                 self.pending_updates_queue.task_done()
             except culsans.QueueEmpty:

src/guidellm/scheduler/worker_group.py

@@ -41,7 +41,7 @@
 )
 from guidellm.scheduler.strategy import SchedulingStrategy
 from guidellm.scheduler.worker import WorkerProcess
-from guidellm.utils import MsgpackEncoding, synchronous_to_exitable_async
+from guidellm.utils import MessageEncoding, synchronous_to_exitable_async
 
 __all__ = ["WorkerProcessGroup"]
 
@@ -508,7 +508,7 @@ def _populate_requests_next_message(
             )
             state, continue_requests, _ = self._update_state(request_info)
 
-            request_msg = MsgpackEncoding.encode((request, request_info))
+            request_msg = MessageEncoding.encode_message((request, request_info))
             update_msg = (None, request, request_info, state)
 
             return (request_msg, update_msg), continue_requests
@@ -575,7 +575,7 @@ def _populate_updates_process_next(
     ) -> tuple[SchedulerState | None, bool]:
         try:
             message = self.updates_queue.get(timeout=settings.scheduler_poll_interval)
-            response, request, request_info = MsgpackEncoding.decode(message)
+            response, request, request_info = MessageEncoding.decode_message(message)
 
             scheduler_state, _, continue_updates = self._update_state(request_info)
             self.pending_updates_queue.sync_put(
@@ -596,7 +596,7 @@ def _populate_updates_cancel_remaining(
                 message = self.requests_queue.get(
                     timeout=settings.scheduler_poll_interval
                 )
-                request, request_info = MsgpackEncoding.decode(message)
+                request, request_info = MessageEncoding.decode_message(message)
 
                 # Send start first
                 request_info.status = "in_progress"

src/guidellm/scheduler/worker_queue.py

@@ -1,17 +1,22 @@
 from .auto_importer import AutoImporterMixin
 from .console import Colors, Console, ConsoleUpdateStep, StatusIcons, StatusStyles
 from .default_group import DefaultGroupHandler
-from .encoding import MsgpackEncoding
-from .general import (
-    UNSET,
-    UnsetType,
+from .encoding import (
+    EncodedTypeAlias,
+    Encoder,
+    EncodingTypesAlias,
+    MessageEncoding,
+    SerializationTypesAlias,
+    SerializedTypeAlias,
+    Serializer,
+)
+from .functions import (
     all_defined,
     safe_add,
     safe_divide,
     safe_format_timestamp,
     safe_getattr,
     safe_multiply,
-    safe_subtract,
 )
 from .hf_datasets import (
     SUPPORTED_TYPES,
@@ -20,6 +25,13 @@
 from .hf_transformers import (
     check_load_processor,
 )
+from .messaging import (
+    InterProcessMessaging,
+    InterProcessMessagingManagerQueue,
+    InterProcessMessagingPipe,
+    InterProcessMessagingQueue,
+    MessageT,
+)
 from .mixins import InfoMixin
 from .pydantic_utils import (
     PydanticClassRegistryMixin,
@@ -52,23 +64,34 @@
 
 __all__ = [
     "SUPPORTED_TYPES",
-    "UNSET",
     "AutoImporterMixin",
     "Colors",
     "Colors",
     "Console",
     "ConsoleUpdateStep",
     "DefaultGroupHandler",
     "DistributionSummary",
+    "EncodedTypeAlias",
+    "Encoder",
+    "EncodingTypesAlias",
     "EndlessTextCreator",
     "InfoMixin",
     "IntegerRangeSampler",
-    "MsgpackEncoding",
+    "InterProcessMessaging",
+    "InterProcessMessagingManagerQueue",
+    "InterProcessMessagingPipe",
+    "InterProcessMessagingQueue",
+    "MessageEncoding",
+    "MessageEncoding",
+    "MessageT",
     "Percentiles",
     "PydanticClassRegistryMixin",
     "RegistryMixin",
     "ReloadableBaseModel",
     "RunningStats",
+    "SerializationTypesAlias",
+    "SerializedTypeAlias",
+    "Serializer",
     "SingletonMixin",
     "StandardBaseDict",
     "StandardBaseModel",
@@ -78,7 +101,6 @@
     "StatusStyles",
     "ThreadSafeSingletonMixin",
     "TimeRunningStats",
-    "UnsetType",
     "all_defined",
     "check_load_processor",
     "clean_text",
@@ -91,7 +113,6 @@
     "safe_format_timestamp",
     "safe_getattr",
     "safe_multiply",
-    "safe_subtract",
     "save_dataset_to_file",
     "split_text",
     "split_text_list_by_length",