Method prefetch_batch

nodedb/src/data/vamana_fetcher.rs:197–243

Submit `IORING_OP_READ` SQEs for each index in `indices`. Indices that are already completed (from a prior prefetch), already in-flight, or cannot get a pool buffer are silently skipped — `fetch_fp32` will handle them via a sync fallback.

(&mut self, indices: &[u32])

    /// in-flight, or cannot get a pool buffer are silently skipped —
    /// `fetch_fp32` will handle them via a sync fallback.
    fn prefetch_batch(&mut self, indices: &[u32]) {
        // Drain any already-completed CQEs to free pool slots.
        self.drain_completions();

        let dim = self.layout.dim as usize;
        let needed = dim * 4;
        let buf_size = align_up(needed, ALIGNMENT);
        let fd = io_uring::types::Fd(self.file.as_raw_fd());

        for &node_idx in indices {
            // Skip: already completed, already in-flight.
            if self.completed.contains_key(&node_idx) || self.pending.contains_key(&node_idx) {
                continue;
            }
            // Skip: no free buffer.
            let Some(slot) = self.free_bufs.pop() else {
                continue;
            };
            // Grow pool buffer to required size if needed (shouldn't happen
            // since pool is sized at open, but guard against it).
            if self.buf_pool[slot].capacity() < buf_size {
                self.free_bufs.push(slot);
                continue;
            }

            let off = vector_offset(&self.layout, node_idx as u64);
            let op =
                io_uring::opcode::Read::new(fd, self.buf_pool[slot].as_mut_ptr(), buf_size as u32)
                    .offset(off)
                    .build()
                    .user_data(node_idx as u64);

            // SAFETY: buf_pool[slot] outlives the SQE and completion.
            // `self.pending` keeps the slot reserved until drain.
            let pushed = unsafe { self.ring.submission().push(&op).is_ok() };
            if pushed {
                self.pending.insert(node_idx, slot);
                self.in_flight += 1;
            } else {
                // SQ full — return the slot.
                self.free_bufs.push(slot);
            }
        }

        // Flush the submission queue to the kernel without waiting.
        let _ = self.ring.submit();
    }

    fn fetch_fp32(&mut self, idx: u32) -> Option<Vec<f32>> {
        // Drain any completions that arrived since last prefetch/fetch.