Test fixes

Author: lalinsky

src/Index.zig

@@ -128,6 +128,7 @@ pub fn init(allocator: std.mem.Allocator, parent_dir: std.fs.Dir, path: []const
 }
 
 pub fn deinit(self: *Self) void {
+    log.info("closing index {}", .{@intFromPtr(self)});
     self.stopping.store(true, .release);
 
     self.stopCheckpointThread();
@@ -229,6 +230,7 @@ fn doCheckpoint(self: *Self) !bool {
 }
 
 fn checkpointThreadFn(self: *Self) void {
+    log.debug("checkpoint thread started", .{});
     while (!self.stopping.load(.acquire)) {
         if (self.doCheckpoint()) |successful| {
             if (successful) {
@@ -238,8 +240,10 @@ fn checkpointThreadFn(self: *Self) void {
         } else |err| {
             log.err("checkpoint failed: {}", .{err});
         }
+        log.debug("waiting for checkpoint event", .{});
         self.checkpoint_event.timedWait(std.time.ns_per_min) catch continue;
     }
+    log.debug("checkpoint thread stopped", .{});
 }
 
 fn startCheckpointThread(self: *Self) !void {
@@ -253,8 +257,10 @@ fn stopCheckpointThread(self: *Self) void {
     log.info("stopping checkpoint thread", .{});
     if (self.checkpoint_thread) |thread| {
         self.checkpoint_event.set();
+        log.debug("waiting for checkpoint thread to exit", .{});
         thread.join();
     }
+    log.debug("checkpoint thread stopped", .{});
     self.checkpoint_thread = null;
 }
 

src/MultiIndex.zig

@@ -20,21 +20,14 @@ pub const IndexRef = struct {
     }
 
     pub fn incRef(self: *IndexRef) void {
-        self.lock.lock();
-        defer self.lock.unlock();
-
         self.references += 1;
         self.last_used_at = std.time.milliTimestamp();
     }
 
     pub fn decRef(self: *IndexRef) bool {
-        self.lock.lock();
-        defer self.lock.unlock();
-
         assert(self.references > 0);
         self.references -= 1;
-        self.last_used_at = std.time.timestamp();
-
+        self.last_used_at = std.time.milliTimestamp();
         return self.references == 0;
     }
 
@@ -119,17 +112,10 @@ fn removeIndex(self: *Self, name: []const u8) void {
 }
 
 pub fn releaseIndex(self: *Self, index_ref: *IndexRef) void {
-    if (index_ref.decRef()) {
-        self.lock.lock();
-        defer self.lock.unlock();
-
-        // index_ref.lock.lock();
-        // defer index_ref.lock.unlock();
+    self.lock.lock();
+    defer self.lock.unlock();
 
-        // if (!index_ref.is_open) {
-        //     self.removeIndex(index_ref.name);
-        // }
-    }
+    _ = index_ref.decRef();
 }
 
 pub fn acquireIndex(self: *Self, name: []const u8) !*IndexRef {
@@ -170,20 +156,29 @@ pub fn getIndex(self: *Self, name: []const u8) !*IndexRef {
 }
 
 pub fn createIndex(self: *Self, name: []const u8) !void {
+    log.info("creating index {s}", .{name});
+
     const index_ref = try self.acquireIndex(name);
     defer self.releaseIndex(index_ref);
 
     try index_ref.ensureOpen(true);
 }
 
 pub fn deleteIndex(self: *Self, name: []const u8) !void {
+    log.info("deleting index {s}", .{name});
+
     if (!isValidName(name)) {
         return error.InvalidIndexName;
     }
 
     self.lock.lock();
     defer self.lock.unlock();
 
-    self.removeIndex(name);
+    if (self.indexes.getEntry(name)) |entry| {
+        entry.value_ptr.index.deinit();
+        self.allocator.free(entry.key_ptr.*);
+        self.indexes.removeByPtr(entry.key_ptr);
+    }
+
     try self.deleteIndexFiles(name);
 }

tests/conftest.py

@@ -6,40 +6,62 @@
 from urllib.parse import urljoin
 
 
-@pytest.fixture()
-def server_port():
-    sock = socket.socket()
-    sock.bind(('', 0))
-    return sock.getsockname()[1]
+class ServerManager:
+
+    def __init__(self, base_dir, port):
+        self.data_dir = base_dir / 'data'
+        self.log_file = base_dir / 'server.log'
+        self.port = port
+        self.process = None
+
+    def start(self):
+        command = [
+            'valgrind',
+            'zig-out/bin/fpindex',
+            '--dir', str(self.data_dir),
+            '--port', str(self.port),
+            '--log-level', 'debug',
+        ]
+        self.process = subprocess.Popen(
+            command,
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.DEVNULL,
+            stderr=self.log_file.open('w'),
+        )
+
+    def stop(self):
+        if self.process is not None:
+            if self.process.returncode is None:
+                self.process.terminate()
+                try:
+                    self.process.wait(timeout=1.0)
+                except subprocess.TimeoutExpired:
+                    self.process.kill()
+                    self.process.wait()
+
+    def error_log(self):
+        for line in self.log_file.read_text().splitlines():
+            yield line
+
+
+@pytest.fixture(scope='session')
+def server(tmp_path_factory):
+    srv = ServerManager(base_dir=tmp_path_factory.mktemp('srv'), port=14502)
+    srv.start()
+    yield srv
+    srv.stop()
+    for line in srv.error_log():
+        print(line)
 
 
-@pytest.fixture()
-def server(server_port, tmp_path):
-    data_dir = tmp_path / 'data'
-    stderr = tmp_path / 'server.stderr.log'
-    command = [
-        'zig-out/bin/fpindex',
-        '--dir', str(data_dir),
-        '--port', str(server_port),
-        '--log-level', 'debug',
-    ]
-    process = subprocess.Popen(
-        command,
-        stdin=subprocess.DEVNULL,
-        stdout=subprocess.DEVNULL,
-        stderr=stderr.open('w'),
-    )
-    yield
-    if process.returncode is None:
-        process.terminate()
-        process.wait()
-    for line in stderr.read_text().splitlines():
-        print(line)
+index_no = 1
 
 
 @pytest.fixture
-def index_name():
-    return 'testidx'
+def index_name(request):
+    global index_no
+    index_no += 1
+    return f't{index_no:03d}'
 
 
 class Client:
@@ -48,18 +70,23 @@ def __init__(self, session, base_url):
         self.base_url = base_url
 
     def head(self, url, **kwargs):
+        kwargs.setdefault('timeout', 1)
         return self.session.head(urljoin(self.base_url, url), **kwargs)
 
     def get(self, url, **kwargs):
+        kwargs.setdefault('timeout', 1)
         return self.session.get(urljoin(self.base_url, url), **kwargs)
 
     def put(self, url, **kwargs):
+        kwargs.setdefault('timeout', 1)
         return self.session.put(urljoin(self.base_url, url), **kwargs)
 
     def post(self, url, **kwargs):
+        kwargs.setdefault('timeout', 1)
         return self.session.post(urljoin(self.base_url, url), **kwargs)
 
     def delete(self, url, **kwargs):
+        kwargs.setdefault('timeout', 1)
         return self.session.delete(urljoin(self.base_url, url), **kwargs)
 
 
@@ -74,17 +101,28 @@ def check_port(port_no):
         return sock.connect_ex(('127.0.0.1', port_no)) == 0
 
 
-@pytest.fixture
-def client(session, server_port, server):
-    deadline = time.time() + 1
-    while not check_port(server_port):
+def wait_for_ready(port, timeout):
+    deadline = time.time() + timeout
+    while not check_port(port):
         if time.time() > deadline:
             raise TimeoutError()
-        time.sleep(0.01)
-    return Client(session, f'http://localhost:{server_port}')
+        time.sleep(timeout / 100.0)
+
+
+@pytest.fixture
+def client(session, server):
+    wait_for_ready(server.port, 1)
+    return Client(session, f'http://localhost:{server.port}')
 
 
 @pytest.fixture()
 def create_index(client, index_name):
     req = client.put(f'/{index_name}')
     req.raise_for_status()
+
+
+@pytest.fixture(autouse=True)
+def delete_index(client, index_name):
+    yield
+    req = client.delete(f'/{index_name}')
+    req.raise_for_status()