fix: replace spin-loops with futex-based sleep to eliminate idle CPU burn

Worker threads and index workers were busy-waiting (spinLoopHint + yield)
when no work was available, causing ~99% CPU at idle. Now they sleep on
futex and are woken on demand. Seqlock and ART spin-locks yield after 16
spins to reduce contention burn. flushIndex uses exponential backoff.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: justrach

src/art.zig

@@ -48,10 +48,17 @@ const NodeHeader = struct {
     }
 
     fn writeLock(self: *NodeHeader) void {
+        var spins: u32 = 0;
         while (true) {
             const v = self.version.load(.acquire);
             if (v & 1 != 0) {
-                std.atomic.spinLoopHint();
+                spins += 1;
+                if (spins > 16) {
+                    std.Thread.yield() catch {};
+                    spins = 0;
+                } else {
+                    std.atomic.spinLoopHint();
+                }
                 continue;
             }
             if (self.version.cmpxchgWeak(v, v + 1, .acq_rel, .acquire)) |_| {

src/collection.zig

@@ -193,6 +193,8 @@ pub const Collection = struct {
     index_thread: ?std.Thread,
     index_thread2: ?std.Thread,
     index_stop: std.atomic.Value(bool),
+    /// Futex signal — index workers sleep on this when queues are empty.
+    index_wake: std.atomic.Value(u32),
     queue_toggle: std.atomic.Value(u32),
     indexing_count: std.atomic.Value(u32),
 
@@ -248,6 +250,7 @@ pub const Collection = struct {
         col.index_queue = IndexQueue.init(alloc);
         col.index_queue2 = IndexQueue.init(alloc);
         col.index_stop = std.atomic.Value(bool).init(false);
+        col.index_wake = std.atomic.Value(u32).init(0);
         col.queue_toggle = std.atomic.Value(u32).init(0);
         col.indexing_count = std.atomic.Value(u32).init(0);
         col.vectors = null;
@@ -271,6 +274,8 @@ pub const Collection = struct {
     pub fn close(self: *Collection) void {
         // Signal background indexers to stop, then join both threads.
         self.index_stop.store(true, .release);
+        // Wake sleeping index workers so they observe the stop flag.
+        std.Thread.Futex.wake(&self.index_wake, std.math.maxInt(u32));
         if (self.index_thread) |t| t.join();
         if (self.index_thread2) |t| t.join();
         // Drain any leftover entries from both queues (free owned slices).
@@ -322,9 +327,14 @@ pub const Collection = struct {
     /// Block until the background indexers have drained all pending items
     /// and finished processing the current batch.
     pub fn flushIndex(self: *Collection) void {
-        var waited: u32 = 0;
-        while ((self.index_queue.len() > 0 or self.index_queue2.len() > 0 or self.indexing_count.load(.acquire) > 0) and waited < 300_000) : (waited += 1) {
-            std.Thread.sleep(100_000); // 100µs
+        var sleep_ns: u64 = 100_000; // start at 100µs
+        var total_ns: u64 = 0;
+        const max_ns: u64 = 30_000_000_000; // 30s total cap
+        while ((self.index_queue.len() > 0 or self.index_queue2.len() > 0 or self.indexing_count.load(.acquire) > 0) and total_ns < max_ns) {
+            std.Thread.sleep(sleep_ns);
+            total_ns += sleep_ns;
+            // Exponential backoff: 100µs → 200µs → ... → 10ms cap.
+            sleep_ns = @min(sleep_ns * 2, 10_000_000);
         }
     }
 
@@ -425,6 +435,9 @@ pub const Collection = struct {
                         }
                         std.Thread.yield() catch {};
                     }
+                    // Wake sleeping index worker to process the new entry.
+                    _ = self.index_wake.fetchAdd(1, .release);
+                    std.Thread.Futex.wake(&self.index_wake, 1);
                 }
             }
         }
@@ -1434,7 +1447,10 @@ fn indexWorkerQ(col: *Collection, queue: *IndexQueue) void {
         }
         if (n == 0) {
             _ = col.indexing_count.fetchSub(1, .release);
-            std.Thread.yield() catch {};
+            // Sleep on futex instead of spinning — woken by index push path.
+            const cur = col.index_wake.load(.acquire);
+            if (!col.index_stop.load(.acquire))
+                std.Thread.Futex.timedWait(&col.index_wake, cur, 50_000_000) catch {};
             continue;
         }
         // Batch trigram indexing (single lock acquisition for all docs).

src/io_engine.zig

@@ -456,6 +456,9 @@ pub const EventLoop = struct {
     work_queue: *MpscRing(WorkItem, 4096),
     workers: []std.Thread,
     worker_count: usize,
+    /// Futex-based wake signal — workers sleep on this instead of spinning.
+    /// Incremented by the I/O thread after pushing work; workers wait on it.
+    wake_signal: std.atomic.Value(u32),
 
     pub fn init(alloc: Allocator, max_conns: usize) !EventLoop {
         const wq = try alloc.create(MpscRing(WorkItem, 4096));
@@ -473,13 +476,16 @@ pub const EventLoop = struct {
             .work_queue = wq,
             .workers = try alloc.alloc(std.Thread, n_workers),
             .worker_count = n_workers,
+            .wake_signal = std.atomic.Value(u32).init(0),
         };
     }
 
     pub fn deinit(self: *EventLoop) void {
         if (self.running.load(.acquire)) {
             self.running.store(false, .release);
         }
+        // Wake all workers so they can observe running=false and exit.
+        std.Thread.Futex.wake(&self.wake_signal, @intCast(self.worker_count));
         if (self.listen_fd >= 0) {
             posix.close(self.listen_fd);
         }
@@ -541,6 +547,8 @@ pub const EventLoop = struct {
     /// Stop the event loop gracefully.
     pub fn stop(self: *EventLoop) void {
         self.running.store(false, .release);
+        // Wake all sleeping workers so they see running=false.
+        std.Thread.Futex.wake(&self.wake_signal, @intCast(self.worker_count));
     }
 
     // ── Internal completion handlers ──
@@ -608,6 +616,10 @@ pub const EventLoop = struct {
             // Queue full — close connection to shed load.
             conn.state = .closing;
             self.engine.submitClose(conn.fd, comp.user_data) catch {};
+        } else {
+            // Wake one sleeping worker to process this item.
+            _ = self.wake_signal.fetchAdd(1, .release);
+            std.Thread.Futex.wake(&self.wake_signal, 1);
         }
     }
 
@@ -650,9 +662,11 @@ pub const EventLoop = struct {
     fn workerThread(self: *EventLoop, handler: *const fn (request: []const u8, response: []u8) usize) void {
         while (self.running.load(.acquire)) {
             const item = self.work_queue.pop() orelse {
-                // No work — brief spin-then-yield.
-                std.atomic.spinLoopHint();
-                std.Thread.yield() catch {};
+                // No work — sleep on futex instead of spinning.
+                const cur = self.wake_signal.load(.acquire);
+                // Re-check running before sleeping (avoid missed wake on shutdown).
+                if (!self.running.load(.acquire)) break;
+                std.Thread.Futex.timedWait(&self.wake_signal, cur, 50_000_000) catch {};
                 continue;
             };
 

src/storage/seqlock.zig

@@ -27,10 +27,17 @@ pub const Seqlock = struct {
     /// Begin a read.  Returns the current sequence number.
     /// Spins until no write is in progress (seq is even).
     pub inline fn readBegin(self: *const Seqlock) u64 {
+        var spins: u32 = 0;
         while (true) {
             const s = self.seq.load(.acquire);
             if (s & 1 == 0) return s;
-            std.atomic.spinLoopHint();
+            spins += 1;
+            if (spins > 16) {
+                std.Thread.yield() catch {};
+                spins = 0;
+            } else {
+                std.atomic.spinLoopHint();
+            }
         }
     }
 
@@ -45,14 +52,21 @@ pub const Seqlock = struct {
     /// Acquire the write lock (spins until no other writer is active).
     /// After this call seq is ODD — readers will spin.
     pub fn writeLock(self: *Seqlock) void {
+        var spins: u32 = 0;
         while (true) {
             const s = self.seq.load(.acquire);
             if (s & 1 == 0) {
                 // Try CAS even → odd
                 if (self.seq.cmpxchgStrong(s, s + 1, .acq_rel, .monotonic) == null)
                     return;
             }
-            std.atomic.spinLoopHint();
+            spins += 1;
+            if (spins > 16) {
+                std.Thread.yield() catch {};
+                spins = 0;
+            } else {
+                std.atomic.spinLoopHint();
+            }
         }
     }