Feat/max error rate #171
Feat/max error rate #171
Conversation
This reverts commit 6059af1.
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
There was a problem hiding this comment.
I left a few minor comments for the less intrusive changes. Also, please be sure to run tox -e style and tox -e types locally and check those are passing -- there are a bunch of styling pieces, such as where parentheses are placed, how new lines are escaped, etc that are inconsistent with the ruff styling setup. Running tox -e style locally applied several fixes and raised some errors that couldn't be resolved without either modifying the code or ignoring them where appropriate.
For implementation, I think defaulting always to use the shutdown event is a good pattern rather than relying on two signals: either None being placed on the queue or the shutdown event being set if the maximum error rate is reached. I'd like to incorporate that into this to reduce complexity.
Also, I think a single parameter, max_errors, would be a great extension where it can be either max_error_rate or an absolute number of errors that can occur, allowing us to support all benchmarking pathways. Along these lines, for ones where we don't know how many requests are going to run, or maybe in general, using a simple sample of the last X requests based on the precision needed could work here to ensure rate can be set for those.
Additionally, with the above, as well as some combination logic in asyncio, we should be able to simplify the changes, so we don't need to make significant changes to the resolve requests pathways and can cancel for any type. For example, we should be able to replace the root process runner to call into async loops and handle the shutdown event globally in a central place with something like the following, where ideally shutdown_poll_interval would grab from the global settings so users can adjust it:
def run_process(
self,
type_: Literal["synchronous", "asynchronous"],
requests_queue: multiprocessing.Queue,
results_queue: multiprocessing.Queue,
shutdown_event: multiprocessing.Event,
shutdown_poll_interval: float,
process_id: int,
max_concurrency: int,
):
async def _process_runner():
if type_ == "synchronous":
loop_task = asyncio.create_task(self.synchronous_loop(...))
elif type_ == "asynchronous":
loop_task = asyncio.create_task(self.asyncronous_loop(...))
else:
raise ValueError(f"Invalid process type: {type_}")
shutdown_task = asyncio.create_task(
self.wait_for_shutdown(shutdown_event, shutdown_poll_interval)
)
done, pending = await asyncio.wait(
[
loop_task,
shutdown_task,
],
return_when=asyncio.FIRST_EXCEPTION,
)
for task in pending:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
for task in done:
if task.exception():
raise task.exception() # need to check and ignore cancelations
try:
asyncio.run(_process_runner())
except Exception as exc: # noqa: BLE001
logger.error(
f"Error in worker process {process_id}: {exc}",
exc_info=True,
stack_info=True,
)
finally:
shutdown_event.set() # ensure shutdown event is set to stop other processes
async def wait_for_shutdown(
self, shutdown_event: multiprocessing.Event, shutdown_poll_interval: float
):
while not shutdown_event.is_set():
await asyncio.sleep(shutdown_poll_interval)
raise asyncio.CancelledError("Shutdown event set, cancelling process loop.")
async def synchronous_loop(self, ...):
while True:
process_request = await self.get_request(requests_queue)
...
async def asyncronous_loop(self, ...):
while True:
process_request = await self.get_request(requests_queue)
...| "The number of errored requests divided by the number " | ||
| "of errored requests. This can be higher than max_error_rate " | ||
| "(if applicable) cause it does not take into " | ||
| "account incomplete requests." |
There was a problem hiding this comment.
I'm not following this point, the error rate we calculate and compare to should never include incomplete, right?
There was a problem hiding this comment.
Isn't max_error also not looking at incomplete requests, though? How would it be higher for error_rate vs max_error since those should be based off the same calculations, right?
src/guidellm/scheduler/scheduler.py
Outdated
| @@ -159,6 +166,17 @@ async def run( | |||
| run_info, | |||
| ) | |||
| if iter_result is not None: | |||
| if iter_result.request_info.errored \ | |||
There was a problem hiding this comment.
NIT: should use paranetheses rather than line escapes for if statements across multiple lines, tox -e style should fix
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
📦 Build Artifacts Available |
|
I've changed the logic as suggested to have a single termination method which is a After implementing the code like you suggested above I ran into an issue with Another super weird thing that happened is the raising specifically Also, I added the context of last X request as suggested but I feel that for the scenario where we do know the exact amount, i.e const rate + fixed duration, it is best to look at the entire run. |
| @@ -4,40 +4,40 @@ repos: | |||
| hooks: | |||
| - id: trailing-whitespace | |||
| - id: end-of-file-fixer | |||
| - repo: https://github.com/astral-sh/ruff-pre-commit | |||
There was a problem hiding this comment.
We'll need these reenabled. If you ever need to push and skip recommit, you can pass in the --no-verify flag with your git commit command
There was a problem hiding this comment.
Also @sjmonson just landed a diff that might clear up some of the issues you were seeing
| @@ -147,6 +147,8 @@ The `guidellm benchmark` command is used to run benchmarks against a generative | |||
|
|
|||
| - `--max-requests`: Sets the maximum number of requests for each benchmark run. If not provided, the benchmark will run until `--max-seconds` is reached or the dataset is exhausted. | |||
|
|
|||
| - `--max-error-rate`: The maximum error rate after which a benchmark will stop. Applicable only for finite deterministic scenarios i.e `rate_type` is `constant` and `--max-seconds` exists OR `--max-requests` exists OR the dataset is finite. If `--max-error-rate` is `None` or not applicable, benchmarks will continue regardless of error rate. | |||
There was a problem hiding this comment.
| - `--max-error-rate`: The maximum error rate after which a benchmark will stop. Applicable only for finite deterministic scenarios i.e `rate_type` is `constant` and `--max-seconds` exists OR `--max-requests` exists OR the dataset is finite. If `--max-error-rate` is `None` or not applicable, benchmarks will continue regardless of error rate. | |
| - `--max-error`: The maximum error rate after which a benchmark will stop. Can either be a rate i.e 0 < rate < 1 or constant number. If rate is given and rate_type is 'constant' and 'max_seconds' exists then the rate will be calculated as part of the total expected requests counts i.e rate * duration. If rate is given and number of requests is not pre-determined than a context window of the last requests will be looked at. Context window size is configurable under GUIDELLM__ERROR_CHECK_WINDOW_SIZE. If a number above 1 is given than we just count the total number of error and check if it's above the threshold. |
| @@ -46,12 +50,15 @@ class SchedulerRunInfo(StandardBaseModel): | |||
| end_number: float | |||
| processes: int | |||
| strategy: SchedulingStrategy | |||
| last_requests_statuses: deque[RequestStatus] | |||
There was a problem hiding this comment.
I think we can simplify the logic here so we don't need to add this in especially since this list can be come very long and it is not json serializable so would break any pydantic serialization in the event someone wants to save that state. See note later down where that logic is located
| "RequestLoader", | ||
| "RequestLoaderDescription", | ||
| ] | ||
|
|
||
|
|
||
| class GetInfiniteDatasetLengthError(Exception): |
There was a problem hiding this comment.
I think this exception and logic is no longer needed since if we can get the length then we go through a specific logic route and if we can't, then we fallback on the window
| if max_duration is not None and max_duration < 0: | ||
| raise ValueError(f"Invalid max_duration: {max_duration}") | ||
| self._validate_scheduler_params( | ||
| scheduling_strategy, max_duration, max_error, max_number |
There was a problem hiding this comment.
NIT: could we keep the arg order the same as the surrounding function? Makes it a bit easier to track especially if additions com in later
| ) | ||
| asyncio.create_task(self.send_result(results_queue, result)) | ||
| asyncio.create_task(self.send_result(results_queue, request_scheduled_result)) |
There was a problem hiding this comment.
good catch on potential threading issues with these. Rather than polluting the local namespace with a bunch of vars that aren't used later, though, could we add the WorkerProcessResult to pass directly to this call function?
| process_request := await self.get_request(requests_queue) | ||
| ) is not None: | ||
| dequeued_time = time.time() | ||
| # We are using a separate internal event |
There was a problem hiding this comment.
See my note on fixes for get_request so we don't hit this race condition ideally and can rely on the shutdown event here
| process_id: int, | ||
| max_concurrency: Optional[int] = None, |
There was a problem hiding this comment.
Can we change this out so it isn't optional? I think we had a change either pushed in or queued for that and that way we don't need to validate it for async. I believe the changes are in so that for sync, it still returns a max_concurrency there as well
| internal_shutdown_event.set() | ||
| try: # noqa: SIM105 | ||
| await task | ||
| except asyncio.CancelledError: |
There was a problem hiding this comment.
Since we are raising a different error in the shutdown task, there could be a race condition here where an error occurs and shutdown is set before we call await causing an uncaught exception here
| while not shutdown_event.is_set(): # noqa: ASYNC110 | ||
| await asyncio.sleep(shutdown_poll_interval) | ||
|
|
||
| # Raising asyncio.CancelledError instead would |
There was a problem hiding this comment.
I think this is the correct way where we raise a specific error signaling this which we can track and catch. Can we remove the comments here since this is pretty self explanatory based on the name of the exception?
Addressing this issue:
#105
--max-error-rateflag was added.It is only applicable in either or the the following scenarios:
max_requestsis setrate_typeisconstantandmax_durationis setsweepfor example)multiprocessing.Eventwhich is used to trigger a shutdown once we hitmax_error_rate.max_error_rate- added toreporterror_ratein the final report is calculated astotal errored / total requests finalizedi.e we excludeincomplete requests