Merge remote-tracking branch 'origin/develop' into ucm-vllm-connector-metrics

# Conflicts:
#	ucm/integration/vllm/hma_connector.py

Author: dante159753

examples/metrics/metrics_configs.yaml

@@ -38,6 +38,22 @@ counter:
     documentation: "Total shards dispatched by Cache dump (mirror of cache_load_shards_total)"
   - name: "cache_dump_backend_shards_total"
     documentation: "Shards actually pushed to backend on dump (excludes !handle.Owner() skips in shared-buffer scenario)"
+  - name: "cache_load_queue_full_total"
+    documentation: "Number of Cache load submissions rejected because the waiting queue was full"
+  - name: "cache_dump_queue_full_total"
+    documentation: "Number of Cache dump submissions rejected because the waiting queue was full"
+  - name: "cache_backend_load_submit_errors_total"
+    documentation: "Number of Cache load backend submit failures"
+  - name: "cache_backend_load_wait_errors_total"
+    documentation: "Number of Cache load backend wait failures"
+  - name: "cache_backend_dump_submit_errors_total"
+    documentation: "Number of Cache dump backend submit failures"
+  - name: "cache_backend_dump_wait_errors_total"
+    documentation: "Number of Cache dump backend wait failures"
+  - name: "cache_h2d_errors_total"
+    documentation: "Number of Cache host-to-device transfer or sync failures"
+  - name: "cache_d2h_errors_total"
+    documentation: "Number of Cache device-to-host transfer, event wait, or sync failures"
 
   # -- Pipeline Cache stage: per-worker real throughput counters ------------
   # Accumulated per completed Cache task with size = shardSize * num_shards.
@@ -58,6 +74,14 @@ counter:
     documentation: "Total bytes transferred from posix storage to host buffer (load path, summed per completed task)"
   - name: "posix_h2s_bytes_total"
     documentation: "Total bytes transferred from host buffer to posix storage (dump path, summed per completed task)"
+  - name: "posix_aio_timeout_total"
+    documentation: "Number of Posix AIO task or submit timeouts"
+  - name: "posix_io_timeout_total"
+    documentation: "Number of Posix synchronous worker task timeouts"
+  - name: "posix_open_errors_total"
+    documentation: "Number of Posix open failures"
+  - name: "posix_io_errors_total"
+    documentation: "Number of Posix read, write, or AIO completion failures"
 
   # -- Connector layer: per-worker real throughput counters ----------------
   # The connector load_speed / save_speed histograms sample per-call speed
@@ -68,6 +92,20 @@ counter:
     documentation: "Total bytes loaded through the UCM connector (summed across all start_load_kv calls)"
   - name: "save_bytes_total"
     documentation: "Total bytes saved through the UCM connector (summed across all wait_for_save calls)"
+  - name: "connector_lookup_errors_total"
+    documentation: "Number of connector lookup errors treated as cache misses"
+  - name: "connector_load_submit_errors_total"
+    documentation: "Number of connector load submit failures"
+  - name: "connector_load_wait_errors_total"
+    documentation: "Number of connector load wait failures"
+  - name: "connector_load_invalid_requests_total"
+    documentation: "Number of connector load failure events that invalidated request blocks"
+  - name: "connector_load_invalid_blocks_total"
+    documentation: "Number of newly invalidated vLLM block ids caused by connector load failures"
+  - name: "connector_dump_submit_errors_total"
+    documentation: "Number of connector dump submit failures"
+  - name: "connector_dump_wait_errors_total"
+    documentation: "Number of connector dump wait failures"
 
 # Gauge metrics configuration
 gauge:

test/test_ucm_connector_metrics.py

@@ -370,13 +370,7 @@ def _install_stubs():
 
 _install_stubs()
 
-import ucm.integration.vllm.hma_connector as hma_connector_module
 import ucm.integration.vllm.ucm_connector as ucm_connector_module
-from ucm.integration.vllm.hma_connector import (
-    FAWARequestDispatchMeta,
-    UCMFAWAConnector,
-    UCMFAWAConnectorMetadata,
-)
 from ucm.integration.vllm.metrics import UCMConnectorStats, UCMPromMetrics
 from ucm.integration.vllm.ucm_connector import (
     PendingDumpTask,
@@ -482,38 +476,6 @@ def _reset_fakes():
     dispatcher_module._DISPATCHER = None
 
 
-class FakeHMAStore:
-    def __init__(self, label, fail_wait=False):
-        self.label = label
-        self.fail_wait = fail_wait
-        self.loaded = []
-        self.dumped = []
-        self.waited = []
-
-    def load_data(self, keys, shard_indices, ptrs):
-        task = f"{self.label}-load-{len(keys)}-{len(self.loaded)}"
-        self.loaded.append((list(keys), list(shard_indices), ptrs, task))
-        return task
-
-    def dump_data(self, keys, shard_indices, ptrs, event_handle):
-        task = f"{self.label}-dump-{len(keys)}-{len(self.dumped)}"
-        self.dumped.append((list(keys), list(shard_indices), ptrs, event_handle, task))
-        return task
-
-    def wait(self, task):
-        self.waited.append(task)
-        if self.fail_wait:
-            raise RuntimeError(f"{self.label} wait failed")
-
-
-class FakeHMADevice:
-    def __init__(self):
-        self.destroyed = []
-
-    def destroy_event_handle(self, event_handle):
-        self.destroyed.append(event_handle)
-
-
 def test_config_definitions_register_enable_list_and_metric_names():
     _reset_fakes()
     config = _metrics_config()
@@ -1023,219 +985,6 @@ def destroy_event_handle(self, event_handle):
     ]
 
 
-def _fawa_request(load_keys=None, dump_keys=None):
-    return FAWARequestDispatchMeta(
-        load_keys=list(load_keys or []),
-        load_hash_start=0,
-        load_hash_end=len(load_keys or []),
-        load_vllm_block_ids=([1, 2],),
-        dump_keys=list(dump_keys or []),
-        dump_hash_start=0,
-        dump_hash_end=len(dump_keys or []),
-        dump_vllm_block_ids=([1, 2],),
-    )
-
-
-def _new_hma_connector(metadata, fa_store=None, wa_store=None):
-    connector = object.__new__(UCMFAWAConnector)
-    connector.fa_store = fa_store or FakeHMAStore("FA")
-    connector.wa_store = wa_store or FakeHMAStore("WA")
-    connector._fawa_store_block_sizes = {"FA": 100, "WA": 20}
-    connector._invalid_block_ids = set()
-    connector.tp_size = 1
-    connector.tp_rank = 0
-    connector.wa_dump_block_wise = True
-    connector.fa_group_ids = set()
-    connector.window_group_ids = set()
-    connector.group_metas = []
-    connector.tp_dump_tasks = {}
-    connector._direct_task_has_load = False
-    connector._direct_task_load_duration_ms = 0.0
-    connector._direct_task_compute_start_ms = None
-    connector.device = FakeHMADevice()
-    connector._get_connector_metadata = lambda: metadata
-    connector._extract_fa_ptr = lambda *args, **kwargs: [[0], [0]]
-    connector._extract_wa_ptr = lambda *args, **kwargs: [[0], [0]]
-    return connector
-
-
-def test_hma_connector_records_load_metrics(monkeypatch):
-    _reset_fakes()
-    times = iter([1.0, 1.1, 1.11])
-    monkeypatch.setattr(
-        hma_connector_module.time,
-        "perf_counter",
-        lambda: next(times),
-    )
-    metadata = UCMFAWAConnectorMetadata(
-        request_meta={"req-1": _fawa_request(load_keys=[b"a", b"b"])}
-    )
-    connector = _new_hma_connector(metadata)
-
-    connector.start_load_kv(None)
-
-    stats = fake_ucmmetrics.updated[-1]
-    assert stats["load_requests_num"] == 1
-    assert stats["load_blocks_num"] == 3
-    assert stats["load_duration"] == pytest.approx(100.0)
-    assert stats["load_bytes_total"] == 220
-    assert stats["load_speed"] == pytest.approx(220 / 100 / 1024 / 1024)
-    assert connector._direct_task_has_load is True
-    assert connector._direct_task_load_duration_ms == pytest.approx(100.0)
-    assert connector._direct_task_compute_start_ms == pytest.approx(1110.0)
-
-
-def test_hma_connector_load_metrics_ignore_wait_failure_for_counting(monkeypatch):
-    _reset_fakes()
-    times = iter([1.0, 1.1, 1.11])
-    monkeypatch.setattr(
-        hma_connector_module.time,
-        "perf_counter",
-        lambda: next(times),
-    )
-    metadata = UCMFAWAConnectorMetadata(
-        request_meta={"req-1": _fawa_request(load_keys=[b"a", b"b"])}
-    )
-    connector = _new_hma_connector(metadata, wa_store=FakeHMAStore("WA", True))
-
-    connector.start_load_kv(None)
-
-    stats = fake_ucmmetrics.updated[-1]
-    assert stats["load_requests_num"] == 1
-    assert stats["load_blocks_num"] == 3
-    assert stats["load_bytes_total"] == 220
-    assert stats["load_speed"] == pytest.approx(220 / 100 / 1024 / 1024)
-    assert connector._invalid_block_ids == {1, 2}
-
-
-@pytest.mark.parametrize("drain_method", ["get_finished", "preemption"])
-def test_hma_connector_records_mixed_fa_wa_save_completion_metrics(
-    monkeypatch, drain_method
-):
-    _reset_fakes()
-    times = iter([1.0, 1.01, 1.02, 1.10, 1.12, 1.13, 1.15])
-    monkeypatch.setattr(
-        hma_connector_module.time,
-        "perf_counter",
-        lambda: next(times),
-    )
-    metadata = UCMFAWAConnectorMetadata(
-        request_meta={"req-1": _fawa_request(dump_keys=[b"a", b"b"])}
-    )
-    connector = _new_hma_connector(metadata)
-    handles = iter([11, 12])
-    connector._get_dump_event_handle = lambda: next(handles)
-    connector._direct_task_has_load = True
-    connector._direct_task_load_duration_ms = 10.0
-    connector._direct_task_compute_start_ms = 900.0
-
-    connector.wait_for_save()
-
-    assert fake_ucmmetrics.updated == [
-        {
-            "save_requests_num": 1,
-            "save_blocks_num": 4,
-            "save_bytes_total": 240,
-        }
-    ]
-    if drain_method == "get_finished":
-        connector.get_finished({"req-1"})
-    else:
-        connector.handle_preemptions(
-            UCMFAWAConnectorMetadata(preempted_req_ids={"req-1"})
-        )
-
-    assert fake_ucmmetrics.updated[1] == {
-        "save_duration": pytest.approx(110.0),
-        "save_speed": pytest.approx(200 / 110 / 1024 / 1024),
-        "save_completion_wait_duration": pytest.approx(20.0),
-        "save_completion_wait_speed": pytest.approx(200 / 20 / 1024 / 1024),
-    }
-    assert fake_ucmmetrics.updated[2] == {
-        "connector_task_load_stage_load_dump_ms": 10.0,
-        "connector_task_compute_stage_load_dump_ms": pytest.approx(100.0),
-        "connector_task_dump_stage_load_dump_ms": pytest.approx(110.0),
-        "connector_task_total_load_dump_ms": pytest.approx(220.0),
-    }
-    assert fake_ucmmetrics.updated[3] == {
-        "save_duration": pytest.approx(130.0),
-        "save_speed": pytest.approx(40 / 130 / 1024 / 1024),
-        "save_completion_wait_duration": pytest.approx(20.0),
-        "save_completion_wait_speed": pytest.approx(40 / 20 / 1024 / 1024),
-    }
-    assert fake_ucmmetrics.updated[4] == {
-        "connector_task_load_stage_load_dump_ms": 10.0,
-        "connector_task_compute_stage_load_dump_ms": pytest.approx(100.0),
-        "connector_task_dump_stage_load_dump_ms": pytest.approx(130.0),
-        "connector_task_total_load_dump_ms": pytest.approx(240.0),
-    }
-    assert connector.device.destroyed == [11, 12]
-    assert connector.tp_dump_tasks == {}
-
-
-def test_hma_connector_save_metrics_count_only_submitted_requests_without_changing_tp_keys(
-    monkeypatch,
-):
-    _reset_fakes()
-    times = iter([1.0, 1.01, 1.02])
-    monkeypatch.setattr(
-        hma_connector_module.time,
-        "perf_counter",
-        lambda: next(times),
-    )
-    metadata = UCMFAWAConnectorMetadata(
-        request_meta={
-            "req-no-rank-work": _fawa_request(dump_keys=[b"a"]),
-            "req-rank-work": _fawa_request(dump_keys=[b"b", b"c"]),
-        }
-    )
-    connector = _new_hma_connector(metadata)
-    connector.tp_size = 2
-    connector.tp_rank = 0
-    handles = iter([11, 12])
-    connector._get_dump_event_handle = lambda: next(handles)
-
-    connector.wait_for_save()
-
-    assert list(connector.tp_dump_tasks) == [("req-no-rank-work", "req-rank-work")]
-    assert fake_ucmmetrics.updated == [
-        {
-            "save_requests_num": 1,
-            "save_blocks_num": 2,
-            "save_bytes_total": 120,
-        }
-    ]
-
-
-def test_hma_connector_no_dump_records_load_only_breakdown(monkeypatch):
-    _reset_fakes()
-    times = iter([1.0])
-    monkeypatch.setattr(
-        hma_connector_module.time,
-        "perf_counter",
-        lambda: next(times),
-    )
-    metadata = UCMFAWAConnectorMetadata(
-        request_meta={"req-1": _fawa_request(load_keys=[b"a"])}
-    )
-    connector = _new_hma_connector(metadata)
-    connector._direct_task_has_load = True
-    connector._direct_task_load_duration_ms = 7.0
-    connector._direct_task_compute_start_ms = 900.0
-
-    connector.wait_for_save()
-
-    task_stats = fake_ucmmetrics.updated[-1]
-    assert task_stats["connector_task_load_stage_load_only_ms"] == 7.0
-    assert task_stats["connector_task_compute_stage_load_only_ms"] == pytest.approx(
-        100.0
-    )
-    assert task_stats["connector_task_total_load_only_ms"] == pytest.approx(107.0)
-    assert connector._direct_task_has_load is False
-    assert connector._direct_task_load_duration_ms == 0.0
-    assert connector._direct_task_compute_start_ms is None
-
-
 def test_multiproc_logger_uses_prefix_and_dispatcher_snapshot(tmp_path):
     _reset_fakes()
     import ucm.observability as observability

ucm/integration/mindie/hash_mindie/pyproject.toml

@@ -0,0 +1,3 @@
+[build-system]
+requires = ["setuptools>=64", "wheel", "pybind11>=2.10"]
+build-backend = "setuptools.build_meta"

ucm/integration/mindie/hash_mindie/setup.py

@@ -0,0 +1,19 @@
+from pybind11.setup_helpers import Pybind11Extension, build_ext
+from setuptools import setup
+
+ext_modules = [
+    Pybind11Extension(
+        "uc_hash_ext",
+        ["uc_hash_ext.cpp"],
+        cxx_std=17,
+        extra_compile_args=["-O3", "-march=native"],
+        extra_link_args=["-Wl,-z,relro,-z,now", "-Wl,-s"],
+    )
+]
+
+setup(
+    name="uc_hash_ext",
+    version="0.0.1",
+    ext_modules=ext_modules,
+    cmdclass={"build_ext": build_ext},
+)