INTERNAL: Make a lane job from each item event of the engine

src/aio.rs

@@ -6,23 +6,31 @@ use std::sync::{Arc, OnceLock};
 
 use crate::error::Error;
 use crate::guard;
-use crate::memcached::server::{Connection, ResponseHandler};
-use crate::protocol::Request;
-use pool::Op;
-pub use pool::start;
+use crate::memcached::server::{Connection, IoWaiter, ResponseHandler};
+pub use pool::{Op, start, submit as begin};
 
-/// Hands `request` to a lane, to be replied to from the io callback; an error is for `execute` to
-/// reply with.
-pub fn begin(conn: Connection, request: Request<'_>) -> Result<(), Error> {
-    let op = Op::new(request)?;
-    let reply = Arc::new(OnceLock::new());
-    // SAFETY: the only caller, `cmd::run`, runs in `execute`.
-    let waiter = unsafe { conn.wait_for_io() };
-    let data = Arc::into_raw(Arc::clone(&reply));
-    // SAFETY: as above, and `wake` takes back the count `data` holds.
-    unsafe { conn.on_io_complete(wake, data.cast_mut().cast()) };
-    pool::submit(op, reply, waiter);
-    Ok(())
+/// A connection waiting for a lane to answer it from the io callback; dropped unanswered, it
+/// answers an internal error.
+pub struct Waiter {
+    reply: Arc<OnceLock<Cow<'static, [u8]>>>,
+    _wait: IoWaiter,
+}
+
+impl Waiter {
+    pub fn new(conn: Connection) -> Self {
+        let reply = Arc::new(OnceLock::new());
+        // SAFETY: its callers, `cmd::run` and `event::offer`, run in `execute`.
+        let wait = unsafe { conn.wait_for_io() };
+        let data = Arc::into_raw(Arc::clone(&reply));
+        // SAFETY: as above, and `wake` takes back the count `data` holds.
+        unsafe { conn.on_io_complete(wake, data.cast_mut().cast()) };
+        Self { reply, _wait: wait }
+    }
+
+    /// Leaves `reply` for the io callback, then wakes the connection.
+    pub fn answer(self, reply: Cow<'static, [u8]>) {
+        let _ = self.reply.set(reply);
+    }
 }
 
 unsafe extern "C" fn wake(
@@ -33,7 +41,7 @@ unsafe extern "C" fn wake(
     guard(
         || false,
         || {
-            // SAFETY: `data` is the count from `begin`, and the core calls this once.
+            // SAFETY: `data` is the count from `Waiter::new`, and the core calls this once.
             let reply = unsafe { Arc::from_raw(data.cast::<OnceLock<Cow<'static, [u8]>>>()) };
             let reply = reply.get().map_or(Error::Internal.reply(), AsRef::as_ref);
             // SAFETY: the server is running the io callback for `cookie`, with `handler`.
@@ -42,6 +50,11 @@ unsafe extern "C" fn wake(
     )
 }
 
+#[cfg(test)]
+pub fn start_once() {
+    pool::start_once();
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::mpsc::channel;
@@ -50,7 +63,7 @@ mod tests {
     use super::*;
     use crate::index;
     use crate::memcached::server::fake;
-    use crate::protocol::{self, Tokens};
+    use crate::protocol::{self, Request, Tokens};
     use crate::quant::Vector;
 
     #[test]
@@ -68,7 +81,10 @@ mod tests {
         let tokens = Tokens::from_line("vop search aio vector 16 3");
         let mut request = protocol::parse(tokens.words()).unwrap();
         request.set_vector(Vector::F32(vec![0.0; 4]));
-        assert_eq!(begin(conn.connection(), request), Ok(()));
+        begin(
+            Op::new(request).unwrap(),
+            Some(Waiter::new(conn.connection())),
+        );
         assert_eq!(rx.recv_timeout(Duration::from_secs(10)).unwrap(), 7);
         assert!(conn.reply().is_empty());
         assert!(conn.run_callback());

src/aio/pool.rs

@@ -5,33 +5,36 @@ use std::panic::{AssertUnwindSafe, catch_unwind};
 use std::ptr;
 use std::sync::atomic::{AtomicPtr, AtomicUsize, Ordering};
 use std::sync::mpsc::{Receiver, Sender, channel};
-use std::sync::{Arc, Barrier, OnceLock};
+use std::sync::{Arc, Barrier};
 
+use super::Waiter;
 use crate::error::Error;
-use crate::memcached::server::IoWaiter;
 use crate::protocol::{Request, Search, Source};
 use crate::quant::Vector;
 
 const NUM_LANES: usize = 32;
 
-/// A command for a lane, where it leaves the reply for the io callback, and the connection to wake
-/// once it is done.
+/// An op for a lane, and the connection it answers: none for a change from replication or replay.
 struct Task {
     op: Op,
-    /// Unset if its lane did not finish it.
-    reply: Arc<OnceLock<Cow<'static, [u8]>>>,
-    _wait: IoWaiter,
+    waiter: Option<Waiter>,
 }
 
 #[expect(dead_code, reason = "no op runs yet")]
+#[rustfmt::skip]
 pub enum Op {
     Search { index: String, vector: Vector },
-    Upsert { index: String, vkey: String },
-    SetMeta { index: String, vkey: String },
-    Delete { index: String, vkey: String },
+    Insert { index: String, key: usize },
+    Replace { index: String, old: usize, new: usize },
+    /// A replace from `vop setmeta`, answered as a store.
+    SetMeta { index: String, old: usize, new: usize },
+    Delete { index: String, key: usize },
+    Create { index: String, config: Vec<u8> },
+    Drop { index: String },
 }
 
 impl Op {
+    /// A command's search; a write's op comes from the element event it raises.
     pub fn new(request: Request<'_>) -> Result<Self, Error> {
         match request {
             // Waits for the engine to give the vkey's vector.
@@ -43,42 +46,39 @@ impl Op {
                 index: index.into(),
                 vector,
             }),
-            Request::Upsert(u) => Ok(Self::Upsert {
-                index: u.index.into(),
-                vkey: u.vkey.into(),
-            }),
-            Request::SetMeta(m) => Ok(Self::SetMeta {
-                index: m.index.into(),
-                vkey: m.vkey.into(),
-            }),
-            Request::Delete { index, vkey } => Ok(Self::Delete {
-                index: index.into(),
-                vkey: vkey.into(),
-            }),
             _ => Err(Error::NotSupported),
         }
     }
 
-    /// A search goes to the least busy lane; a write goes to its vkey's lane, so that writes on one
-    /// vkey run in the order they came.
+    /// A search goes to the least busy lane; a write goes to its index's lane, so that writes on
+    /// one index run in the order the engine made them.
     fn lane(&self) -> &'static Lane {
         match self {
             Self::Search { .. } => idlest(),
-            Self::Upsert { index, vkey }
-            | Self::SetMeta { index, vkey }
-            | Self::Delete { index, vkey } => lane_of(index, vkey),
+            Self::Insert { index, .. }
+            | Self::Replace { index, .. }
+            | Self::SetMeta { index, .. }
+            | Self::Delete { index, .. }
+            | Self::Create { index, .. }
+            | Self::Drop { index } => lane_of(index),
         }
     }
 }
 
 impl Task {
     fn run(self) {
         let reply = match self.op {
-            Op::Search { .. } | Op::Upsert { .. } | Op::SetMeta { .. } | Op::Delete { .. } => {
-                Error::NotSupported.reply()
-            }
+            Op::Search { .. }
+            | Op::Insert { .. }
+            | Op::Replace { .. }
+            | Op::SetMeta { .. }
+            | Op::Delete { .. }
+            | Op::Create { .. }
+            | Op::Drop { .. } => Error::NotSupported.reply(),
         };
-        let _ = self.reply.set(Cow::Borrowed(reply));
+        if let Some(waiter) = self.waiter {
+            waiter.answer(Cow::Borrowed(reply));
+        }
     }
 }
 
@@ -177,22 +177,15 @@ fn idlest() -> &'static Lane {
     idlest
 }
 
-fn lane_of(index: &str, vkey: &str) -> &'static Lane {
+fn lane_of(index: &str) -> &'static Lane {
     let mut hasher = DefaultHasher::new();
-    (index, vkey).hash(&mut hasher);
+    index.hash(&mut hasher);
     &lanes()[(hasher.finish() % NUM_LANES as u64) as usize]
 }
 
-/// Runs `op` on its lane, which leaves its reply in `reply`, then wakes the connection.
-pub fn submit(op: Op, reply: Arc<OnceLock<Cow<'static, [u8]>>>, waiter: IoWaiter) {
-    send(
-        op.lane(),
-        Task {
-            op,
-            reply,
-            _wait: waiter,
-        },
-    );
+/// Runs `op` on its lane, which then answers `waiter`.
+pub fn submit(op: Op, waiter: Option<Waiter>) {
+    send(op.lane(), Task { op, waiter });
 }
 
 fn send(lane: &Lane, task: Task) {
@@ -228,17 +221,8 @@ mod tests {
                 index: "idx".into(),
                 vector: Vector::default(),
             };
-            let reply = Arc::default();
-            // SAFETY: the fake server takes a wait on a `Conn` at any time.
-            let waiter = unsafe { conn.connection().wait_for_io() };
-            send(
-                &lanes()[3],
-                Task {
-                    op,
-                    reply,
-                    _wait: waiter,
-                },
-            );
+            let waiter = Some(Waiter::new(conn.connection()));
+            send(&lanes()[3], Task { op, waiter });
         }
         let got: Vec<usize> = (0..100)
             .map(|_| rx.recv_timeout(Duration::from_secs(10)).unwrap())
@@ -247,11 +231,11 @@ mod tests {
     }
 
     #[test]
-    fn one_vkey_always_gets_one_lane() {
+    fn one_index_always_gets_one_lane() {
         start_once();
-        assert!(ptr::eq(lane_of("idx", "a"), lane_of("idx", "a")));
+        assert!(ptr::eq(lane_of("idx"), lane_of("idx")));
         let spread = (0..1000)
-            .map(|n| ptr::from_ref(lane_of("idx", &format!("v{n}"))))
+            .map(|n| ptr::from_ref(lane_of(&format!("idx{n}"))))
             .collect::<HashSet<_>>();
         assert_eq!(spread.len(), NUM_LANES);
     }

src/cmd.rs

@@ -1,4 +1,4 @@
-use crate::aio;
+use crate::aio::{self, Op, Waiter};
 use crate::error::Error;
 use crate::index;
 use crate::memcached::server::Connection;
@@ -13,8 +13,11 @@ pub fn run(conn: Connection, request: Result<Request<'_>, Error>) -> bool {
             Err(e) => e.reply(),
         },
         Ok(request) if index::dimensions(request.index()).is_none() => Error::NotFound.reply(),
-        Ok(request) => match aio::begin(conn, request) {
-            Ok(()) => return true,
+        Ok(request) => match Op::new(request) {
+            Ok(op) => {
+                aio::begin(op, Some(Waiter::new(conn)));
+                return true;
+            }
             Err(e) => e.reply(),
         },
         Err(e) => e.reply(),