Merge pull request #1190 from spacejam/tyler_inline_maxed_header

inline maxed header, remove unnecessarily complex linearizer

Author: spacejam

Cargo.toml

@@ -44,14 +44,14 @@ mutex = []
 crossbeam-epoch = "0.9.0"
 crossbeam-utils = "0.8.0"
 fxhash = "0.2.1"
-libc = "0.2.79"
+libc = "0.2.80"
 zstd = { version = "0.5.3", optional = true }
 crc32fast = "1.2.1"
 log = "0.4.11"
 parking_lot = "0.11.0"
 color-backtrace = { version = "0.4.2", optional = true }
 rio = { version = "0.9.4", optional = true }
-backtrace = { version = "0.3.53", optional = true }
+backtrace = { version = "0.3.54", optional = true }
 
 [target.'cfg(any(target_os = "linux", target_os = "macos", target_os="windows"))'.dependencies]
 fs2 = "0.4.3"

src/pagecache/header.rs

@@ -0,0 +1,66 @@
+use super::*;
+
+// This is the most writers in a single IO buffer
+// that we have space to accommodate in the counter
+// for writers in the IO buffer header.
+pub(in crate::pagecache) const MAX_WRITERS: Header = 127;
+
+pub(in crate::pagecache) type Header = u64;
+
+// salt: 31 bits
+// maxed: 1 bit
+// seal: 1 bit
+// n_writers: 7 bits
+// offset: 24 bits
+
+pub(crate) const fn is_maxed(v: Header) -> bool {
+    v & (1 << 32) == 1 << 32
+}
+
+pub(crate) const fn mk_maxed(v: Header) -> Header {
+    v | (1 << 32)
+}
+
+pub(crate) const fn is_sealed(v: Header) -> bool {
+    v & (1 << 31) == 1 << 31
+}
+
+pub(crate) const fn mk_sealed(v: Header) -> Header {
+    v | (1 << 31)
+}
+
+pub(crate) const fn n_writers(v: Header) -> Header {
+    (v << 33) >> 57
+}
+
+#[inline]
+pub(crate) fn incr_writers(v: Header) -> Header {
+    assert_ne!(n_writers(v), MAX_WRITERS);
+    v + (1 << 24)
+}
+
+#[inline]
+pub(crate) fn decr_writers(v: Header) -> Header {
+    assert_ne!(n_writers(v), 0);
+    v - (1 << 24)
+}
+
+#[inline]
+pub(crate) fn offset(v: Header) -> usize {
+    let ret = (v << 40) >> 40;
+    usize::try_from(ret).unwrap()
+}
+
+#[inline]
+pub(crate) fn bump_offset(v: Header, by: usize) -> Header {
+    assert_eq!(by >> 24, 0);
+    v + (by as Header)
+}
+
+pub(crate) const fn bump_salt(v: Header) -> Header {
+    (v + (1 << 33)) & 0xFFFF_FFFD_0000_0000
+}
+
+pub(crate) const fn salt(v: Header) -> Header {
+    (v >> 33) << 33
+}

src/pagecache/iobuf.rs

@@ -1,18 +1,11 @@
 use std::{
     alloc::{alloc, dealloc, Layout},
     cell::UnsafeCell,
-    sync::atomic::{AtomicBool, AtomicPtr},
+    sync::atomic::AtomicPtr,
 };
 
 use crate::{pagecache::*, *};
 
-// This is the most writers in a single IO buffer
-// that we have space to accommodate in the counter
-// for writers in the IO buffer header.
-pub(in crate::pagecache) const MAX_WRITERS: Header = 127;
-
-pub(in crate::pagecache) type Header = u64;
-
 macro_rules! io_fail {
     ($self:expr, $e:expr) => {
         #[cfg(feature = "failpoints")]
@@ -69,8 +62,6 @@ pub(crate) struct IoBuf {
     pub offset: LogOffset,
     pub lsn: Lsn,
     pub capacity: usize,
-    maxed: AtomicBool,
-    linearizer: Mutex<()>,
     stored_max_stable_lsn: Lsn,
 }
 
@@ -119,16 +110,6 @@ impl IoBuf {
         }
     }
 
-    // use this for operations on an `IoBuf` that must be
-    // linearized together, and can't fit in the header!
-    pub(crate) fn linearized<F, B>(&self, f: F) -> B
-    where
-        F: FnOnce() -> B,
-    {
-        let _l = self.linearizer.lock();
-        f()
-    }
-
     // This is called upon the initialization of a fresh segment.
     // We write a new segment header to the beginning of the buffer
     // for assistance during recovery. The caller is responsible
@@ -159,22 +140,12 @@ impl IoBuf {
         }
 
         // ensure writes to the buffer land after our header.
-        let last_salt = salt(last);
-        let new_salt = bump_salt(last_salt);
-        let bumped = bump_offset(new_salt, SEG_HEADER_LEN);
+        let last_salt = header::salt(last);
+        let new_salt = header::bump_salt(last_salt);
+        let bumped = header::bump_offset(new_salt, SEG_HEADER_LEN);
         self.set_header(bumped);
     }
 
-    pub(crate) fn set_maxed(&self, maxed: bool) {
-        debug_delay();
-        self.maxed.store(maxed, Release);
-    }
-
-    pub(crate) fn get_maxed(&self) -> bool {
-        debug_delay();
-        self.maxed.load(Acquire)
-    }
-
     pub(crate) fn get_header(&self) -> Header {
         debug_delay();
         self.header.load(Acquire)
@@ -425,8 +396,6 @@ impl IoBufs {
             offset: next_lid,
             lsn: next_lsn,
             capacity: segment_size - base,
-            maxed: AtomicBool::new(false),
-            linearizer: Mutex::new(()),
             stored_max_stable_lsn: -1,
         };
 
@@ -617,9 +586,9 @@ impl IoBufs {
             "created reservation for uninitialized slot",
         );
 
-        assert!(is_sealed(header));
+        assert!(header::is_sealed(header));
 
-        let bytes_to_write = offset(header);
+        let bytes_to_write = header::offset(header);
 
         trace!(
             "write_to_log log_offset {} lsn {} len {}",
@@ -628,7 +597,7 @@ impl IoBufs {
             bytes_to_write
         );
 
-        let maxed = iobuf.linearized(|| iobuf.get_maxed());
+        let maxed = header::is_maxed(header);
         let unused_space = capacity - bytes_to_write;
         let should_pad = maxed && unused_space >= MAX_MSG_HEADER_LEN;
 
@@ -887,18 +856,18 @@ impl IoBufs {
 pub(crate) fn roll_iobuf(iobufs: &Arc<IoBufs>) -> Result<usize> {
     let iobuf = iobufs.current_iobuf();
     let header = iobuf.get_header();
-    if is_sealed(header) {
+    if header::is_sealed(header) {
         trace!("skipping roll_iobuf due to already-sealed header");
         return Ok(0);
     }
-    if offset(header) == 0 {
+    if header::offset(header) == 0 {
         trace!("skipping roll_iobuf due to empty segment");
     } else {
         trace!("sealing ioubuf from  roll_iobuf");
         maybe_seal_and_write_iobuf(iobufs, &iobuf, header, false)?;
     }
 
-    Ok(offset(header))
+    Ok(header::offset(header))
 }
 
 /// Blocks until the specified log sequence number has
@@ -957,7 +926,10 @@ pub(in crate::pagecache) fn make_stable_inner(
 
         let iobuf = iobufs.current_iobuf();
         let header = iobuf.get_header();
-        if offset(header) == 0 || is_sealed(header) || iobuf.lsn > lsn {
+        if header::offset(header) == 0
+            || header::is_sealed(header)
+            || iobuf.lsn > lsn
+        {
             // nothing to write, don't bother sealing
             // current IO buffer.
         } else {
@@ -1051,7 +1023,7 @@ pub(in crate::pagecache) fn maybe_seal_and_write_iobuf(
     header: Header,
     from_reserve: bool,
 ) -> Result<()> {
-    if is_sealed(header) {
+    if header::is_sealed(header) {
         // this buffer is already sealed. nothing to do here.
         return Ok(());
     }
@@ -1063,39 +1035,27 @@ pub(in crate::pagecache) fn maybe_seal_and_write_iobuf(
     let capacity = iobuf.capacity;
     let segment_size = iobufs.config.segment_size;
 
-    if offset(header) > capacity {
+    if header::offset(header) > capacity {
         // a race happened, nothing we can do
         return Ok(());
     }
 
-    let sealed = mk_sealed(header);
-    let res_len = offset(sealed);
-
+    let res_len = header::offset(header);
     let maxed = from_reserve || capacity - res_len < MAX_MSG_HEADER_LEN;
+    let sealed = if maxed {
+        trace!("setting maxed to true for iobuf with lsn {}", lsn);
+        header::mk_maxed(header::mk_sealed(header))
+    } else {
+        header::mk_sealed(header)
+    };
 
-    let worked = iobuf.linearized(|| {
-        if iobuf.cas_header(header, sealed).is_err() {
-            // cas failed, don't try to continue
-            return false;
-        }
-
-        trace!("sealed iobuf with lsn {}", lsn);
-
-        if maxed {
-            // NB we linearize this together with sealing
-            // the header here to guarantee that in write_to_log,
-            // which may be executing as soon as the seal is set
-            // by another thread, the thread that calls
-            // iobuf.get_maxed() is linearized with this one!
-            trace!("setting maxed to true for iobuf with lsn {}", lsn);
-            iobuf.set_maxed(true);
-        }
-        true
-    });
+    let worked = iobuf.cas_header(header, sealed).is_ok();
     if !worked {
         return Ok(());
     }
 
+    trace!("sealed iobuf with lsn {}", lsn);
+
     assert!(
         capacity + SEG_HEADER_LEN >= res_len,
         "res_len of {} higher than buffer capacity {}",
@@ -1166,8 +1126,6 @@ pub(in crate::pagecache) fn maybe_seal_and_write_iobuf(
             offset: next_offset,
             lsn: next_lsn,
             capacity: segment_size,
-            maxed: AtomicBool::new(false),
-            linearizer: Mutex::new(()),
             stored_max_stable_lsn: -1,
         };
 
@@ -1177,8 +1135,8 @@ pub(in crate::pagecache) fn maybe_seal_and_write_iobuf(
     } else {
         let new_cap = capacity - res_len;
         assert_ne!(new_cap, 0);
-        let last_salt = salt(sealed);
-        let new_salt = bump_salt(last_salt);
+        let last_salt = header::salt(sealed);
+        let new_salt = header::bump_salt(last_salt);
 
         IoBuf {
             // reuse the previous io buffer
@@ -1188,8 +1146,6 @@ pub(in crate::pagecache) fn maybe_seal_and_write_iobuf(
             offset: next_offset,
             lsn: next_lsn,
             capacity: new_cap,
-            maxed: AtomicBool::new(false),
-            linearizer: Mutex::new(()),
             stored_max_stable_lsn: -1,
         }
     };
@@ -1216,7 +1172,7 @@ pub(in crate::pagecache) fn maybe_seal_and_write_iobuf(
     drop(measure_assign_offset);
 
     // if writers is 0, it's our responsibility to write the buffer.
-    if n_writers(sealed) == 0 {
+    if header::n_writers(sealed) == 0 {
         iobufs.config.global_error()?;
         trace!(
             "asynchronously writing iobuf with lsn {} to log from maybe_seal",
@@ -1273,53 +1229,9 @@ impl Debug for IoBuf {
             "\n\tIoBuf {{ lid: {}, n_writers: {}, offset: \
              {}, sealed: {} }}",
             self.offset,
-            n_writers(header),
-            offset(header),
-            is_sealed(header)
+            header::n_writers(header),
+            header::offset(header),
+            header::is_sealed(header)
         ))
     }
 }
-
-pub(crate) const fn is_sealed(v: Header) -> bool {
-    v & 1 << 31 == 1 << 31
-}
-
-pub(crate) const fn mk_sealed(v: Header) -> Header {
-    v | 1 << 31
-}
-
-pub(crate) const fn n_writers(v: Header) -> Header {
-    v << 33 >> 57
-}
-
-#[inline]
-pub(crate) fn incr_writers(v: Header) -> Header {
-    assert_ne!(n_writers(v), MAX_WRITERS);
-    v + (1 << 24)
-}
-
-#[inline]
-pub(crate) fn decr_writers(v: Header) -> Header {
-    assert_ne!(n_writers(v), 0);
-    v - (1 << 24)
-}
-
-#[inline]
-pub(crate) fn offset(v: Header) -> usize {
-    let ret = v << 40 >> 40;
-    usize::try_from(ret).unwrap()
-}
-
-#[inline]
-pub(crate) fn bump_offset(v: Header, by: usize) -> Header {
-    assert_eq!(by >> 24, 0);
-    v + (by as Header)
-}
-
-pub(crate) const fn bump_salt(v: Header) -> Header {
-    (v + (1 << 32)) & 0xFFFF_FFFF_0000_0000
-}
-
-pub(crate) const fn salt(v: Header) -> Header {
-    v >> 32 << 32
-}