240 lines
8.3 KiB
Python
240 lines
8.3 KiB
Python
# Copyright 2025-present the zvec project
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
"""Tests to verify that the GIL is released during native C++ query calls,
|
|
enabling true thread-level concurrency for multi-threaded Python applications."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
|
import pytest
|
|
import zvec
|
|
from zvec import (
|
|
Collection,
|
|
CollectionOption,
|
|
DataType,
|
|
Doc,
|
|
FieldSchema,
|
|
HnswIndexParam,
|
|
Query,
|
|
VectorSchema,
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def gil_test_collection(tmp_path_factory) -> Collection:
|
|
"""Create a collection with enough data to make queries take measurable time."""
|
|
schema = zvec.CollectionSchema(
|
|
name="gil_test",
|
|
fields=[
|
|
FieldSchema("id", DataType.INT64, nullable=False),
|
|
],
|
|
vectors=[
|
|
VectorSchema(
|
|
"vec",
|
|
DataType.VECTOR_FP32,
|
|
dimension=128,
|
|
index_param=HnswIndexParam(),
|
|
),
|
|
],
|
|
)
|
|
option = CollectionOption(read_only=False, enable_mmap=True)
|
|
temp_dir = tmp_path_factory.mktemp("zvec_gil_test")
|
|
collection_path = temp_dir / "gil_test_collection"
|
|
|
|
coll = zvec.create_and_open(path=str(collection_path), schema=schema, option=option)
|
|
|
|
# Insert enough docs to make queries non-trivial
|
|
docs = [
|
|
Doc(
|
|
id=str(i),
|
|
fields={"id": i},
|
|
vectors={"vec": [float(i % 100) + 0.1 * j for j in range(128)]},
|
|
)
|
|
for i in range(500)
|
|
]
|
|
result = coll.insert(docs)
|
|
for r in result:
|
|
assert r.ok()
|
|
|
|
yield coll
|
|
|
|
try:
|
|
coll.destroy()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class TestGILRelease:
|
|
"""Verify that C++ query calls release the GIL, allowing true thread concurrency."""
|
|
|
|
def test_gil_released_during_query(self, gil_test_collection: Collection):
|
|
"""Prove the GIL is explicitly released during C++ Query calls.
|
|
|
|
Strategy:
|
|
- Set switch_interval to 0.5s (100x the default 5ms). This means CPython's
|
|
involuntary GIL switching will NOT occur for 500ms after a thread acquires.
|
|
- Run queries that complete in total < 500ms (about 100-200ms).
|
|
- A background thread (using time.sleep(0) to avoid deadlock) counts how many
|
|
times it got to run.
|
|
- Since total query time < switch_interval, the bg thread can ONLY run if
|
|
the C++ code explicitly releases the GIL.
|
|
- Reset counter just before queries; check counter > 0 after queries.
|
|
"""
|
|
old_interval = sys.getswitchinterval()
|
|
# 500ms - much longer than the total query time (~100-200ms)
|
|
sys.setswitchinterval(0.5)
|
|
|
|
try:
|
|
counter = {"value": 0}
|
|
stop_event = threading.Event()
|
|
|
|
def background_counter():
|
|
while not stop_event.is_set():
|
|
counter["value"] += 1
|
|
time.sleep(0) # Yield GIL to prevent deadlock
|
|
|
|
bg_thread = threading.Thread(target=background_counter, daemon=True)
|
|
bg_thread.start()
|
|
|
|
# Let bg thread start (sleep releases GIL)
|
|
time.sleep(0.05)
|
|
|
|
# --- Critical section: reset counter, run queries, capture counter ---
|
|
counter["value"] = 0
|
|
|
|
query_vec = [1.0] * 128
|
|
start = time.monotonic()
|
|
for _ in range(100):
|
|
gil_test_collection.query(
|
|
Query(field_name="vec", vector=query_vec),
|
|
topk=100,
|
|
)
|
|
elapsed = time.monotonic() - start
|
|
|
|
count_during_queries = counter["value"]
|
|
# --- End critical section ---
|
|
|
|
stop_event.set()
|
|
time.sleep(0.01)
|
|
bg_thread.join(timeout=5)
|
|
|
|
print(f"\nQuery elapsed: {elapsed:.4f}s (switch_interval=0.5s)")
|
|
print(f"Counter during queries: {count_during_queries}")
|
|
|
|
# Verify queries completed within the switch_interval window.
|
|
# If they did, the ONLY way bg thread could run is via explicit GIL release.
|
|
assert elapsed < 0.5, (
|
|
f"Queries took {elapsed:.3f}s >= switch_interval (0.5s). "
|
|
"Test is inconclusive; increase switch_interval or reduce query count."
|
|
)
|
|
assert count_during_queries > 0, (
|
|
"Background thread could not run during C++ execution despite "
|
|
"query time < switch_interval. GIL was NOT released."
|
|
)
|
|
finally:
|
|
sys.setswitchinterval(old_interval)
|
|
|
|
def test_parallel_queries_correctness(self, gil_test_collection: Collection):
|
|
"""Verify parallel queries return correct results and print timing info.
|
|
|
|
NOTE: The definitive proof of GIL release is test_gil_released_during_query
|
|
(counter + setswitchinterval). This test focuses on parallel correctness and
|
|
logs timing for manual inspection, since CI timing is too noisy for assertions.
|
|
"""
|
|
num_queries = 1000
|
|
query_vec = [1.0] * 128
|
|
|
|
def do_query():
|
|
return gil_test_collection.query(
|
|
Query(field_name="vec", vector=query_vec),
|
|
topk=100,
|
|
)
|
|
|
|
# Serial execution (baseline)
|
|
start_serial = time.monotonic()
|
|
for _ in range(num_queries):
|
|
do_query()
|
|
serial_time = time.monotonic() - start_serial
|
|
|
|
# Parallel execution
|
|
num_workers = os.cpu_count() or 2
|
|
start_parallel = time.monotonic()
|
|
with ThreadPoolExecutor(max_workers=num_workers) as executor:
|
|
futures = [executor.submit(do_query) for _ in range(num_queries)]
|
|
for future in as_completed(futures):
|
|
result = future.result()
|
|
assert len(result) > 0
|
|
parallel_time = time.monotonic() - start_parallel
|
|
|
|
print(f"\nSerial time: {serial_time:.4f}s, Parallel time: {parallel_time:.4f}s")
|
|
print(
|
|
f"Speedup ratio: {serial_time / parallel_time:.2f}x (workers={num_workers})"
|
|
)
|
|
|
|
def test_thread_safety_concurrent_queries(self, gil_test_collection: Collection):
|
|
"""Verify no crashes or data corruption under concurrent query load."""
|
|
num_threads = 8
|
|
queries_per_thread = 10
|
|
errors = []
|
|
|
|
def worker(thread_id):
|
|
try:
|
|
for i in range(queries_per_thread):
|
|
vec = [float(thread_id + i) + 0.1 * j for j in range(128)]
|
|
result = gil_test_collection.query(
|
|
Query(field_name="vec", vector=vec),
|
|
topk=10,
|
|
)
|
|
assert len(result) > 0
|
|
except Exception as e:
|
|
errors.append((thread_id, e))
|
|
|
|
threads = [
|
|
threading.Thread(target=worker, args=(tid,)) for tid in range(num_threads)
|
|
]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join(timeout=60)
|
|
|
|
assert len(errors) == 0, f"Errors in threads: {errors}"
|
|
|
|
def test_concurrent_fetch_release_gil(self, gil_test_collection: Collection):
|
|
"""Verify Fetch operations also release the GIL correctly."""
|
|
num_threads = 4
|
|
errors = []
|
|
|
|
def worker(thread_id):
|
|
try:
|
|
ids = [str(i) for i in range(thread_id * 10, thread_id * 10 + 10)]
|
|
result = gil_test_collection.fetch(ids)
|
|
assert len(result) > 0
|
|
except Exception as e:
|
|
errors.append((thread_id, e))
|
|
|
|
threads = [
|
|
threading.Thread(target=worker, args=(tid,)) for tid in range(num_threads)
|
|
]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join(timeout=30)
|
|
|
|
assert len(errors) == 0, f"Errors in threads: {errors}"
|