Merge branch 'main' of github.com:spacejam/sled into tyler_clippy

spacejam · spacejam · commit 90ec0de94b4d · 2022-02-25T16:00:01.000+01:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -23,6 +23,8 @@
 
 ## Breaking Changes
 
+* #1399 Bump MSRV to 1.51.
+* #1399 Thread support is now required on all platforms.
 * #1135 The "no_metrics" anti-feature has been replaced with
   the "metrics" positive feature.
 * #1178 the `Event` enum has become a unified struct that allows
@@ -50,19 +52,23 @@
   doesn't make sense for things that must fit in memory anyway.
 * #1314 `Subscriber::next_timeout` now requires a mutable self
   reference.
-* #1337 Bump MSRV to 1.48.
 * #1349 The "measure_allocs" feature has been removed.
 * #1354 `Error` has been modified to be Copy, removing all
   heap-allocated variants.
 
-
 ## Bug Fixes
 
 * #1202 Fix a space leak where blobs were not
   removed when replaced by another blob.
 * #1229 the powerful ALICE crash consistency tool has been
   used to discover several crash vulnerabilities, now fixed.
 
+# 0.34.7
+
+## Bug Fixes
+
+* #1314 Fix a bug in Subscriber's Future impl.
+
 # 0.34.6
 
 ## Improvements
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "sled"
-version = "0.34.6"
+version = "0.34.7"
 authors = ["Tyler Neely <t@jujit.su>"]
 description = "Lightweight high-performance pure-rust transactional embedded database."
 license = "MIT/Apache-2.0"
@@ -48,10 +48,10 @@ memshred = []
 
 [dependencies]
 libc = "0.2.96"
-zstd = { version = "0.9.0", optional = true }
+zstd = { version = "0.10.0", optional = true }
 crc32fast = "1.2.1"
 log = "0.4.14"
-parking_lot = "0.11.1"
+parking_lot = "0.12.0"
 color-backtrace = { version = "0.5.1", optional = true }
 num-format = { version = "0.4.0", optional = true }
 rio = { version = "0.9.4", optional = true }
diff --git a/README.md b/README.md
@@ -161,7 +161,7 @@ extreme::run(async move {
 
 # minimum supported Rust version (MSRV)
 
-We support Rust 1.48.0 and up.
+We support Rust 1.51.0 and up.
 
 # architecture
 
diff --git a/scripts/cross_compile.sh b/scripts/cross_compile.sh
@@ -12,10 +12,10 @@ rustup update --no-self-update
 
 RUSTFLAGS="--cfg miri" cargo check
 
-rustup toolchain install 1.48.0 --no-self-update
+rustup toolchain install 1.51.0 --no-self-update
 cargo clean
 rm Cargo.lock
-cargo +1.48.0 check
+cargo +1.51.0 check
 
 for target in $targets; do
   echo "setting up $target..."
diff --git a/src/context.rs b/src/context.rs
@@ -12,19 +12,7 @@ pub struct Context {
     /// When the last high-level reference is dropped, it
     /// should trigger all background threads to clean
     /// up synchronously.
-    #[cfg(all(
-        not(miri),
-        any(
-            windows,
-            target_os = "linux",
-            target_os = "macos",
-            target_os = "dragonfly",
-            target_os = "freebsd",
-            target_os = "openbsd",
-            target_os = "netbsd",
-            target_os = "ios",
-        )
-    ))]
+    #[cfg(not(miri))]
     pub(crate) flusher: Arc<Mutex<Option<flusher::Flusher>>>,
     #[doc(hidden)]
     pub pagecache: PageCache,
@@ -47,19 +35,7 @@ impl Context {
         Ok(Self {
             config,
             pagecache,
-            #[cfg(all(
-                not(miri),
-                any(
-                    windows,
-                    target_os = "linux",
-                    target_os = "macos",
-                    target_os = "dragonfly",
-                    target_os = "freebsd",
-                    target_os = "openbsd",
-                    target_os = "netbsd",
-                    target_os = "ios",
-                )
-            ))]
+            #[cfg(not(miri))]
             flusher: Arc::new(parking_lot::Mutex::new(None)),
         })
     }
diff --git a/src/db.rs b/src/db.rs
@@ -53,19 +53,7 @@ impl Db {
 
         let context = Context::start(config)?;
 
-        #[cfg(all(
-            not(miri),
-            any(
-                windows,
-                target_os = "linux",
-                target_os = "macos",
-                target_os = "dragonfly",
-                target_os = "freebsd",
-                target_os = "openbsd",
-                target_os = "netbsd",
-                target_os = "ios",
-            )
-        ))]
+        #[cfg(not(miri))]
         {
             let flusher_pagecache = context.pagecache.clone();
             let flusher = context.flush_every_ms.map(move |fem| {
diff --git a/src/ivec.rs b/src/ivec.rs
@@ -263,23 +263,12 @@ impl std::borrow::Borrow<[u8]> for &IVec {
     }
 }
 
-macro_rules! from_array {
-    ($($s:expr),*) => {
-        $(
-            impl From<&[u8; $s]> for IVec {
-                fn from(v: &[u8; $s]) -> Self {
-                    Self::from(&v[..])
-                }
-            }
-        )*
+impl<const N: usize> From<&[u8; N]> for IVec {
+    fn from(v: &[u8; N]) -> Self {
+        Self::from(&v[..])
     }
 }
 
-from_array!(
-    0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20,
-    21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32
-);
-
 impl Ord for IVec {
     fn cmp(&self, other: &Self) -> std::cmp::Ordering {
         self.as_ref().cmp(other.as_ref())
diff --git a/src/lib.rs b/src/lib.rs
@@ -223,19 +223,7 @@ pub mod fail;
 #[cfg(feature = "docs")]
 pub mod doc;
 
-#[cfg(all(
-    not(miri),
-    any(
-        windows,
-        target_os = "linux",
-        target_os = "macos",
-        target_os = "dragonfly",
-        target_os = "freebsd",
-        target_os = "openbsd",
-        target_os = "netbsd",
-        target_os = "ios",
-    )
-))]
+#[cfg(not(miri))]
 mod flusher;
 
 #[cfg(feature = "event_log")]
diff --git a/src/pagecache/iobuf.rs b/src/pagecache/iobuf.rs
@@ -873,7 +873,7 @@ impl IoBufs {
         let max_header_stable_lsn = self.max_header_stable_lsn.clone();
         guard.defer(move || {
             trace!("bumping atomic header lsn to {}", stored_max_stable_lsn);
-            bump_atomic_lsn(&max_header_stable_lsn, stored_max_stable_lsn)
+            max_header_stable_lsn.fetch_max(stored_max_stable_lsn, SeqCst)
         });
         guard.flush();
         drop(guard);
diff --git a/src/pagecache/logger.rs b/src/pagecache/logger.rs
@@ -1,9 +1,9 @@
 use std::fs::File;
 
 use super::{
-    arr_to_lsn, arr_to_u32, assert_usize, bump_atomic_lsn, decompress, header,
-    iobuf, lsn_to_arr, pread_exact, pread_exact_or_eof, roll_iobuf, u32_to_arr,
-    Arc, BasedBuf, DiskPtr, HeapId, IoBuf, IoBufs, LogKind, LogOffset, Lsn,
+    arr_to_lsn, arr_to_u32, assert_usize, decompress, header, iobuf,
+    lsn_to_arr, pread_exact, pread_exact_or_eof, roll_iobuf, u32_to_arr, Arc,
+    BasedBuf, DiskPtr, HeapId, IoBuf, IoBufs, LogKind, LogOffset, Lsn,
     MessageKind, Reservation, Serialize, Snapshot, BATCH_MANIFEST_PID,
     COUNTER_PID, MAX_MSG_HEADER_LEN, META_PID, SEG_HEADER_LEN,
 };
@@ -140,7 +140,7 @@ impl Log {
         #[cfg(feature = "compression")]
         {
             if self.config.use_compression && pid != BATCH_MANIFEST_PID {
-                use zstd::block::compress;
+                use zstd::bulk::compress;
 
                 let buf = item.serialize();
 
@@ -417,10 +417,9 @@ impl Log {
                 reservation_lid,
             );
 
-            bump_atomic_lsn(
-                &self.iobufs.max_reserved_lsn,
-                reservation_lsn + inline_buf_len as Lsn - 1,
-            );
+            self.iobufs
+                .max_reserved_lsn
+                .fetch_max(reservation_lsn + inline_buf_len as Lsn - 1, SeqCst);
 
             let (heap_reservation, heap_id_opt) = if over_heap_threshold {
                 let heap_reservation = self
diff --git a/src/pagecache/mod.rs b/src/pagecache/mod.rs
@@ -206,23 +206,6 @@ where
     usize::try_from(from).expect("lost data cast while converting to usize")
 }
 
-// TODO remove this with atomic fetch_max if we increase MSRV to 1.45 or higher
-fn bump_atomic_lsn(atomic_lsn: &AtomicLsn, to: Lsn) {
-    let mut current = atomic_lsn.load(Acquire);
-    loop {
-        if current >= to {
-            return;
-        }
-        let last =
-            atomic_lsn.compare_exchange_weak(current, to, SeqCst, SeqCst);
-        if last.is_ok() {
-            // we succeeded.
-            return;
-        }
-        current = last.unwrap_err();
-    }
-}
-
 use std::convert::{TryFrom, TryInto};
 
 pub(in crate::pagecache) const fn lsn_to_arr(number: Lsn) -> [u8; 8] {
@@ -997,7 +980,7 @@ impl PageCache {
         snapshot::write_snapshot(&self.config, &snapshot)?;
 
         // NB: this must only happen after writing the snapshot to disk
-        bump_atomic_lsn(&self.snapshot_min_lsn, stable_lsn_before);
+        self.snapshot_min_lsn.fetch_max(stable_lsn_before, SeqCst);
 
         // explicitly drop this to make it clear that it needs to
         // be held for the duration of the snapshot operation.
@@ -1100,19 +1083,7 @@ impl PageCacheInner {
     /// move a page. Returns Ok(false) if there were no pages
     /// to GC. Returns an Err if we encountered an IO problem
     /// while performing this GC.
-    #[cfg(all(
-        not(miri),
-        any(
-            windows,
-            target_os = "linux",
-            target_os = "macos",
-            target_os = "dragonfly",
-            target_os = "freebsd",
-            target_os = "openbsd",
-            target_os = "netbsd",
-            target_os = "ios",
-        )
-    ))]
+    #[cfg(not(miri))]
     pub(crate) fn attempt_gc(&self) -> Result<bool> {
         let guard = pin();
         let cc = concurrency_control::read();
diff --git a/src/pagecache/snapshot.rs b/src/pagecache/snapshot.rs
@@ -1,5 +1,5 @@
 #[cfg(feature = "zstd")]
-use zstd::block::{compress, decompress};
+use zstd::bulk::{compress, decompress};
 
 use crate::*;
 
diff --git a/src/subscriber.rs b/src/subscriber.rs
@@ -287,7 +287,8 @@ impl Subscribers {
 
                     for (_id, (waker, sender)) in subs.iter() {
                         let (tx, rx) = OneShot::pair();
-                        if sender.send(rx).is_err() {
+                        if let Err(err) = sender.try_send(rx) {
+                            error!("send error: {:?}", err);
                             continue;
                         }
                         subscribers.push((waker.clone(), tx));
diff --git a/src/threadpool.rs b/src/threadpool.rs
@@ -4,19 +4,7 @@ use std::sync::Arc;
 
 use crate::{OneShot, Result};
 
-#[cfg(all(
-    not(miri),
-    any(
-        windows,
-        target_os = "linux",
-        target_os = "macos",
-        target_os = "dragonfly",
-        target_os = "freebsd",
-        target_os = "openbsd",
-        target_os = "netbsd",
-        target_os = "ios",
-    )
-))]
+#[cfg(not(miri))]
 mod queue {
     use std::{
         cell::RefCell,
@@ -220,19 +208,7 @@ where
     spawn_to(work, &queue::BLOCKING_QUEUE)
 }
 
-#[cfg(any(
-    miri,
-    not(any(
-        windows,
-        target_os = "linux",
-        target_os = "macos",
-        target_os = "dragonfly",
-        target_os = "freebsd",
-        target_os = "openbsd",
-        target_os = "netbsd",
-        target_os = "ios",
-    ))
-))]
+#[cfg(miri)]
 mod queue {
     /// This is the polyfill that just executes things synchronously.
     use crate::{OneShot, Result};
diff --git a/src/tree.rs b/src/tree.rs
@@ -205,8 +205,19 @@ impl Tree {
         let last_value_ivec = last_value.map(IVec::from);
 
         if value == last_value_ivec {
+            // NB: always broadcast event
+            if let Some(Some(res)) = subscriber_reservation.take() {
+                let event = subscriber::Event::single_update(
+                    self.clone(),
+                    key.as_ref().into(),
+                    value,
+                );
+
+                res.complete(&event);
+            }
+
             // short-circuit a no-op set or delete
-            return Ok(Ok(value));
+            return Ok(Ok(last_value));
         }
 
         let frag = if let Some(value_ivec) = value.clone() {
@@ -1362,8 +1373,8 @@ impl Tree {
         }
     }
 
-    /// Create an iterator over tuples of keys and values,
-    /// where the all the keys starts with the given prefix.
+    /// Create an iterator over tuples of keys and values
+    /// where all keys start with the given prefix.
     ///
     /// # Examples
     ///
