chore(connector): remove stale load and benchmark helpers

benchmarks/utils/metrics.py

@@ -7,110 +7,6 @@
 from typing import Any
 
 
-def correctness_check(
-    name: str,
-    cold_outputs: list[Any],
-    warm_outputs: list[Any],
-    prompts: list[list[int]],
-    max_num_seqs: int,
-) -> dict[str, Any]:
-    """Compare cold vs warm generated output exactly.
-
-    Args:
-        name: System label used in diagnostics.
-        cold_outputs: Outputs from the cold pass.
-        warm_outputs: Outputs from the warm pass.
-        prompts: Tokenized benchmark inputs in output order.
-        max_num_seqs: vLLM admission limit used for mismatch diagnostics.
-
-    Returns:
-        Correctness counters. Only exact generated text and token-ID matches
-        are accepted.
-
-    Thread-safety:
-        Pure function apart from reading output objects.
-    """
-    del name
-    mismatches = 0
-    mismatch_indices: list[int] = []
-    mismatch_details: list[dict[str, Any]] = []
-    prompt_alignment_mismatches = 0
-    total = len(cold_outputs)
-    for i, (cold, warm) in enumerate(zip(cold_outputs, warm_outputs, strict=False)):
-        cold_prompt = list(getattr(cold, "prompt_token_ids", prompts[i]))
-        warm_prompt = list(getattr(warm, "prompt_token_ids", prompts[i]))
-        if cold_prompt != warm_prompt or cold_prompt != list(prompts[i]):
-            prompt_alignment_mismatches += 1
-        if _generated_token_ids(cold) == _generated_token_ids(warm) and _output_text(
-            cold
-        ) == _output_text(warm):
-            continue
-        mismatches += 1
-        mismatch_indices.append(i)
-        mismatch_details.append(
-            {
-                "index": i,
-                "wave": i // max(1, max_num_seqs),
-                "position": i % max(1, max_num_seqs),
-                "prompt_tokens": len(prompts[i]),
-                "cold_token_ids": _generated_token_ids(cold),
-                "warm_token_ids": _generated_token_ids(warm),
-                "cold_text": _output_text(cold),
-                "warm_text": _output_text(warm),
-            }
-        )
-    return {
-        "mismatches": mismatches,
-        "total": total,
-        "indices": mismatch_indices,
-        "mismatch_details": mismatch_details,
-        "prompt_alignment_mismatches": prompt_alignment_mismatches,
-    }
-
-
-def correctness_check_with_visibility(
-    name: str,
-    cold_outputs: list[Any],
-    warm_outputs: list[Any],
-    prompts: list[list[int]],
-    max_num_seqs: int,
-    visible_mask: list[bool],
-) -> dict[str, Any]:
-    """Compare cold/warm outputs and split exact mismatches by visible hits.
-
-    Args:
-        name: System label used in diagnostics.
-        cold_outputs: Outputs from the cold pass.
-        warm_outputs: Outputs from the warm pass.
-        prompts: Tokenized benchmark inputs in output order.
-        max_num_seqs: vLLM admission limit used for mismatch diagnostics.
-        visible_mask: Per-prompt cache-visibility mask.
-
-    Returns:
-        Correctness counters including visible-hit mismatch counters.
-
-    Thread-safety:
-        Pure function apart from reading output objects.
-    """
-    result = correctness_check(name, cold_outputs, warm_outputs, prompts, max_num_seqs)
-    visible_total = 0
-    visible_mismatches = 0
-    for cold, warm, visible in zip(
-        cold_outputs, warm_outputs, visible_mask, strict=False
-    ):
-        if not visible:
-            continue
-        visible_total += 1
-        if _generated_token_ids(cold) == _generated_token_ids(warm) and _output_text(
-            cold
-        ) == _output_text(warm):
-            continue
-        visible_mismatches += 1
-    result["visible_mismatches"] = visible_mismatches
-    result["visible_total"] = visible_total
-    return result
-
-
 def contains_accuracy(
     results: list[Any], answers_by_id: dict[int, list[str]]
 ) -> float | None:
@@ -363,17 +259,3 @@ def _first_number(value: dict[str, Any], keys: tuple[str, ...]) -> float | None:
         if isinstance(raw, int | float):
             return float(raw)
     return None
-
-
-def _generated_token_ids(output: Any) -> list[int]:
-    """Return generated token IDs from a vLLM RequestOutput-like object."""
-    if not getattr(output, "outputs", None):
-        return []
-    return [int(token_id) for token_id in getattr(output.outputs[0], "token_ids", [])]
-
-
-def _output_text(output: Any) -> str:
-    """Return generated text from a vLLM RequestOutput-like object."""
-    if not getattr(output, "outputs", None):
-        return ""
-    return str(getattr(output.outputs[0], "text", ""))

benchmarks/utils/servers.py

@@ -399,25 +399,6 @@ def _daser_server_command(self) -> list[str]:
             cmd.extend(["--l2-size", str(self.l2_size_bytes)])
         return cmd
 
-    async def stop_all(self) -> None:
-        """Terminate all child processes."""
-        for proc in reversed(self._procs):
-            if proc.poll() is not None:
-                continue
-            proc.terminate()
-        deadline = time.monotonic() + 15.0
-        for proc in self._procs:
-            if proc.poll() is not None:
-                continue
-            try:
-                proc.wait(timeout=max(0.1, deadline - time.monotonic()))
-            except subprocess.TimeoutExpired:
-                pass
-        for proc in self._procs:
-            if proc.poll() is None:
-                proc.kill()
-        self._procs.clear()
-
     async def _start_vllm(
         self,
         log_name: str,

daser/connector/daser_connector.py

@@ -145,7 +145,6 @@ def __init__(
             self._pending_loads: dict[str, Any] = {}
             self._invalid_load_block_ids: set[int] = set()
             self._load_request_queue = None
-            self._load_request_dispatcher = None
             self._load_request_queue_lock = threading.Lock()
             self._load_request_dispatcher_future = None
             self._load_loop = asyncio.new_event_loop()

daser/connector/scheduler.py

@@ -815,18 +815,6 @@ def has_pending_store(self, req_id: str) -> bool:
         """
         return req_id in self._pending_stores
 
-    def count_pending_stores_for_request(self, req_id: str) -> int:
-        """Return number of synthetic slot stores pending for a request.
-
-        Args:
-            req_id: base vLLM request ID.
-
-        Returns:
-            Count of pending slot-store entries.
-        """
-        prefix = f"{req_id}:store:"
-        return len([key for key in self._pending_stores if key.startswith(prefix)])
-
     def drop_pending_alloc(self, req_id: str) -> None:
         """Remove pending allocation state for a request.
 

daser/connector/staging.py

@@ -122,7 +122,6 @@ def __init__(
             raise ValueError("buffer_bytes must be positive")
         if depth <= 0:
             raise ValueError("depth must be positive")
-        self._device = device
         self._buffer_bytes = buffer_bytes
         self._buffers: list[torch.Tensor] = [
             torch.empty(buffer_bytes, dtype=torch.uint8, device=device)
@@ -438,41 +437,6 @@ def apply_rope_delta_to_kv_key_block(
     )
 
 
-def _transform_loaded_kv_batch(
-    layer_batch: torch.Tensor,
-    load_key_scale: float,
-    load_value_scale: float,
-    pos_offset: int,
-    rope_delta_scale: float,
-    rope_base: float,
-    rope_rotary_dim: int,
-    rope_is_neox_style: bool,
-) -> None:
-    """Apply load-time scaling and RoPE relocation to a batch of KV blocks."""
-    if layer_batch.dim() < 2 or layer_batch.shape[1] < 2:
-        return
-    if load_key_scale != 1.0:
-        layer_batch[:, 0].mul_(load_key_scale)
-    if load_value_scale != 1.0:
-        layer_batch[:, 1].mul_(load_value_scale)
-    if (
-        not pos_offset
-        or layer_batch.dim() != 5
-        or rope_rotary_dim <= 0
-        or layer_batch.shape[-1] < rope_rotary_dim
-    ):
-        return
-    key_batch = layer_batch[:, 0].contiguous()
-    apply_rope_delta_to_key_block(
-        key_batch,
-        delta=round(pos_offset * rope_delta_scale),
-        rope_base=rope_base,
-        rotary_dim=rope_rotary_dim,
-        is_neox_style=rope_is_neox_style,
-    )
-    layer_batch[:, 0].copy_(key_batch)
-
-
 def _transform_loaded_staging_batch(
     staging_by_layer: torch.Tensor,
     layer_sample: torch.Tensor,