Allocate a ring's slots zeroed instead of writing every one (#88)

crates/turnloop/src/queue.rs

@@ -18,16 +18,54 @@ pub(crate) struct Queue<T> {
 unsafe impl<T: Send> Sync for Queue<T> {}
 // SAFETY: no slot is externally borrowed; only Send values can cross threads.
 unsafe impl<T: Send> Send for Queue<T> {}
+/// A ring's slots, without touching them.
+///
+/// `Slot`'s initial value **is** the all-zero bit pattern: `state` starts at 0,
+/// which is the empty state, and `value` is a `MaybeUninit` for which every
+/// pattern is valid. Building the slots with `map`/`collect` writes each one,
+/// which makes the entire ring resident at construction — 3.15 MiB for a
+/// 32 768-slot `WorkPort`, in a process that may never submit a blocking job.
+///
+/// Asking the allocator for zeroed memory instead gives a ready ring with no
+/// writes at all, and a zeroed allocation this large is fresh pages the OS
+/// faults in lazily. The capacity is unchanged, so every bound that depends on
+/// it — including the blocking pool's "cannot overflow" credit invariant — is
+/// untouched; only the pages the ring has actually used are resident.
+///
+/// `Queue::drop` pops until empty, and an untouched slot is state 0 (empty), so
+/// no unwritten slot is ever read as a value.
+#[cfg(not(loom))]
+fn zeroed_slots<T>(capacity: usize) -> Box<[Slot<T>]> {
+    let layout = std::alloc::Layout::array::<Slot<T>>(capacity).expect("ring layout");
+    // SAFETY: `capacity >= 2` so the layout is non-zero-sized, and the all-zero
+    // bit pattern is a valid `Slot<T>` as argued above.
+    unsafe {
+        let ptr = std::alloc::alloc_zeroed(layout).cast::<Slot<T>>();
+        if ptr.is_null() {
+            std::alloc::handle_alloc_error(layout);
+        }
+        Box::from_raw(std::ptr::slice_from_raw_parts_mut(ptr, capacity))
+    }
+}
+
+/// Under loom, `AtomicUsize` and `UnsafeCell` are instrumented types whose
+/// representation is loom's business and is not all-zero, so the model build
+/// keeps building each slot.
+#[cfg(loom)]
+fn zeroed_slots<T>(capacity: usize) -> Box<[Slot<T>]> {
+    (0..capacity)
+        .map(|_| Slot {
+            state: AtomicUsize::new(0),
+            value: UnsafeCell::new(MaybeUninit::uninit()),
+        })
+        .collect()
+}
+
 impl<T> Queue<T> {
     pub fn new(capacity: usize) -> Self {
         assert!(capacity >= 2 && capacity.is_power_of_two());
         Self {
-            slots: (0..capacity)
-                .map(|_| Slot {
-                    state: AtomicUsize::new(0),
-                    value: UnsafeCell::new(MaybeUninit::uninit()),
-                })
-                .collect(),
+            slots: zeroed_slots(capacity),
             enqueue: AtomicUsize::new(0),
             dequeue: AtomicUsize::new(0),
             occupied: AtomicUsize::new(0),
@@ -138,3 +176,77 @@ mod publication_models {
         });
     }
 }
+
+#[cfg(all(test, not(loom)))]
+mod zeroed_ring {
+    use super::*;
+
+    /// A ring whose slots were never written must behave exactly like one whose
+    /// slots were constructed individually.
+    ///
+    /// The slots come from `alloc_zeroed` so that a large ring costs only the
+    /// pages it touches — `WorkPort` sizes its ring from `max_operations`, which
+    /// a host legitimately sets to 32 768, and writing every slot made all
+    /// 3.15 MiB of it resident in a process that may never submit a blocking
+    /// job. That is only sound because `Slot`'s empty state IS the all-zero bit
+    /// pattern, so this checks the consequences rather than the argument:
+    /// every slot is usable, the ring still refuses at exactly its capacity,
+    /// and nothing unwritten is ever read back as a value.
+    #[test]
+    fn an_unwritten_slot_is_empty_usable_and_bounded() {
+        const CAP: usize = 64;
+        let q: Queue<usize> = Queue::new(CAP);
+        assert!(q.is_empty(), "a freshly zeroed ring reads as empty");
+        assert_eq!(q.pop(), None, "no unwritten slot may read back as a value");
+
+        // Every slot is usable, including ones no constructor ever touched.
+        for i in 0..CAP {
+            q.push(i)
+                .unwrap_or_else(|_| panic!("slot {i} must accept a value"));
+        }
+        // And the capacity still binds at exactly the declared size — the bound
+        // the blocking pool's "cannot overflow" credit invariant relies on.
+        assert_eq!(q.push(CAP), Err(CAP), "a full ring returns the value");
+
+        let mut seen = vec![false; CAP];
+        for _ in 0..CAP {
+            let v = q.pop().expect("every pushed value comes back");
+            assert!(!seen[v], "value {v} delivered twice");
+            seen[v] = true;
+        }
+        assert!(seen.into_iter().all(|s| s), "every value was delivered");
+        assert_eq!(q.pop(), None);
+        assert!(q.is_empty());
+
+        // Recycled slots still work after the ring has wrapped.
+        for i in 0..CAP {
+            q.push(i).expect("recycled slot");
+        }
+        assert_eq!(q.push(CAP), Err(CAP));
+    }
+
+    /// Dropping a ring with values still in it must drop those values and no
+    /// others — an untouched zero slot must not be read as an initialised `T`.
+    #[test]
+    fn dropping_a_partly_filled_ring_drops_only_its_values() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        static DROPPED: AtomicUsize = AtomicUsize::new(0);
+        struct Counted;
+        impl Drop for Counted {
+            fn drop(&mut self) {
+                DROPPED.fetch_add(1, Ordering::SeqCst);
+            }
+        }
+        {
+            let q: Queue<Counted> = Queue::new(1024);
+            for _ in 0..3 {
+                q.push(Counted).map_err(|_| ()).expect("slot");
+            }
+        }
+        assert_eq!(
+            DROPPED.load(Ordering::SeqCst),
+            3,
+            "exactly the three pushed values are dropped, not 1024 zero slots"
+        );
+    }
+}

crates/turnloop/src/sync.rs

@@ -14,6 +14,11 @@ pub(crate) use std::sync::{
 pub(crate) struct UnsafeCell<T>(std::cell::UnsafeCell<T>);
 #[cfg(not(loom))]
 impl<T> UnsafeCell<T> {
+    // The queue's slots are allocated zeroed rather than constructed one at a
+    // time, so nothing in the non-loom build calls this any more. Kept because
+    // it is half of this shim's parity with loom's `UnsafeCell`, and a shim
+    // that only sometimes mirrors its subject is worse than an unused fn.
+    #[allow(dead_code)]
     pub fn new(v: T) -> Self {
         Self(std::cell::UnsafeCell::new(v))
     }