[Bugfix] E2E metric incorrect qwen3-omni with async chunk feature (#1018)
Signed-off-by: Junhong Liu <98734602+LJH-LBJ@users.noreply.github.com> Signed-off-by: Junhong Liu <ljh_lbj@163.com> Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>
This commit is contained in:
Junhong LiuandHongsheng Liu authored and GitHub committed
1 parent
f95f8e936d
commit
959b3ca8c8
1 file changed
+26
-32
@@ -345,14 +345,22 @@ class AsyncOmni(OmniBase):
|
||||
):
|
||||
yield output
|
||||
|
||||
logger.debug(f"[{self._name}] All requests completed")
|
||||
|
||||
# Summarize and print stats
|
||||
logger.debug(f"[{self._name}] Request {request_id} finalized at stage-{final_stage_id_for_e2e}")
|
||||
try:
|
||||
# Finalize E2E metrics if not already done
|
||||
if str(request_id) not in metrics.e2e_done:
|
||||
metrics.on_finalize_request(
|
||||
final_stage_id_for_e2e,
|
||||
request_id,
|
||||
_req_start_ts.get(request_id, _wall_start_ts),
|
||||
)
|
||||
|
||||
logger.debug(f"[{self._name}] All requests completed")
|
||||
# Summarize and print stats
|
||||
summary = metrics.build_and_log_summary(final_stage_id_for_e2e)
|
||||
logger.info("[Summary] %s", pformat(summary, sort_dicts=False))
|
||||
except Exception as e:
|
||||
logger.exception(f"[{self._name}] Failed to build/log summary: {e}")
|
||||
logger.exception(f"[{self._name}] Request {request_id} Failed to finalized/build/log summary: {e}")
|
||||
finally:
|
||||
self.request_states.pop(request_id, None)
|
||||
except (asyncio.CancelledError, GeneratorExit):
|
||||
@@ -382,9 +390,11 @@ class AsyncOmni(OmniBase):
|
||||
except asyncio.QueueEmpty:
|
||||
await asyncio.sleep(0.001)
|
||||
continue
|
||||
|
||||
engine_outputs, finished, output_to_yield = self._process_single_result(
|
||||
result, stage, stage_id, metrics, req_start_ts, wall_start_ts, final_stage_id_for_e2e
|
||||
result,
|
||||
stage,
|
||||
stage_id,
|
||||
metrics,
|
||||
)
|
||||
if submit_flag and stage_id == 0:
|
||||
submit_flag = False
|
||||
@@ -421,9 +431,11 @@ class AsyncOmni(OmniBase):
|
||||
while not finished:
|
||||
result = await req_state.queue.get()
|
||||
assert stage_id == req_state.stage_id
|
||||
req_id = result.get("request_id")
|
||||
engine_outputs, finished, output_to_yield = self._process_single_result(
|
||||
result, stage, stage_id, metrics, req_start_ts, wall_start_ts, final_stage_id_for_e2e
|
||||
result,
|
||||
stage,
|
||||
stage_id,
|
||||
metrics,
|
||||
)
|
||||
if output_to_yield:
|
||||
yield output_to_yield
|
||||
@@ -432,7 +444,7 @@ class AsyncOmni(OmniBase):
|
||||
stage.set_engine_outputs(engine_outputs)
|
||||
# Forward to next stage if there is one
|
||||
next_stage_id = stage_id + 1
|
||||
if next_stage_id <= final_stage_id_for_e2e and finished:
|
||||
if next_stage_id <= final_stage_id_for_e2e:
|
||||
next_stage: OmniStage = self.stage_list[next_stage_id]
|
||||
next_inputs = next_stage.process_engine_inputs(self.stage_list, prompt)
|
||||
sp_next: SamplingParams = sampling_params_list[next_stage_id]
|
||||
@@ -447,7 +459,7 @@ class AsyncOmni(OmniBase):
|
||||
connector=connector,
|
||||
stage_id=stage_id,
|
||||
next_stage_id=next_stage_id,
|
||||
req_id=req_id,
|
||||
req_id=request_id,
|
||||
next_inputs=next_inputs,
|
||||
sampling_params=sp_next,
|
||||
original_prompt=prompt,
|
||||
@@ -461,14 +473,14 @@ class AsyncOmni(OmniBase):
|
||||
# because continuing would cause the request to be silently dropped
|
||||
# and the orchestrator to hang waiting for completion.
|
||||
error_msg = (
|
||||
f"[{self._name}] Failed to send request {req_id} to stage-{next_stage_id} via connector. "
|
||||
f"[{self._name}] Failed to send request {request_id} to stage-{next_stage_id} via connector. "
|
||||
"Configure a connector for this edge or inspect connector logs for details."
|
||||
)
|
||||
logger.error(error_msg)
|
||||
raise RuntimeError(error_msg)
|
||||
logger.debug(f"[{self._name}] Forwarded request {req_id} to stage-{next_stage_id}")
|
||||
logger.debug(f"[{self._name}] Forwarded request {request_id} to stage-{next_stage_id}")
|
||||
else:
|
||||
logger.debug(f"[{self._name}] Request {req_id} fully completed")
|
||||
logger.debug(f"[{self._name}] Request {request_id} fully completed")
|
||||
|
||||
def _process_single_result(
|
||||
self,
|
||||
@@ -476,9 +488,6 @@ class AsyncOmni(OmniBase):
|
||||
stage: OmniStage,
|
||||
stage_id: int,
|
||||
metrics: OrchestratorMetrics,
|
||||
req_start_ts: dict[int, float],
|
||||
wall_start_ts: float,
|
||||
final_stage_id_for_e2e: int,
|
||||
) -> tuple[Any, bool, OmniRequestOutput | None]:
|
||||
"""
|
||||
Process a single result dictionary from a stage.
|
||||
@@ -517,23 +526,8 @@ class AsyncOmni(OmniBase):
|
||||
)
|
||||
|
||||
output_to_yield = None
|
||||
|
||||
if getattr(stage, "final_output", False):
|
||||
logger.debug(f"[{self._name}] Request {req_id} finalized at stage-{stage_id}")
|
||||
|
||||
# Finalize request metrics if this is the E2E final stage and it's finished
|
||||
try:
|
||||
rid_key = str(req_id)
|
||||
if stage_id == final_stage_id_for_e2e and rid_key not in metrics.e2e_done and finished:
|
||||
metrics.on_finalize_request(
|
||||
stage_id,
|
||||
req_id,
|
||||
req_start_ts.get(req_id, wall_start_ts),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.exception(
|
||||
f"[{self._name}] Finalize request handling error for req {req_id} at stage {stage_id}: {e}",
|
||||
)
|
||||
|
||||
# Construct output to yield
|
||||
images = []
|
||||
if stage.final_output_type == "image":
|
||||
|
||||
Reference in new issue
Block a user