feat: magic in headers; proper atomic sync (#53)

Author: OliverNChalk

src/error.rs

@@ -2,6 +2,7 @@ use std::fmt::Display;
 
 #[derive(Debug)]
 pub enum Error {
+    InvalidMagic,
     InvalidVersion { expected: u32, actual: u32 },
     InvalidBufferSize,
     Io(std::io::Error),
@@ -13,6 +14,7 @@ impl std::error::Error for Error {}
 impl Display for Error {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
+            Self::InvalidMagic => write!(f, "invalid magic"),
             Self::InvalidVersion { expected, actual } => write!(
                 f,
                 "invalid version; expected={}.{}; found={}.{}",

src/mpmc.rs

@@ -1,12 +1,13 @@
 //! DPDK-style bounded MPMC ring queue
 
 use crate::{error::Error, shmem::MappedRegion, CacheAlignedAtomicSize, VERSION};
-use core::{
-    marker::PhantomData,
-    ptr::NonNull,
-    sync::atomic::{AtomicU32, Ordering},
+use core::{marker::PhantomData, ptr::NonNull, sync::atomic::Ordering};
+use std::{
+    fs::File,
+    sync::{atomic::AtomicU64, Arc},
 };
-use std::{fs::File, sync::Arc};
+
+const MAGIC: u64 = 0x7368_6171_6d70_6d63; // b"shaqmpmc"
 
 pub struct Producer<T> {
     queue: SharedQueue<T>,
@@ -443,7 +444,7 @@ impl<T> SharedQueue<T> {
         header: NonNull<SharedQueueHeader>,
     ) -> Result<Self, Error> {
         let header_ref = unsafe { header.as_ref() };
-        let buffer_mask = header_ref.buffer_mask;
+        let buffer_mask = header_ref.buffer_mask as usize;
         let buffer_size_in_items = buffer_mask.wrapping_add(1);
         if !buffer_size_in_items.is_power_of_two()
             || buffer_size_in_items == 0
@@ -484,6 +485,11 @@ impl<T> SharedQueue<T> {
 
 #[repr(C)]
 struct SharedQueueHeader {
+    // Cold metadata cacheline.
+    magic: AtomicU64,
+    version: u32,
+    buffer_mask: u32,
+
     /// Producer reservation cursor.
     ///
     /// Producers atomically advance this with CAS to claim slots, but claimed
@@ -506,8 +512,6 @@ struct SharedQueueHeader {
     /// Consumers advance this in-order after dropping/reading claimed slots.
     /// Producers use it to determine how much free space is available.
     consumer_release: CacheAlignedAtomicSize,
-    buffer_mask: usize,
-    version: AtomicU32,
 }
 
 impl SharedQueueHeader {
@@ -567,8 +571,9 @@ impl SharedQueueHeader {
         header.producer_publication.store(0, Ordering::Release);
         header.consumer_reservation.store(0, Ordering::Release);
         header.consumer_release.store(0, Ordering::Release);
-        header.buffer_mask = buffer_size_in_items - 1;
-        header.version.store(VERSION, Ordering::SeqCst);
+        header.buffer_mask = (buffer_size_in_items - 1) as u32;
+        header.version = VERSION;
+        header.magic.store(MAGIC, Ordering::Release);
     }
 
     fn join<T: Sized>(file: &File) -> Result<(Arc<MappedRegion>, NonNull<Self>), Error> {
@@ -581,14 +586,16 @@ impl SharedQueueHeader {
             //         memory is aligned to the page size, which is sufficient for the
             //         alignment of `SharedQueueHeader`.
             let header = unsafe { header.as_ref() };
-            let actual_version = header.version.load(Ordering::SeqCst);
-            if actual_version != VERSION {
+            if header.magic.load(Ordering::Acquire) != MAGIC {
+                return Err(Error::InvalidMagic);
+            }
+            if header.version != VERSION {
                 return Err(Error::InvalidVersion {
                     expected: VERSION,
-                    actual: actual_version,
+                    actual: header.version,
                 });
             }
-            let buffer_size_in_items = header.buffer_mask.wrapping_add(1);
+            let buffer_size_in_items = (header.buffer_mask as usize).wrapping_add(1);
             if buffer_size_in_items != Self::calculate_buffer_size_in_items::<T>(file_size)? {
                 return Err(Error::InvalidBufferSize);
             }

src/spsc.rs

@@ -3,11 +3,13 @@ use core::ptr::NonNull;
 use std::{
     fs::File,
     sync::{
-        atomic::{AtomicU32, Ordering},
+        atomic::{AtomicU64, Ordering},
         Arc,
     },
 };
 
+const MAGIC: u64 = 0x7368_6171_7370_7363; // b"shaqspsc"
+
 /// Calculates the minimum file size required for a queue with given capacity.
 /// Note that file size MAY need to be increased beyond this to account for
 /// page-size requirements.
@@ -114,11 +116,9 @@ impl<T: Sized> Producer<T> {
     /// All reserved positions must be fully initialized before calling `commit`.
     /// Pointers should be dropped before calling `commit`.
     pub unsafe fn reserve(&mut self) -> Option<NonNull<T>> {
-        // If write is >= read + buffer_size, the queue is written one iteration
+        // If write is > read + buffer_mask, the queue is written one iteration
         // ahead of the consumer, and we cannot reserve more space.
-        if self.queue.cached_write.wrapping_sub(self.queue.cached_read)
-            >= self.queue.header().buffer_size
-        {
+        if self.queue.cached_write.wrapping_sub(self.queue.cached_read) > self.queue.buffer_mask {
             return None;
         }
 
@@ -294,10 +294,9 @@ impl<T: Sized> SharedQueue<T> {
         header: NonNull<SharedQueueHeader>,
     ) -> Result<Self, Error> {
         // SAFETY: `header` is non-null and aligned properly.
-        let size = unsafe { header.as_ref().buffer_size };
+        let size = unsafe { (header.as_ref().buffer_mask as usize).wrapping_add(1) };
 
         if !size.is_power_of_two()
-            || size == 0
             || SharedQueueHeader::calculate_buffer_size_in_items::<T>(region.file_size())? != size
         {
             return Err(Error::InvalidBufferSize);
@@ -374,10 +373,14 @@ impl<T: Sized> SharedQueue<T> {
 /// Header in shared memory for the queue.
 #[repr(C)]
 struct SharedQueueHeader {
+    // Cold cache line.
+    magic: AtomicU64,
+    version: u32,
+    buffer_mask: u32,
+
+    // Hot cache lines.
     write: CacheAlignedAtomicSize,
     read: CacheAlignedAtomicSize,
-    buffer_size: usize,
-    version: AtomicU32,
 }
 
 impl SharedQueueHeader {
@@ -436,8 +439,9 @@ impl SharedQueueHeader {
         let header = unsafe { header.as_mut() };
         header.write.store(0, Ordering::Release);
         header.read.store(0, Ordering::Release);
-        header.buffer_size = buffer_size_in_items;
-        header.version.store(VERSION, Ordering::SeqCst);
+        header.buffer_mask = (buffer_size_in_items - 1) as u32;
+        header.version = VERSION;
+        header.magic.store(MAGIC, Ordering::Release);
     }
 
     fn join<T: Sized>(file: &File) -> Result<(Arc<MappedRegion>, NonNull<Self>), Error> {
@@ -450,14 +454,18 @@ impl SharedQueueHeader {
             //         memory is aligned to the page size, which is sufficient for the
             //         alignment of `SharedQueueHeader`.
             let header = unsafe { header.as_ref() };
-            let actual_version = header.version.load(Ordering::SeqCst);
-            if actual_version != VERSION {
+            if header.magic.load(Ordering::Acquire) != MAGIC {
+                return Err(Error::InvalidMagic);
+            }
+            if header.version != VERSION {
                 return Err(Error::InvalidVersion {
                     expected: VERSION,
-                    actual: actual_version,
+                    actual: header.version,
                 });
             }
-            if header.buffer_size != Self::calculate_buffer_size_in_items::<T>(file_size)? {
+            if (header.buffer_mask as usize).wrapping_add(1)
+                != Self::calculate_buffer_size_in_items::<T>(file_size)?
+            {
                 return Err(Error::InvalidBufferSize);
             }
         }