Merge pull request #1399 from spacejam/tyler_always_use_threads

Require threads on all platforms.

Author: spacejam

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,7 +52,6 @@
   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.

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
 

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..."

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)),
         })
     }

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| {

src/lib.rs

@@ -225,19 +225,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")]

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);

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,
 };
@@ -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) = if over_heap_threshold {
                 let heap_reservation = self

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(crate) 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();

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};