Merge branch 'moka-rs:main' into main

.ci_extras/pin-crate-vers-nightly.sh

@@ -3,4 +3,4 @@
 set -eux
 
 # Pin some dependencies to specific versions for the nightly toolchain.
-cargo update -p proc-macro2 --precise 1.0.60
+cargo update -p proc-macro2 --precise 1.0.63

.github/workflows/CI.yml

@@ -19,6 +19,8 @@ jobs:
   test:
     runs-on: ubuntu-latest
     strategy:
+      # Continue running other jobs in the matrix even if one fails.
+      fail-fast: false
       matrix:
         rust:
           - stable

.github/workflows/CIQuantaDisabled.yml

@@ -19,6 +19,8 @@ jobs:
   test:
     runs-on: ubuntu-latest
     strategy:
+      # Continue running other jobs in the matrix even if one fails.
+      fail-fast: false
       matrix:
         rust:
           - stable

CHANGELOG.md

@@ -1,5 +1,28 @@
 # Moka Cache — Change Log
 
+## Version 0.12.2
+
+### Fixed
+
+- Prevent timing issues in writes that cause inconsistencies between the cache's
+  internal data structures ([#348][gh-pull-0348]):
+    - One way to trigger the issue is that insert the same key twice quickly, once
+      when the cache is full and a second time when there is a room in the cache.
+      - When it occurs, the cache will not return the value inserted in the second
+        call (which is wrong), and the `entry_count` method will keep returning a non
+        zero value after calling the `invalidate_all` method (which is also wrong).
+    - These issues were already present in `v0.11.x` and older versions, but less
+      likely to occur because these versions had smaller time windows for the issues
+      to occur by having a background threads to periodically process pending tasks.
+
+### Changed
+
+- Updated the Rust edition from 2018 to 2021. ([#339][gh-pull-0339], by
+  [@nyurik][gh-nyurik])
+- Changed to use inline format arguments throughout the code, including examples.
+  ([#340][gh-pull-0340], by [@nyurik][gh-nyurik])
+
+
 ## Version 0.12.1
 
 ### Fixed
@@ -711,6 +734,7 @@ The minimum supported Rust version (MSRV) is now 1.51.0 (Mar 25, 2021).
 [gh-LMJW]: https://github.com/LMJW
 [gh-Milo123459]: https://github.com/Milo123459
 [gh-messense]: https://github.com/messense
+[gh-nyurik]: https://github.com/nyurik
 [gh-paolobarbolini]: https://github.com/paolobarbolini
 [gh-peter-scholtens]: https://github.com/peter-scholtens
 [gh-saethlin]: https://github.com/saethlin
@@ -738,6 +762,9 @@ The minimum supported Rust version (MSRV) is now 1.51.0 (Mar 25, 2021).
 [gh-issue-0034]: https://github.com/moka-rs/moka/issues/34/
 [gh-issue-0031]: https://github.com/moka-rs/moka/issues/31/
 
+[gh-pull-0348]: https://github.com/moka-rs/moka/pull/348/
+[gh-pull-0340]: https://github.com/moka-rs/moka/pull/340/
+[gh-pull-0339]: https://github.com/moka-rs/moka/pull/339/
 [gh-pull-0331]: https://github.com/moka-rs/moka/pull/331/
 [gh-pull-0316]: https://github.com/moka-rs/moka/pull/316/
 [gh-pull-0309]: https://github.com/moka-rs/moka/pull/309/

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "moka"
-version = "0.12.1"
+version = "0.12.2"
 edition = "2021"
 # Rust 1.65 was released on Nov 3, 2022.
 rust-version = "1.65"

MIGRATION-GUIDE.md

@@ -2,7 +2,7 @@
 
 ## Migrating to v0.12 from a prior version
 
-v0.12.0 has major breaking changes on the API and internal behavior. This section
+v0.12.0 had major breaking changes on the API and internal behavior. This section
 describes the code changes required to migrate to v0.12.0.
 
 ### Highlights v0.12

README.md

@@ -9,7 +9,7 @@
 [![FOSSA Status](https://app.fossa.com/api/projects/git%2Bgh.neting.cc%2Fmoka-rs%2Fmoka.svg?type=shield)](https://app.fossa.com/projects/git%2Bgh.neting.cc%2Fmoka-rs%2Fmoka?ref=badge_shield)
 
 > **note**
-> `v0.12.0` has major breaking changes on the API and internal behavior. Please read
+> `v0.12.0` had major breaking changes on the API and internal behavior. Please read
 > the [MIGRATION-GUIDE.md][migration-guide-v012] for the details.
 
 * * *
@@ -118,7 +118,7 @@ routers. Here are some highlights:
 ## Recent Changes
 
 > **Note**
-> `v0.12.0` has major breaking changes on the API and internal behavior. Please read
+> `v0.12.0` had major breaking changes on the API and internal behavior. Please read
 > the [MIGRATION-GUIDE.md][migration-guide-v012] for the details.
 
 - [MIGRATION-GUIDE.md][migration-guide-v012]

src/common.rs

@@ -35,7 +35,6 @@ impl From<usize> for CacheRegion {
     }
 }
 
-#[cfg(feature = "future")]
 impl CacheRegion {
     pub(crate) fn name(self) -> &'static str {
         match self {

src/common/concurrent.rs

@@ -90,6 +90,10 @@ impl<K> KeyHashDate<K> {
         self.entry_info.last_modified()
     }
 
+    pub(crate) fn last_accessed(&self) -> Option<Instant> {
+        self.entry_info.last_accessed()
+    }
+
     pub(crate) fn is_dirty(&self) -> bool {
         self.entry_info.is_dirty()
     }
@@ -215,10 +219,6 @@ impl<K, V> ValueEntry<K, V> {
         self.info.is_dirty()
     }
 
-    pub(crate) fn set_dirty(&self, value: bool) {
-        self.info.set_dirty(value);
-    }
-
     #[inline]
     pub(crate) fn policy_weight(&self) -> u32 {
         self.info.policy_weight()
@@ -303,7 +303,6 @@ impl<K, V> AccessTime for TrioArc<ValueEntry<K, V>> {
 pub(crate) enum ReadOp<K, V> {
     Hit {
         value_entry: TrioArc<ValueEntry<K, V>>,
-        timestamp: Instant,
         is_expiry_modified: bool,
     },
     // u64 is the hash of the key.
@@ -314,10 +313,15 @@ pub(crate) enum WriteOp<K, V> {
     Upsert {
         key_hash: KeyHash<K>,
         value_entry: TrioArc<ValueEntry<K, V>>,
+        /// Entry generation after the operation.
+        entry_gen: u16,
         old_weight: u32,
         new_weight: u32,
     },
-    Remove(KvEntry<K, V>),
+    Remove {
+        kv_entry: KvEntry<K, V>,
+        entry_gen: u16,
+    },
 }
 
 /// Cloning a `WriteOp` is safe and cheap because it uses `Arc` and `TrioArc` pointers to
@@ -328,15 +332,23 @@ impl<K, V> Clone for WriteOp<K, V> {
             Self::Upsert {
                 key_hash,
                 value_entry,
+                entry_gen,
                 old_weight,
                 new_weight,
             } => Self::Upsert {
                 key_hash: key_hash.clone(),
                 value_entry: TrioArc::clone(value_entry),
+                entry_gen: *entry_gen,
                 old_weight: *old_weight,
                 new_weight: *new_weight,
             },
-            Self::Remove(kv_hash) => Self::Remove(kv_hash.clone()),
+            Self::Remove {
+                kv_entry,
+                entry_gen,
+            } => Self::Remove {
+                kv_entry: kv_entry.clone(),
+                entry_gen: *entry_gen,
+            },
         }
     }
 }
@@ -345,7 +357,28 @@ impl<K, V> fmt::Debug for WriteOp<K, V> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::Upsert { .. } => f.debug_struct("Upsert").finish(),
-            Self::Remove(..) => f.debug_tuple("Remove").finish(),
+            Self::Remove { .. } => f.debug_tuple("Remove").finish(),
+        }
+    }
+}
+
+impl<K, V> WriteOp<K, V> {
+    pub(crate) fn new_upsert(
+        key: &Arc<K>,
+        hash: u64,
+        value_entry: &TrioArc<ValueEntry<K, V>>,
+        entry_generation: u16,
+        old_weight: u32,
+        new_weight: u32,
+    ) -> Self {
+        let key_hash = KeyHash::new(Arc::clone(key), hash);
+        let value_entry = TrioArc::clone(value_entry);
+        Self::Upsert {
+            key_hash,
+            value_entry,
+            entry_gen: entry_generation,
+            old_weight,
+            new_weight,
         }
     }
 }

src/common/concurrent/deques.rs

@@ -34,7 +34,6 @@ impl<K> Default for Deques<K> {
 }
 
 impl<K> Deques<K> {
-    #[cfg(feature = "future")]
     pub(crate) fn select_mut(
         &mut self,
         selector: CacheRegion,

src/common/concurrent/entry_info.rs

@@ -1,19 +1,22 @@
-use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
+use std::sync::atomic::{self, AtomicBool, AtomicU16, AtomicU32, Ordering};
 
 use super::{AccessTime, KeyHash};
 use crate::common::{concurrent::atomic_time::AtomicInstant, time::Instant};
 
 #[derive(Debug)]
 pub(crate) struct EntryInfo<K> {
     key_hash: KeyHash<K>,
-    /// `is_admitted` indicates that the entry has been admitted to the
-    /// cache. When `false`, it means the entry is _temporary_ admitted to
-    /// the cache or evicted from the cache (so it should not have LRU nodes).
+    /// `is_admitted` indicates that the entry has been admitted to the cache. When
+    /// `false`, it means the entry is _temporary_ admitted to the cache or evicted
+    /// from the cache (so it should not have LRU nodes).
     is_admitted: AtomicBool,
-    /// `is_dirty` indicates that the entry has been inserted (or updated)
-    /// in the hash table, but the history of the insertion has not yet
-    /// been applied to the LRU deques and LFU estimator.
-    is_dirty: AtomicBool,
+    /// `entry_gen` (entry generation) is incremented every time the entry is updated
+    /// in the concurrent hash table.
+    entry_gen: AtomicU16,
+    /// `policy_gen` (policy generation) is incremented every time entry's `WriteOp`
+    /// is applied to the cache policies including the access-order queue (the LRU
+    /// deque).
+    policy_gen: AtomicU16,
     last_accessed: AtomicInstant,
     last_modified: AtomicInstant,
     expiration_time: AtomicInstant,
@@ -29,7 +32,9 @@ impl<K> EntryInfo<K> {
         Self {
             key_hash,
             is_admitted: AtomicBool::default(),
-            is_dirty: AtomicBool::new(true),
+            // `entry_gen` starts at 1 and `policy_gen` start at 0.
+            entry_gen: AtomicU16::new(1),
+            policy_gen: AtomicU16::new(0),
             last_accessed: AtomicInstant::new(timestamp),
             last_modified: AtomicInstant::new(timestamp),
             expiration_time: AtomicInstant::default(),
@@ -52,14 +57,54 @@ impl<K> EntryInfo<K> {
         self.is_admitted.store(value, Ordering::Release);
     }
 
+    /// Returns `true` if the `ValueEntry` having this `EntryInfo` is dirty.
+    ///
+    /// Dirty means that the entry has been updated in the concurrent hash table but
+    /// not yet in the cache policies such as access-order queue.
     #[inline]
     pub(crate) fn is_dirty(&self) -> bool {
-        self.is_dirty.load(Ordering::Acquire)
+        let result =
+            self.entry_gen.load(Ordering::Relaxed) != self.policy_gen.load(Ordering::Relaxed);
+        atomic::fence(Ordering::Acquire);
+        result
     }
 
     #[inline]
-    pub(crate) fn set_dirty(&self, value: bool) {
-        self.is_dirty.store(value, Ordering::Release);
+    pub(crate) fn entry_gen(&self) -> u16 {
+        self.entry_gen.load(Ordering::Acquire)
+    }
+
+    /// Increments the entry generation and returns the new value.
+    #[inline]
+    pub(crate) fn incr_entry_gen(&self) -> u16 {
+        // NOTE: This operation wraps around on overflow.
+        let prev = self.entry_gen.fetch_add(1, Ordering::AcqRel);
+        // Need to add `1` to the previous value to get the current value.
+        prev.wrapping_add(1)
+    }
+
+    /// Sets the policy generation to the given value.
+    #[inline]
+    pub(crate) fn set_policy_gen(&self, value: u16) {
+        let g = &self.policy_gen;
+        loop {
+            let current = g.load(Ordering::Acquire);
+
+            // Do not set the given value if it is smaller than the current value of
+            // `policy_gen`. Note that the current value may have been wrapped
+            // around. If the value is much larger than the current value, it is
+            // likely that the value of `policy_gen` has been wrapped around.
+            if current >= value || value.wrapping_sub(current) > u16::MAX / 2 {
+                break;
+            }
+
+            // Try to set the value.
+            if g.compare_exchange_weak(current, value, Ordering::AcqRel, Ordering::Acquire)
+                .is_ok()
+            {
+                break;
+            }
+        }
     }
 
     #[inline]
@@ -134,7 +179,9 @@ mod test {
         #[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)]
         enum TargetArch {
             Linux64,
-            Linux32,
+            Linux32X86,
+            Linux32Arm,
+            Linux32Mips,
             MacOS64,
         }
 
@@ -149,9 +196,17 @@ mod test {
             if cfg!(target_pointer_width = "64") {
                 Linux64
             } else if cfg!(target_pointer_width = "32") {
-                Linux32
+                if cfg!(target_arch = "x86") {
+                    Linux32X86
+                } else if cfg!(target_arch = "arm") {
+                    Linux32Arm
+                } else if cfg!(target_arch = "mips") {
+                    Linux32Mips
+                } else {
+                    unimplemented!();
+                }
             } else {
-                panic!("Unsupported pointer width for Linux");
+                unimplemented!();
             }
         } else if cfg!(target_os = "macos") {
             MacOS64
@@ -160,12 +215,14 @@ mod test {
         };
 
         let expected_sizes = match (arch, is_quanta_enabled) {
-            (Linux64, true) => vec![("1.51", 48)],
-            (Linux32, true) => vec![("1.51", 48)],
-            (MacOS64, true) => vec![("1.62", 48)],
-            (Linux64, false) => vec![("1.66", 96), ("1.60", 120)],
-            (Linux32, false) => vec![("1.66", 96), ("1.62", 120), ("1.60", 72)],
-            (MacOS64, false) => vec![("1.62", 96)],
+            (Linux64 | Linux32Arm, true) => vec![("1.51", 56)],
+            (Linux32X86, true) => vec![("1.51", 48)],
+            (Linux32Mips, true) => unimplemented!(),
+            (MacOS64, true) => vec![("1.62", 56)],
+            (Linux64, false) => vec![("1.66", 104), ("1.60", 128)],
+            (Linux32X86, false) => unimplemented!(),
+            (Linux32Arm | Linux32Mips, false) => vec![("1.66", 104), ("1.62", 128), ("1.60", 80)],
+            (MacOS64, false) => vec![("1.62", 104)],
         };
 
         let mut expected = None;