diff --git a/python_bindings/bindings.cpp b/python_bindings/bindings.cpp index dd09e80a2..220f8cf37 100644 --- a/python_bindings/bindings.cpp +++ b/python_bindings/bindings.cpp @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -326,8 +327,16 @@ class Index { } std::vector> data; - for (auto id : ids) { - data.push_back(appr_alg->template getDataByLabel(id)); + { + // Pure C++ work only (no Python objects touched) -- safe to + // release the GIL for the (potentially large) copy loop, same + // pattern as knnQuery_return_numpy() above. The GIL is + // reacquired when this scope ends, before any py::cast/ + // py::array_t construction below. + py::gil_scoped_release release_for_copy_loop; + for (auto id : ids) { + data.push_back(appr_alg->template getDataByLabel(id)); + } } if (return_type == "list") { return py::cast(data); @@ -720,6 +729,351 @@ class Index { size_t getCurrentCount() const { return appr_alg->cur_element_count; } + + + py::dict checkIntegrity() { + /** + * Python-friendly integrity check that returns detailed results + * instead of crashing on assert failures. + * + * Returns a dict with: + * - valid: bool - whether integrity check passed + * - connections_checked: int - total connections verified + * - min_inbound: int - minimum inbound connections per node + * - max_inbound: int - maximum inbound connections per node + * - errors: list[str] - list of any errors found + */ + if (!appr_alg) { + return py::dict( + "valid"_a = false, + "connections_checked"_a = 0, + "min_inbound"_a = 0, + "max_inbound"_a = 0, + "errors"_a = py::list(py::cast(std::vector{"Index not initialized"})) + ); + } + + std::vector errors; + int connections_checked = 0; + std::vector inbound_connections_num(appr_alg->cur_element_count, 0); + int min_inbound = 0, max_inbound = 0; + + { + // Pure C++ scan only (no Python objects touched) -- safe to + // release the GIL for the (potentially large) connection scan, + // same pattern as knnQuery_return_numpy()/getData() above. The + // GIL is reacquired when this scope ends, before any + // py::list/py::dict construction below. + py::gil_scoped_release release_for_scan; + for (size_t i = 0; i < appr_alg->cur_element_count; i++) { + for (int l = 0; l <= appr_alg->element_levels_[i]; l++) { + hnswlib::linklistsizeint *ll_cur = appr_alg->get_linklist_at_level(i, l); + int size = appr_alg->getListCount(ll_cur); + hnswlib::tableint *data = (hnswlib::tableint *) (ll_cur + 1); + std::unordered_set s; + + for (int j = 0; j < size; j++) { + // Check: connection points to valid element + if (data[j] >= appr_alg->cur_element_count) { + errors.push_back("Element " + std::to_string(i) + " at level " + + std::to_string(l) + " has invalid connection to " + std::to_string(data[j])); + } + // Check: no self-loops + if (data[j] == i) { + errors.push_back("Element " + std::to_string(i) + " at level " + + std::to_string(l) + " has self-loop"); + } + // Track for duplicate check + if (s.find(data[j]) != s.end()) { + errors.push_back("Element " + std::to_string(i) + " at level " + + std::to_string(l) + " has duplicate connection to " + std::to_string(data[j])); + } + s.insert(data[j]); + if (data[j] < appr_alg->cur_element_count) { + inbound_connections_num[data[j]]++; + } + connections_checked++; + } + } + } + + // Check for orphan nodes (no inbound connections) + if (appr_alg->cur_element_count > 1) { + min_inbound = inbound_connections_num[0]; + max_inbound = inbound_connections_num[0]; + for (size_t i = 0; i < appr_alg->cur_element_count; i++) { + if (inbound_connections_num[i] == 0) { + errors.push_back("Element " + std::to_string(i) + " has no inbound connections (orphan)"); + } + min_inbound = std::min(inbound_connections_num[i], min_inbound); + max_inbound = std::max(inbound_connections_num[i], max_inbound); + } + } + } + + py::list error_list; + for (const auto& err : errors) { + error_list.append(err); + } + + return py::dict( + "valid"_a = errors.empty(), + "connections_checked"_a = connections_checked, + "element_count"_a = (size_t)appr_alg->cur_element_count, + "min_inbound"_a = min_inbound, + "max_inbound"_a = max_inbound, + "errors"_a = error_list + ); + } + + + py::dict repairOrphans() { + /** + * Deterministic repair of zero-inbound ("orphan") HNSW nodes. + * See Story #1358 / spike #1330 (docs/research/hnsw-temporal-orphans-1330.md). + * + * For each orphan, forces a back-edge from its own existing level-0 + * neighbors (the same neighbor set chosen by getNeighborsByHeuristic2 + * at insertion time), guaranteeing inclusion even where the original + * construction's heuristic later pruned it out of every neighbor's + * list. + * + * SAFE EVICTION (Messi Rule 13/anti-silent-failure -- do not trade + * one orphan for another): if a neighbor's link list is already at + * maxM0_, repair only evicts an existing entry whose CURRENT total + * inbound-connection count is strictly greater than 1 -- i.e. an + * entry that has at least one OTHER inbound edge and therefore + * cannot become a new orphan as a result of the eviction. An + * earlier "evict the farthest-by-distance" design was measured to + * thrash indefinitely (113k+ evictions, never converging) because + * in a near-tie cluster ALL pairwise distances are nearly equal, + * giving no stable "weakest" signal -- ties keep flipping which + * node looks farthest from one pass to the next. The inbound-count + * guard is a structural (not distance-based) safety criterion that + * is immune to that instability: eviction is skipped (try the next + * anchor instead) whenever no safe candidate exists at the current + * anchor. + * + * Repair iterates in bounded passes (at most cur_element_count + 1 + * -- a provable termination bound, Messi Rule 14), re-scanning + * live inbound counts after every pass, until convergence or until + * a pass makes no further progress (a genuinely stuck residual, + * reported via `valid: false` rather than silently accepted). + * + * Returns a dict with: + * - orphans_before: int - orphan count on the first scan + * - orphans_after: int - orphan count on the final scan + * - repaired_count: int - orphans_before - orphans_after + * - passes_used: int - number of repair passes actually run + * - forced_evictions: int - number of safe weakest-edge evictions performed + * - valid: bool - whether orphans_after == 0 + */ + if (!appr_alg) { + return py::dict( + "orphans_before"_a = (size_t)0, + "orphans_after"_a = (size_t)0, + "repaired_count"_a = (size_t)0, + "passes_used"_a = (size_t)0, + "forced_evictions"_a = (size_t)0, + "valid"_a = false + ); + } + + const size_t n = appr_alg->cur_element_count; + + if (n <= 1) { + return py::dict( + "orphans_before"_a = (size_t)0, + "orphans_after"_a = (size_t)0, + "repaired_count"_a = (size_t)0, + "passes_used"_a = (size_t)0, + "forced_evictions"_a = (size_t)0, + "valid"_a = true + ); + } + + size_t orphans_before = 0; + size_t forced_evictions = 0; + size_t passes_used = 0; + size_t orphans_after = 0; + + { + // Pure C++ scan-and-repair only (no Python objects touched) -- + // safe to release the GIL for the (potentially long-running) + // repair loop, same pattern as knnQuery_return_numpy()/ + // getData()/checkIntegrity() above. The GIL is reacquired when + // this scope ends, before the final py::dict construction below. + // NOTE: orphans_before/forced_evictions/passes_used/orphans_after + // are the OUTER variables declared just above (and returned by + // the py::dict below) -- deliberately NOT redeclared here, so + // the increments/assignments in this block write directly into + // them rather than into a shadowed, discarded local copy. + py::gil_scoped_release release_for_repair; + + const size_t maxM0 = appr_alg->maxM0_; + const size_t max_passes = n + 1; + + std::vector inbound(n, 0); + for (size_t i = 0; i < n; i++) { + for (int l = 0; l <= appr_alg->element_levels_[i]; l++) { + hnswlib::linklistsizeint *ll = appr_alg->get_linklist_at_level((hnswlib::tableint)i, l); + int size = appr_alg->getListCount(ll); + hnswlib::tableint *data = (hnswlib::tableint *) (ll + 1); + for (int j = 0; j < size; j++) { + if ((size_t)data[j] < n) { + inbound[data[j]]++; + } + } + } + } + + for (size_t i = 0; i < n; i++) { + if (inbound[i] == 0) orphans_before++; + } + + for (size_t pass = 0; pass < max_passes; pass++) { + std::vector orphans; + for (size_t i = 0; i < n; i++) { + if (inbound[i] == 0) orphans.push_back((hnswlib::tableint) i); + } + if (orphans.empty()) break; + + passes_used = pass + 1; + bool progress = false; + + // Attempts to connect `o` into `anchor`'s level-0 list: appends + // if there is room, otherwise evicts a SAFE candidate (current + // inbound > 1, so eviction cannot create a new orphan) with the + // highest inbound count. Returns true if `o` was connected. + auto try_connect = [&](hnswlib::tableint o, hnswlib::tableint anchor) -> bool { + if (anchor == o) return false; + // Bounds guard (mirrors the inbound-counting loop's own + // `if ((size_t)data[j] < n)` check above): `anchor` is read + // out of a link list and could in principle be corrupted + // (e.g. a production index with damage beyond simple + // orphans -- an invalid-id connection from a torn write). + // get_linklist0(anchor) indexes data_level0_memory_ by + // anchor unconditionally; an out-of-range anchor would + // write outside cur_element_count. Fail safe: skip it. + if ((size_t) anchor >= n) return false; + + hnswlib::linklistsizeint *ll_anchor = appr_alg->get_linklist0(anchor); + int sz_anchor = appr_alg->getListCount(ll_anchor); + hnswlib::tableint *data_anchor = (hnswlib::tableint *) (ll_anchor + 1); + + for (int j = 0; j < sz_anchor; j++) { + if (data_anchor[j] == o) return false; // already present + } + + if ((size_t) sz_anchor < maxM0) { + data_anchor[sz_anchor] = o; + appr_alg->setListCount(ll_anchor, sz_anchor + 1); + inbound[o]++; + return true; + } + + int victim_idx = -1; + int victim_inbound = 0; + for (int j = 0; j < sz_anchor; j++) { + hnswlib::tableint cand = data_anchor[j]; + // Same guard: `cand` is read out of anchor's link list + // and could be an out-of-range id; `inbound[cand]` would + // otherwise be undefined behavior (out-of-range + // std::vector::operator[]). Skip invalid candidates + // rather than crash or corrupt memory. + if ((size_t) cand >= n) continue; + if (inbound[cand] > 1 && inbound[cand] > victim_inbound) { + victim_inbound = inbound[cand]; + victim_idx = j; + } + } + if (victim_idx >= 0) { + hnswlib::tableint victim = data_anchor[victim_idx]; + data_anchor[victim_idx] = o; + inbound[victim]--; + inbound[o]++; + forced_evictions++; + return true; + } + return false; // no room, no safe eviction candidate here + }; + + for (hnswlib::tableint o : orphans) { + if (inbound[o] > 0) continue; // fixed earlier this pass as a side effect + + hnswlib::linklistsizeint *ll0_o = appr_alg->get_linklist0(o); + int sz0_o = appr_alg->getListCount(ll0_o); + hnswlib::tableint *data0_o = (hnswlib::tableint *) (ll0_o + 1); + + bool connected = false; + for (int j = 0; j < sz0_o && !connected; j++) { + connected = try_connect(o, data0_o[j]); + } + + if (!connected) { + // o's own local neighborhood offered no anchor with + // room or a safe eviction candidate (a fragile + // sub-clique lockup -- measured during Story #1358 + // calibration). Widen the search: a distance-sorted + // scan of the WHOLE graph. Pigeonhole guarantee: total + // inbound edges == total outbound edges == (roughly) + // n * maxM0, far more than n, so some node somewhere + // must have inbound > 1 (or room) -- this scan is + // bounded O(n) and only runs for genuinely stuck + // orphans (rare). + std::vector> by_distance; + by_distance.reserve(n - 1); + for (size_t k = 0; k < n; k++) { + if ((hnswlib::tableint) k == o) continue; + dist_t d = appr_alg->fstdistfunc_( + appr_alg->getDataByInternalId(o), + appr_alg->getDataByInternalId((hnswlib::tableint) k), + appr_alg->dist_func_param_); + by_distance.emplace_back(d, (hnswlib::tableint) k); + } + std::sort(by_distance.begin(), by_distance.end(), + [](const std::pair &a, + const std::pair &b) { + return a.first < b.first; + }); + + for (auto &pr : by_distance) { + if (try_connect(o, pr.second)) { + connected = true; + break; + } + } + } + + if (connected) { + progress = true; + } + } + + if (!progress) { + // Genuinely stuck: no orphan in this pass had any anchor + // with room or a safe eviction candidate. Stop rather than + // burn the remaining pass budget; the final scan below + // reports the true residual instead of silently pretending + // convergence. + break; + } + } + + for (size_t i = 0; i < n; i++) { + if (inbound[i] == 0) orphans_after++; + } + } + + return py::dict( + "orphans_before"_a = orphans_before, + "orphans_after"_a = orphans_after, + "repaired_count"_a = orphans_before - orphans_after, + "passes_used"_a = passes_used, + "forced_evictions"_a = forced_evictions, + "valid"_a = (orphans_after == 0) + ); + } }; template @@ -921,7 +1275,8 @@ PYBIND11_PLUGIN(hnswlib) { py::arg("M") = 16, py::arg("ef_construction") = 200, py::arg("random_seed") = 100, - py::arg("allow_replace_deleted") = false) + py::arg("allow_replace_deleted") = false, + py::call_guard()) .def("knn_query", &Index::knnQuery_return_numpy, py::arg("data"), @@ -935,21 +1290,47 @@ PYBIND11_PLUGIN(hnswlib) { py::arg("num_threads") = -1, py::arg("replace_deleted") = false) .def("get_items", &Index::getData, py::arg("ids") = py::none(), py::arg("return_type") = "numpy") - .def("get_ids_list", &Index::getIdsList) + .def("get_ids_list", &Index::getIdsList, py::call_guard()) .def("set_ef", &Index::set_ef, py::arg("ef")) .def("set_num_threads", &Index::set_num_threads, py::arg("num_threads")) .def("index_file_size", &Index::indexFileSize) - .def("save_index", &Index::saveIndex, py::arg("path_to_index")) + .def("save_index", &Index::saveIndex, py::arg("path_to_index"), + py::call_guard()) .def("load_index", &Index::loadIndex, py::arg("path_to_index"), py::arg("max_elements") = 0, - py::arg("allow_replace_deleted") = false) - .def("mark_deleted", &Index::markDeleted, py::arg("label")) + py::arg("allow_replace_deleted") = false, + py::call_guard()) + .def("mark_deleted", &Index::markDeleted, py::arg("label"), + py::call_guard()) .def("unmark_deleted", &Index::unmarkDeleted, py::arg("label")) - .def("resize_index", &Index::resizeIndex, py::arg("new_size")) + .def("resize_index", &Index::resizeIndex, py::arg("new_size"), + py::call_guard()) .def("get_max_elements", &Index::getMaxElements) .def("get_current_count", &Index::getCurrentCount) + .def("check_integrity", &Index::checkIntegrity, + "Check index integrity and return detailed results.\n\n" + "Returns a dict with:\n" + " - valid: bool - whether integrity check passed\n" + " - connections_checked: int - total connections verified\n" + " - element_count: int - number of elements in index\n" + " - min_inbound: int - minimum inbound connections per node\n" + " - max_inbound: int - maximum inbound connections per node\n" + " - errors: list[str] - list of any errors found\n") + .def("repair_orphans", &Index::repairOrphans, + "Deterministically repair zero-inbound (orphan) HNSW nodes.\n\n" + "Forces a back-edge from each orphan into its own existing\n" + "level-0 neighbors, evicting the weakest existing edge when a\n" + "neighbor's list is full. Idempotent and bounded (at most\n" + "cur_element_count + 1 passes).\n\n" + "Returns a dict with:\n" + " - orphans_before: int - orphan count on the first scan\n" + " - orphans_after: int - orphan count on the final scan\n" + " - repaired_count: int - orphans_before - orphans_after\n" + " - passes_used: int - number of repair passes actually run\n" + " - forced_evictions: int - number of weakest-edge evictions performed\n" + " - valid: bool - whether orphans_after == 0\n") .def_readonly("space", &Index::space_name) .def_readonly("dim", &Index::dim) .def_readwrite("num_threads", &Index::num_threads_default) @@ -1001,8 +1382,10 @@ PYBIND11_PLUGIN(hnswlib) { .def("add_items", &BFIndex::addItems, py::arg("data"), py::arg("ids") = py::none()) .def("delete_vector", &BFIndex::deleteVector, py::arg("label")) .def("set_num_threads", &BFIndex::set_num_threads, py::arg("num_threads")) - .def("save_index", &BFIndex::saveIndex, py::arg("path_to_index")) - .def("load_index", &BFIndex::loadIndex, py::arg("path_to_index"), py::arg("max_elements") = 0) + .def("save_index", &BFIndex::saveIndex, py::arg("path_to_index"), + py::call_guard()) + .def("load_index", &BFIndex::loadIndex, py::arg("path_to_index"), py::arg("max_elements") = 0, + py::call_guard()) .def("__repr__", [](const BFIndex &a) { return ""; }) diff --git a/tests/python/bindings_test_gil_release.py b/tests/python/bindings_test_gil_release.py new file mode 100644 index 000000000..55f027570 --- /dev/null +++ b/tests/python/bindings_test_gil_release.py @@ -0,0 +1,170 @@ +""" +Concurrency regression test: load_index()/save_index() must release the GIL. + +GitHub issue #1437 (LightspeedDMS/code-indexer): the Index.load_index() and +Index.save_index() pybind11 bindings never released the Python GIL during +the native file read + graph (de)serialization. This blocked ALL Python +threads in the process for the full duration of every HNSW shard load +(multi-hundred-MB/GB files over NFS = seconds to tens of seconds in +production), freezing the code-indexer server's Web UI and MCP front door. + +This test builds a real ~300MB on-disk HNSW index, then proves a concurrent +pure-Python "recorder" thread keeps making real wall-clock progress WHILE +load_index() (and, separately, save_index()) runs on the main thread. + +Technique: the recorder thread continuously appends time.monotonic() +timestamps to a list while doing a tiny chunk of pure-Python work between +each append. If the GIL is held for the whole native call (unfixed +binding), the recorder thread cannot run AT ALL during that window -- the +largest gap between two consecutive recorded timestamps will be +approximately equal to the call's duration. If the GIL is released (fixed +binding, py::call_guard()), the recorder keeps +appending timestamps every few milliseconds throughout the call, so the +largest gap stays a small fraction of the call duration. + +This was validated empirically against the unfixed binding before the fix +was applied: max_gap/call_duration measured ~0.90 for load_index() and +~0.79 for save_index() (i.e. the GIL was held for the vast majority of +each call). An Event.wait()-based rate-comparison harness was tried first +and discarded: it can itself stall waiting to reacquire the GIL from the +very thread it is timing, producing unreliable measurements. +""" +import os +import tempfile +import threading +import time +import unittest + +import numpy as np + +import hnswlib + + +class GILReleaseTestCase(unittest.TestCase): + """Prove load_index()/save_index() release the GIL for the native call.""" + + # Tuned (see fork commit message) so save+load each take a meaningful + # fraction of a second on typical CI/dev hardware -- large enough that + # recorder-thread starvation cannot be missed, small enough to keep the + # whole test under ~30s wall clock. + NUM_ELEMENTS = 500_000 + DIM = 128 + + # A stalled (unfixed) binding produces a gap ratio near 0.8-0.9; a fixed + # binding produces a gap ratio of a few percent. 0.5 sits comfortably + # between the two, with margin on both sides. + MAX_GAP_RATIO = 0.5 + + @classmethod + def setUpClass(cls): + rng = np.random.default_rng(42) + data = np.float32(rng.random((cls.NUM_ELEMENTS, cls.DIM))) + + index = hnswlib.Index(space="l2", dim=cls.DIM) + index.init_index(max_elements=cls.NUM_ELEMENTS, ef_construction=40, M=8) + index.set_num_threads(8) + index.add_items(data) + + fd, cls.index_path = tempfile.mkstemp(suffix=".bin") + os.close(fd) + index.save_index(cls.index_path) + cls.built_index = index # reused by the save_index() half of the test + + @classmethod + def tearDownClass(cls): + if os.path.exists(cls.index_path): + os.remove(cls.index_path) + + def _max_recorder_gap(self, blocking_fn, *, min_call_seconds, warmup=0.05, cooldown=0.15): + """Run `blocking_fn()` on the main thread while a background + "recorder" thread continuously appends monotonic timestamps. + + Returns (max_gap_seconds, call_duration_seconds). + """ + stop_flag = threading.Event() + timestamps = [] + + def _recorder(): + while not stop_flag.is_set(): + for _ in range(200): + pass + timestamps.append(time.monotonic()) + + thread = threading.Thread(target=_recorder) + thread.start() + time.sleep(warmup) # let the recorder get running before the timed call + + call_start = time.monotonic() + blocking_fn() + call_duration = time.monotonic() - call_start + + time.sleep(cooldown) # gather a few post-call samples too + stop_flag.set() + thread.join(timeout=60) + self.assertFalse(thread.is_alive(), "recorder thread never stopped") + + self.assertGreaterEqual( + call_duration, + min_call_seconds, + "native call completed too fast to meaningfully exercise the " + "GIL-release behavior under test -- increase NUM_ELEMENTS", + ) + + ordered = sorted(timestamps) + self.assertGreaterEqual( + len(ordered), 2, "recorder thread produced too few samples to measure a gap" + ) + gaps = [b - a for a, b in zip(ordered, ordered[1:])] + return max(gaps), call_duration + + def test_load_index_releases_gil_during_native_call(self): + """A concurrent Python recorder thread must keep making real progress + while load_index() runs on the main thread -- proving the GIL was + released for the native file-read + deserialize duration.""" + + def _do_load(): + fresh_index = hnswlib.Index(space="l2", dim=self.DIM) + fresh_index.load_index(self.index_path, max_elements=self.NUM_ELEMENTS) + + max_gap, call_duration = self._max_recorder_gap( + _do_load, min_call_seconds=0.2 + ) + + gap_ratio = max_gap / call_duration + self.assertLess( + gap_ratio, + self.MAX_GAP_RATIO, + f"recorder thread was silent for {gap_ratio:.1%} of load_index()'s " + f"{call_duration:.3f}s duration (max_gap={max_gap:.3f}s) -- the GIL " + "was likely held for the whole native call (regression of the " + "#1437 fix)", + ) + + def test_save_index_releases_gil_during_native_call(self): + """Same proof as above, for save_index().""" + save_path = tempfile.mktemp(suffix=".bin") + try: + + def _do_save(): + self.built_index.save_index(save_path) + + max_gap, call_duration = self._max_recorder_gap( + _do_save, min_call_seconds=0.05 + ) + + gap_ratio = max_gap / call_duration + self.assertLess( + gap_ratio, + self.MAX_GAP_RATIO, + f"recorder thread was silent for {gap_ratio:.1%} of " + f"save_index()'s {call_duration:.3f}s duration " + f"(max_gap={max_gap:.3f}s) -- the GIL was likely held for " + "the whole native call (regression of the #1437 fix)", + ) + finally: + if os.path.exists(save_path): + os.remove(save_path) + + +if __name__ == "__main__": + unittest.main()