From 9a956d5bfb093f1c65c2f8c10903db3468ca8a69 Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Fri, 22 Jul 2022 18:16:55 -0700
Subject: [PATCH 01/13] add new function to get age

---
 src/storage/seg/src/hashtable/mod.rs     | 21 +++++++++++++++++++++
 src/storage/seg/src/seg.rs               |  4 ++++
 src/storage/seg/src/segments/segments.rs |  5 +++++
 3 files changed, 30 insertions(+)

diff --git a/src/storage/seg/src/hashtable/mod.rs b/src/storage/seg/src/hashtable/mod.rs
index c2bd6b896..2de21805b 100644
--- a/src/storage/seg/src/hashtable/mod.rs
+++ b/src/storage/seg/src/hashtable/mod.rs
@@ -322,6 +322,27 @@ impl HashTable {
         None
     }
 
+    /// Lookup an item by key and return it
+    pub fn get_age(&mut self, key: &[u8], segments: &mut Segments) -> Option<u32> {
+        let hash = self.hash(key);
+        let tag = tag_from_hash(hash);
+
+        let iter = IterMut::new(self, hash);
+
+        for item_info in iter {
+            if get_tag(*item_info) == tag {
+                let current_item = segments.get_item(*item_info).unwrap();
+                if current_item.key() != key {
+                    HASH_TAG_COLLISION.increment();
+                } else {
+                    return segments.get_age(*item_info);
+                }
+            }
+        }
+
+        None
+    }
+
     /// Lookup an item by key and return it without incrementing the item
     /// frequency. This may be used to compose higher-level functions which do
     /// not want a successful item lookup to count as a hit for that item.
diff --git a/src/storage/seg/src/seg.rs b/src/storage/seg/src/seg.rs
index 4ec4f686b..257a6a4d8 100644
--- a/src/storage/seg/src/seg.rs
+++ b/src/storage/seg/src/seg.rs
@@ -84,6 +84,10 @@ impl Seg {
         self.hashtable.get(key, self.time, &mut self.segments)
     }
 
+    pub fn get_age(&mut self, key: &[u8]) -> Option<u32> {
+        self.hashtable.get_age(key, &mut self.segments)
+    }
+
     /// Get the item in the `Seg` with the provided key without
     /// increasing the item frequency - useful for combined operations that
     /// check for presence - eg replace is a get + set
diff --git a/src/storage/seg/src/segments/segments.rs b/src/storage/seg/src/segments/segments.rs
index cde24ce8a..e39caf1ca 100644
--- a/src/storage/seg/src/segments/segments.rs
+++ b/src/storage/seg/src/segments/segments.rs
@@ -147,6 +147,11 @@ impl Segments {
         self.get_item_at(seg_id, offset)
     }
 
+    pub(crate) fn get_age(&self, item_info: u64) -> Option<u32> {
+        let seg_id = get_seg_id(item_info).map(|v| v.get())?;
+        return Some(self.headers[seg_id as usize - 1].create_at().elapsed().as_secs());
+    }
+
     /// Retrieve a `RawItem` from a specific segment id at the given offset
     // TODO(bmartin): consider changing the return type here and removing asserts?
     pub(crate) fn get_item_at(

From 80dcd620721b86fa9e74fe4675e9d33b7f56970c Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Fri, 22 Jul 2022 19:42:07 -0700
Subject: [PATCH 02/13] add age field to item

---
 src/storage/seg/src/hashtable/mod.rs | 4 ++++
 src/storage/seg/src/item/mod.rs      | 9 +++++++--
 2 files changed, 11 insertions(+), 2 deletions(-)

diff --git a/src/storage/seg/src/hashtable/mod.rs b/src/storage/seg/src/hashtable/mod.rs
index 2de21805b..0351e409e 100644
--- a/src/storage/seg/src/hashtable/mod.rs
+++ b/src/storage/seg/src/hashtable/mod.rs
@@ -308,8 +308,10 @@ impl HashTable {
                         *item_info = (*item_info & !FREQ_MASK) | freq;
                     }
 
+                    let age = segments.get_age(*item_info).unwrap();
                     let item = Item::new(
                         current_item,
+                        age, 
                         get_cas(self.data[(hash & self.mask) as usize].data[0]),
                     );
                     item.check_magic();
@@ -359,8 +361,10 @@ impl HashTable {
                 if current_item.key() != key {
                     HASH_TAG_COLLISION.increment();
                 } else {
+                    let age = segments.get_age(*item_info).unwrap();
                     let item = Item::new(
                         current_item,
+                        age, 
                         get_cas(self.data[(hash & self.mask) as usize].data[0]),
                     );
                     item.check_magic();
diff --git a/src/storage/seg/src/item/mod.rs b/src/storage/seg/src/item/mod.rs
index abf944421..fbb1d5762 100644
--- a/src/storage/seg/src/item/mod.rs
+++ b/src/storage/seg/src/item/mod.rs
@@ -21,13 +21,14 @@ pub(crate) use reserved::ReservedItem;
 /// Items are the base unit of data stored within the cache.
 pub struct Item {
     cas: u32,
+    age: u32,
     raw: RawItem,
 }
 
 impl Item {
     /// Creates a new `Item` from its parts
-    pub(crate) fn new(raw: RawItem, cas: u32) -> Self {
-        Item { cas, raw }
+    pub(crate) fn new(raw: RawItem, age: u32, cas: u32) -> Self {
+        Item { cas, age, raw }
     }
 
     /// If the `magic` or `debug` features are enabled, this allows for checking
@@ -56,6 +57,10 @@ impl Item {
         self.cas
     }
 
+    pub fn age(&self) -> u32 {
+        self.age
+    }
+
     /// Borrow the optional data
     pub fn optional(&self) -> Option<&[u8]> {
         self.raw.optional()

From 331e4c1e857cd8e1fea0478bddf3ec8cf5374b5f Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Sun, 24 Jul 2022 14:18:18 -0700
Subject: [PATCH 03/13] * add RichItem to return more information * add support
 for hash table entry verification * this is used to support opportunistic
 concurrency control with multiple readers and a single writer

This PR allows us to use multiple readers
each reader reads the cache without coordination,
after copying/using the value, we verify hash table entry,
if the hash table entry is updated, it means the object is evicted/updated,
we may have read corrupt or stale value, and we should roll back.
---
 src/storage/seg/src/hashtable/mod.rs | 73 ++++++++++++++++++++++++-
 src/storage/seg/src/item/mod.rs      | 81 ++++++++++++++++++++++++++++
 src/storage/seg/src/seg.rs           | 39 ++++++++++++++
 3 files changed, 191 insertions(+), 2 deletions(-)

diff --git a/src/storage/seg/src/hashtable/mod.rs b/src/storage/seg/src/hashtable/mod.rs
index 0351e409e..37ddd6599 100644
--- a/src/storage/seg/src/hashtable/mod.rs
+++ b/src/storage/seg/src/hashtable/mod.rs
@@ -311,7 +311,7 @@ impl HashTable {
                     let age = segments.get_age(*item_info).unwrap();
                     let item = Item::new(
                         current_item,
-                        age, 
+                        age,
                         get_cas(self.data[(hash & self.mask) as usize].data[0]),
                     );
                     item.check_magic();
@@ -345,6 +345,75 @@ impl HashTable {
         None
     }
 
+    /// Lookup an item by key and return it
+    /// compare to get, this is designed to support multiple readers and single writer. 
+    /// because eviction always remove hashtable entry first, 
+    /// so if an object is evicted, its hash table entry must have been removed, 
+    /// as a result, we can verify hash table entry after reading/copying the value.
+    /// 
+    /// Therefore, we can leverage opportunistic concurrency control to support
+    /// multiple readers and a single writer. 
+    /// we check the hash table after a reader reads the data, 
+    /// if the data is evicted, then its hash table entry must have been removed.
+    ///  
+    pub fn get_with_item_info(&mut self, key: &[u8], time: Instant, segments: &mut Segments) -> Option<RichItem> {
+        let hash = self.hash(key);
+        let tag = tag_from_hash(hash);
+        let bucket_id = hash & self.mask;
+
+        let bucket_info = self.data[bucket_id as usize].data[0];
+
+        let curr_ts = (time - self.started).as_secs() & PROC_TS_MASK;
+
+        if curr_ts != get_ts(bucket_info) as u32 {
+            self.data[bucket_id as usize].data[0] = (bucket_info & !TS_MASK) | (curr_ts as u64);
+
+            let iter = IterMut::new(self, hash);
+            for item_info in iter {
+                *item_info &= CLEAR_FREQ_SMOOTH_MASK;
+            }
+        }
+
+        let iter = IterMut::new(self, hash);
+
+        for item_info in iter {
+            let item_info_val = *item_info;
+            if get_tag(item_info_val) == tag {
+                let current_item = segments.get_item(*item_info).unwrap();
+                if current_item.key() != key {
+                    HASH_TAG_COLLISION.increment();
+                } else {
+                    // update item frequency
+                    let mut freq = get_freq(*item_info);
+                    if freq < 127 {
+                        let rand = thread_rng().gen::<u64>();
+                        if freq <= 16 || rand % freq == 0 {
+                            freq = ((freq + 1) | 0x80) << FREQ_BIT_SHIFT;
+                        } else {
+                            freq = (freq | 0x80) << FREQ_BIT_SHIFT;
+                        }
+                        // TODO: this needs to be atomic
+                        *item_info = (*item_info & !FREQ_MASK) | freq;
+                    }
+
+                    let age = segments.get_age(item_info_val).unwrap();
+                    let item = RichItem::new(
+                        current_item,
+                        age,
+                        item_info_val,
+                        item_info,
+                        get_cas(self.data[(hash & self.mask) as usize].data[0]),
+                    );
+                    item.check_magic();
+
+                    return Some(item);
+                }
+            }
+        }
+
+        None
+    }    
+
     /// Lookup an item by key and return it without incrementing the item
     /// frequency. This may be used to compose higher-level functions which do
     /// not want a successful item lookup to count as a hit for that item.
@@ -364,7 +433,7 @@ impl HashTable {
                     let age = segments.get_age(*item_info).unwrap();
                     let item = Item::new(
                         current_item,
-                        age, 
+                        age,
                         get_cas(self.data[(hash & self.mask) as usize].data[0]),
                     );
                     item.check_magic();
diff --git a/src/storage/seg/src/item/mod.rs b/src/storage/seg/src/item/mod.rs
index fbb1d5762..8aa6e9ba6 100644
--- a/src/storage/seg/src/item/mod.rs
+++ b/src/storage/seg/src/item/mod.rs
@@ -88,6 +88,87 @@ impl std::fmt::Debug for Item {
     }
 }
 
+/// Items are the base unit of data stored within the cache.
+pub struct RichItem {
+    cas: u32,
+    age: u32,
+    item_info: u64,
+    item_info_ptr: *const u64,
+    raw: RawItem,
+}
+
+impl RichItem {
+    /// Creates a new `Item` from its parts
+    pub(crate) fn new(raw: RawItem, age: u32, item_info: u64, item_info_ptr: *const u64, cas: u32) -> Self {
+        RichItem { cas, age, item_info, item_info_ptr, raw }
+    }
+
+    /// If the `magic` or `debug` features are enabled, this allows for checking
+    /// that the magic bytes at the start of an item match the expected value.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the magic bytes are incorrect, indicating that the data has
+    /// become corrupted or the item was loaded from the wrong offset.
+    pub(crate) fn check_magic(&self) {
+        self.raw.check_magic()
+    }
+
+    /// Borrow the item key
+    pub fn key(&self) -> &[u8] {
+        self.raw.key()
+    }
+
+    /// Borrow the item value
+    pub fn value(&self) -> Value {
+        self.raw.value()
+    }
+
+    /// CAS value for the item
+    pub fn cas(&self) -> u32 {
+        self.cas
+    }
+
+    pub fn age(&self) -> u32 {
+        self.age
+    }
+
+    // used to support multi readers and single writer
+    pub fn verify_hashtable_entry(&self) -> bool {
+        unsafe {
+            return self.item_info == *self.item_info_ptr
+        }
+    }
+
+    /// Borrow the optional data
+    pub fn optional(&self) -> Option<&[u8]> {
+        self.raw.optional()
+    }
+
+    /// Perform a wrapping addition on the value. Returns an error if the item
+    /// is not a numeric type.
+    pub fn wrapping_add(&mut self, rhs: u64) -> Result<(), SegError> {
+        self.raw.wrapping_add(rhs)
+    }
+
+    /// Perform a saturating subtraction on the value. Returns an error if the
+    /// item is not a numeric type.
+    pub fn saturating_sub(&mut self, rhs: u64) -> Result<(), SegError> {
+        self.raw.saturating_sub(rhs)
+    }
+}
+
+impl std::fmt::Debug for RichItem {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::result::Result<(), std::fmt::Error> {
+        f.debug_struct("Item")
+            .field("cas", &self.cas())
+            .field("raw", &self.raw)
+            .field("item_info", &self.item_info)
+            .field("item_info_ptr", &self.item_info_ptr)
+            .finish()
+    }
+}
+
 pub fn size_of(value: &Value) -> usize {
     match value {
         Value::Bytes(v) => v.len(),
diff --git a/src/storage/seg/src/seg.rs b/src/storage/seg/src/seg.rs
index 257a6a4d8..d8e83eda9 100644
--- a/src/storage/seg/src/seg.rs
+++ b/src/storage/seg/src/seg.rs
@@ -84,10 +84,49 @@ impl Seg {
         self.hashtable.get(key, self.time, &mut self.segments)
     }
 
+    /// Get the age of the item in the `Seg` with the provided key
+    ///
+    /// ```
+    /// use seg::{Policy, Seg};
+    /// use std::time::Duration;
+    ///
+    /// let mut cache = Seg::builder().build().expect("failed to create cache");
+    /// assert!(cache.get(b"coffee").is_none());
+    ///
+    /// cache.insert(b"coffee", b"strong", None, Duration::ZERO);
+    /// let age = cache.get_age(b"coffee").expect("didn't get item back");
+    /// assert_eq!(age, 0);
+    /// ```
     pub fn get_age(&mut self, key: &[u8]) -> Option<u32> {
         self.hashtable.get_age(key, &mut self.segments)
     }
 
+    /// Get the item in the `Seg` with the provided key
+    /// this differs from get by returning information about hash table entry
+    /// this allows opportunistic concurrency control and enables 
+    /// multiple readers and a single writer 
+    /// To use it, one simply checks the hash table entry does not change after 
+    /// copying/using the item value, if it has changed, it means the item 
+    /// is evicted or updated by another thread and we need to roll back 
+    ///
+    /// ```
+    /// use seg::{Policy, Seg};
+    /// use std::time::Duration;
+    ///
+    /// let mut cache = Seg::builder().build().expect("failed to create cache");
+    /// assert!(cache.get(b"coffee").is_none());
+    ///
+    /// cache.insert(b"coffee", b"strong", None, Duration::ZERO);
+    /// let item = cache.get_with_item_info(b"coffee").expect("didn't get item back");
+    /// assert_eq!(item.value(), b"strong");
+    /// assert!(item.verify_hashtable_entry());
+    /// cache.insert(b"coffee", b"notStrong", None, Duration::ZERO);
+    /// assert!(!item.verify_hashtable_entry()); 
+    /// ```
+    pub fn get_with_item_info(&mut self, key: &[u8]) -> Option<RichItem> {
+        self.hashtable.get_with_item_info(key, self.time, &mut self.segments)
+    }
+
     /// Get the item in the `Seg` with the provided key without
     /// increasing the item frequency - useful for combined operations that
     /// check for presence - eg replace is a get + set

From 5a15cb2e810b9057a7737a360dcabc1a0543f9b0 Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Sun, 24 Jul 2022 14:56:45 -0700
Subject: [PATCH 04/13] add debug function to print item_info

---
 src/storage/seg/src/item/mod.rs | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/src/storage/seg/src/item/mod.rs b/src/storage/seg/src/item/mod.rs
index 8aa6e9ba6..ded63e4cf 100644
--- a/src/storage/seg/src/item/mod.rs
+++ b/src/storage/seg/src/item/mod.rs
@@ -133,6 +133,21 @@ impl RichItem {
         self.age
     }
 
+    // used to support multi readers and single writer
+    pub fn item_info(&self) -> u64 {
+        return self.item_info; 
+    }
+
+    pub fn item_info_ptr(&self) -> *const u64 {
+        return self.item_info_ptr;
+    }
+
+    pub fn item_info_ptr_val(&self) -> u64 {
+        unsafe {
+            return *self.item_info_ptr;
+        }
+    }
+
     // used to support multi readers and single writer
     pub fn verify_hashtable_entry(&self) -> bool {
         unsafe {

From fe8b0ff0d9c6c5b3aaa6fbee04c1b5d3a54fb932 Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Sun, 24 Jul 2022 15:07:00 -0700
Subject: [PATCH 05/13] add debug function to print item_info

---
 src/storage/seg/src/hashtable/mod.rs | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/src/storage/seg/src/hashtable/mod.rs b/src/storage/seg/src/hashtable/mod.rs
index 37ddd6599..316a59d69 100644
--- a/src/storage/seg/src/hashtable/mod.rs
+++ b/src/storage/seg/src/hashtable/mod.rs
@@ -393,9 +393,11 @@ impl HashTable {
                             freq = (freq | 0x80) << FREQ_BIT_SHIFT;
                         }
                         // TODO: this needs to be atomic
+                        // worse case new item insert fails
                         *item_info = (*item_info & !FREQ_MASK) | freq;
                     }
 
+                    println!("{} == {}", item_info_val, *item_info);
                     let age = segments.get_age(item_info_val).unwrap();
                     let item = RichItem::new(
                         current_item,
@@ -404,6 +406,7 @@ impl HashTable {
                         item_info,
                         get_cas(self.data[(hash & self.mask) as usize].data[0]),
                     );
+                    assert!(item.verify_hashtable_entry()); 
                     item.check_magic();
 
                     return Some(item);

From 4b9bf7647adf9fd26bec6e9ec1be39df2008ee2d Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Sun, 24 Jul 2022 15:20:59 -0700
Subject: [PATCH 06/13] debug

---
 src/storage/seg/src/hashtable/mod.rs | 3 +--
 src/storage/seg/src/item/mod.rs      | 3 ++-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/storage/seg/src/hashtable/mod.rs b/src/storage/seg/src/hashtable/mod.rs
index 316a59d69..e0256fc48 100644
--- a/src/storage/seg/src/hashtable/mod.rs
+++ b/src/storage/seg/src/hashtable/mod.rs
@@ -397,12 +397,11 @@ impl HashTable {
                         *item_info = (*item_info & !FREQ_MASK) | freq;
                     }
 
-                    println!("{} == {}", item_info_val, *item_info);
                     let age = segments.get_age(item_info_val).unwrap();
                     let item = RichItem::new(
                         current_item,
                         age,
-                        item_info_val,
+                        item_info_val & !FREQ_MASK,
                         item_info,
                         get_cas(self.data[(hash & self.mask) as usize].data[0]),
                     );
diff --git a/src/storage/seg/src/item/mod.rs b/src/storage/seg/src/item/mod.rs
index ded63e4cf..27d364e20 100644
--- a/src/storage/seg/src/item/mod.rs
+++ b/src/storage/seg/src/item/mod.rs
@@ -13,6 +13,7 @@ pub(crate) use header::ITEM_MAGIC_SIZE;
 
 use crate::SegError;
 use crate::Value;
+use crate::hashtable::FREQ_MASK;
 
 pub(crate) use header::{ItemHeader, ITEM_HDR_SIZE};
 pub(crate) use raw::RawItem;
@@ -151,7 +152,7 @@ impl RichItem {
     // used to support multi readers and single writer
     pub fn verify_hashtable_entry(&self) -> bool {
         unsafe {
-            return self.item_info == *self.item_info_ptr
+            return self.item_info == *self.item_info_ptr & !FREQ_MASK
         }
     }
 

From 87096f213f1d284faf88595fa2c4d26036281fd8 Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Sun, 24 Jul 2022 15:24:16 -0700
Subject: [PATCH 07/13] rename and remove debug related information and
 functions

---
 src/storage/seg/src/hashtable/mod.rs |  1 -
 src/storage/seg/src/item/mod.rs      | 19 +++----------------
 src/storage/seg/src/seg.rs           |  4 ++--
 3 files changed, 5 insertions(+), 19 deletions(-)

diff --git a/src/storage/seg/src/hashtable/mod.rs b/src/storage/seg/src/hashtable/mod.rs
index e0256fc48..0e4ca0c0b 100644
--- a/src/storage/seg/src/hashtable/mod.rs
+++ b/src/storage/seg/src/hashtable/mod.rs
@@ -405,7 +405,6 @@ impl HashTable {
                         item_info,
                         get_cas(self.data[(hash & self.mask) as usize].data[0]),
                     );
-                    assert!(item.verify_hashtable_entry()); 
                     item.check_magic();
 
                     return Some(item);
diff --git a/src/storage/seg/src/item/mod.rs b/src/storage/seg/src/item/mod.rs
index 27d364e20..ae70481aa 100644
--- a/src/storage/seg/src/item/mod.rs
+++ b/src/storage/seg/src/item/mod.rs
@@ -135,22 +135,9 @@ impl RichItem {
     }
 
     // used to support multi readers and single writer
-    pub fn item_info(&self) -> u64 {
-        return self.item_info; 
-    }
-
-    pub fn item_info_ptr(&self) -> *const u64 {
-        return self.item_info_ptr;
-    }
-
-    pub fn item_info_ptr_val(&self) -> u64 {
-        unsafe {
-            return *self.item_info_ptr;
-        }
-    }
-
-    // used to support multi readers and single writer
-    pub fn verify_hashtable_entry(&self) -> bool {
+    // return true, if the item is evicted/updated since being 
+    // read from the hash table
+    pub fn is_not_changed(&self) -> bool {
         unsafe {
             return self.item_info == *self.item_info_ptr & !FREQ_MASK
         }
diff --git a/src/storage/seg/src/seg.rs b/src/storage/seg/src/seg.rs
index d8e83eda9..8b51d076b 100644
--- a/src/storage/seg/src/seg.rs
+++ b/src/storage/seg/src/seg.rs
@@ -119,9 +119,9 @@ impl Seg {
     /// cache.insert(b"coffee", b"strong", None, Duration::ZERO);
     /// let item = cache.get_with_item_info(b"coffee").expect("didn't get item back");
     /// assert_eq!(item.value(), b"strong");
-    /// assert!(item.verify_hashtable_entry());
+    /// assert!(item.is_not_changed());
     /// cache.insert(b"coffee", b"notStrong", None, Duration::ZERO);
-    /// assert!(!item.verify_hashtable_entry()); 
+    /// assert!(!item.is_not_changed());
     /// ```
     pub fn get_with_item_info(&mut self, key: &[u8]) -> Option<RichItem> {
         self.hashtable.get_with_item_info(key, self.time, &mut self.segments)

From 5c1a71a2e6c096f967e8252898e019b5557b001d Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Wed, 27 Jul 2022 20:38:28 -0700
Subject: [PATCH 08/13] remove expired objects before we evict, this allows the
 users of Segcache to avoid calling expire themselves

---
 src/storage/seg/src/seg.rs | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/src/storage/seg/src/seg.rs b/src/storage/seg/src/seg.rs
index 8b51d076b..9f7582a17 100644
--- a/src/storage/seg/src/seg.rs
+++ b/src/storage/seg/src/seg.rs
@@ -176,6 +176,8 @@ impl Seg {
 
         // try to get a `ReservedItem`
         let mut retries = RESERVE_RETRIES;
+        let mut has_removed_expired = false; 
+        
         let reserved;
         loop {
             match self
@@ -192,6 +194,11 @@ impl Seg {
                     return Err(SegError::ItemOversized { size });
                 }
                 Err(TtlBucketsError::NoFreeSegments) => {
+                    if !has_removed_expired {
+                        self.expire();
+                        has_removed_expired = true;
+                        continue;
+                    }
                     if self
                         .segments
                         .evict(&mut self.ttl_buckets, &mut self.hashtable)

From 491060d65803db00354327181918dd0ea749e0b9 Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Thu, 28 Jul 2022 13:05:18 -0700
Subject: [PATCH 09/13] rust format

---
 src/storage/seg/src/hashtable/mod.rs     | 21 +++++++++++++--------
 src/storage/seg/src/item/mod.rs          | 24 +++++++++++++++++-------
 src/storage/seg/src/seg.rs               | 17 +++++++++--------
 src/storage/seg/src/segments/segments.rs |  7 ++++++-
 4 files changed, 45 insertions(+), 24 deletions(-)

diff --git a/src/storage/seg/src/hashtable/mod.rs b/src/storage/seg/src/hashtable/mod.rs
index 0e4ca0c0b..c2fba3d1e 100644
--- a/src/storage/seg/src/hashtable/mod.rs
+++ b/src/storage/seg/src/hashtable/mod.rs
@@ -346,17 +346,22 @@ impl HashTable {
     }
 
     /// Lookup an item by key and return it
-    /// compare to get, this is designed to support multiple readers and single writer. 
-    /// because eviction always remove hashtable entry first, 
-    /// so if an object is evicted, its hash table entry must have been removed, 
+    /// compare to get, this is designed to support multiple readers and single writer.
+    /// because eviction always remove hashtable entry first,
+    /// so if an object is evicted, its hash table entry must have been removed,
     /// as a result, we can verify hash table entry after reading/copying the value.
-    /// 
+    ///
     /// Therefore, we can leverage opportunistic concurrency control to support
-    /// multiple readers and a single writer. 
-    /// we check the hash table after a reader reads the data, 
+    /// multiple readers and a single writer.
+    /// we check the hash table after a reader reads the data,
     /// if the data is evicted, then its hash table entry must have been removed.
     ///  
-    pub fn get_with_item_info(&mut self, key: &[u8], time: Instant, segments: &mut Segments) -> Option<RichItem> {
+    pub fn get_with_item_info(
+        &mut self,
+        key: &[u8],
+        time: Instant,
+        segments: &mut Segments,
+    ) -> Option<RichItem> {
         let hash = self.hash(key);
         let tag = tag_from_hash(hash);
         let bucket_id = hash & self.mask;
@@ -413,7 +418,7 @@ impl HashTable {
         }
 
         None
-    }    
+    }
 
     /// Lookup an item by key and return it without incrementing the item
     /// frequency. This may be used to compose higher-level functions which do
diff --git a/src/storage/seg/src/item/mod.rs b/src/storage/seg/src/item/mod.rs
index ae70481aa..394cf85aa 100644
--- a/src/storage/seg/src/item/mod.rs
+++ b/src/storage/seg/src/item/mod.rs
@@ -11,9 +11,9 @@ mod reserved;
 #[cfg(any(feature = "magic", feature = "debug"))]
 pub(crate) use header::ITEM_MAGIC_SIZE;
 
+use crate::hashtable::FREQ_MASK;
 use crate::SegError;
 use crate::Value;
-use crate::hashtable::FREQ_MASK;
 
 pub(crate) use header::{ItemHeader, ITEM_HDR_SIZE};
 pub(crate) use raw::RawItem;
@@ -100,8 +100,20 @@ pub struct RichItem {
 
 impl RichItem {
     /// Creates a new `Item` from its parts
-    pub(crate) fn new(raw: RawItem, age: u32, item_info: u64, item_info_ptr: *const u64, cas: u32) -> Self {
-        RichItem { cas, age, item_info, item_info_ptr, raw }
+    pub(crate) fn new(
+        raw: RawItem,
+        age: u32,
+        item_info: u64,
+        item_info_ptr: *const u64,
+        cas: u32,
+    ) -> Self {
+        RichItem {
+            cas,
+            age,
+            item_info,
+            item_info_ptr,
+            raw,
+        }
     }
 
     /// If the `magic` or `debug` features are enabled, this allows for checking
@@ -135,12 +147,10 @@ impl RichItem {
     }
 
     // used to support multi readers and single writer
-    // return true, if the item is evicted/updated since being 
+    // return true, if the item is evicted/updated since being
     // read from the hash table
     pub fn is_not_changed(&self) -> bool {
-        unsafe {
-            return self.item_info == *self.item_info_ptr & !FREQ_MASK
-        }
+        unsafe { return self.item_info == *self.item_info_ptr & !FREQ_MASK }
     }
 
     /// Borrow the optional data
diff --git a/src/storage/seg/src/seg.rs b/src/storage/seg/src/seg.rs
index 9f7582a17..29b93cda0 100644
--- a/src/storage/seg/src/seg.rs
+++ b/src/storage/seg/src/seg.rs
@@ -103,11 +103,11 @@ impl Seg {
 
     /// Get the item in the `Seg` with the provided key
     /// this differs from get by returning information about hash table entry
-    /// this allows opportunistic concurrency control and enables 
-    /// multiple readers and a single writer 
-    /// To use it, one simply checks the hash table entry does not change after 
-    /// copying/using the item value, if it has changed, it means the item 
-    /// is evicted or updated by another thread and we need to roll back 
+    /// this allows opportunistic concurrency control and enables
+    /// multiple readers and a single writer
+    /// To use it, one simply checks the hash table entry does not change after
+    /// copying/using the item value, if it has changed, it means the item
+    /// is evicted or updated by another thread and we need to roll back
     ///
     /// ```
     /// use seg::{Policy, Seg};
@@ -124,7 +124,8 @@ impl Seg {
     /// assert!(!item.is_not_changed());
     /// ```
     pub fn get_with_item_info(&mut self, key: &[u8]) -> Option<RichItem> {
-        self.hashtable.get_with_item_info(key, self.time, &mut self.segments)
+        self.hashtable
+            .get_with_item_info(key, self.time, &mut self.segments)
     }
 
     /// Get the item in the `Seg` with the provided key without
@@ -176,8 +177,8 @@ impl Seg {
 
         // try to get a `ReservedItem`
         let mut retries = RESERVE_RETRIES;
-        let mut has_removed_expired = false; 
-        
+        let mut has_removed_expired = false;
+
         let reserved;
         loop {
             match self
diff --git a/src/storage/seg/src/segments/segments.rs b/src/storage/seg/src/segments/segments.rs
index e39caf1ca..2781a660b 100644
--- a/src/storage/seg/src/segments/segments.rs
+++ b/src/storage/seg/src/segments/segments.rs
@@ -149,7 +149,12 @@ impl Segments {
 
     pub(crate) fn get_age(&self, item_info: u64) -> Option<u32> {
         let seg_id = get_seg_id(item_info).map(|v| v.get())?;
-        return Some(self.headers[seg_id as usize - 1].create_at().elapsed().as_secs());
+        return Some(
+            self.headers[seg_id as usize - 1]
+                .create_at()
+                .elapsed()
+                .as_secs(),
+        );
     }
 
     /// Retrieve a `RawItem` from a specific segment id at the given offset

From f37078664ca4d78dba0db4e7f2a26433c2726e25 Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Thu, 28 Jul 2022 13:10:17 -0700
Subject: [PATCH 10/13] add check for key size

---
 src/storage/seg/src/error.rs | 2 ++
 src/storage/seg/src/seg.rs   | 4 ++++
 2 files changed, 6 insertions(+)

diff --git a/src/storage/seg/src/error.rs b/src/storage/seg/src/error.rs
index 0c92e99c5..fc695cfdc 100644
--- a/src/storage/seg/src/error.rs
+++ b/src/storage/seg/src/error.rs
@@ -9,6 +9,8 @@ use thiserror::Error;
 #[derive(Error, Debug, PartialEq, Eq, Copy, Clone)]
 /// Possible errors returned by the top-level API
 pub enum SegError {
+    #[error("key size too large")]
+    KeySizeTooLargeEx,
     #[error("hashtable insert exception")]
     HashTableInsertEx,
     #[error("eviction exception")]
diff --git a/src/storage/seg/src/seg.rs b/src/storage/seg/src/seg.rs
index 29b93cda0..b757a8d65 100644
--- a/src/storage/seg/src/seg.rs
+++ b/src/storage/seg/src/seg.rs
@@ -167,6 +167,10 @@ impl Seg {
     ) -> Result<(), SegError> {
         let value: Value = value.into();
 
+        if key.len() > 255 {
+            return Err(SegError::KeySizeTooLargeEx);
+        }
+
         // default optional data is empty
         let optional = optional.unwrap_or(&[]);
 

From 14cb8ecaf93912082fc43ca6f9266059cabb1bf0 Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Thu, 28 Jul 2022 17:24:38 -0700
Subject: [PATCH 11/13] add RichItem to public interface; fix according to
 comments

---
 src/storage/seg/src/hashtable/mod.rs |  5 ++--
 src/storage/seg/src/item/mod.rs      | 37 ++++++++++++++++------------
 src/storage/seg/src/lib.rs           |  2 +-
 3 files changed, 25 insertions(+), 19 deletions(-)

diff --git a/src/storage/seg/src/hashtable/mod.rs b/src/storage/seg/src/hashtable/mod.rs
index c2fba3d1e..de2fcb1f9 100644
--- a/src/storage/seg/src/hashtable/mod.rs
+++ b/src/storage/seg/src/hashtable/mod.rs
@@ -379,6 +379,7 @@ impl HashTable {
             }
         }
 
+        let cas = get_cas(self.data[(hash & self.mask) as usize].data[0]); 
         let iter = IterMut::new(self, hash);
 
         for item_info in iter {
@@ -406,9 +407,9 @@ impl HashTable {
                     let item = RichItem::new(
                         current_item,
                         age,
+                        cas,
                         item_info_val & !FREQ_MASK,
-                        item_info,
-                        get_cas(self.data[(hash & self.mask) as usize].data[0]),
+                        item_info
                     );
                     item.check_magic();
 
diff --git a/src/storage/seg/src/item/mod.rs b/src/storage/seg/src/item/mod.rs
index 394cf85aa..e35a24a13 100644
--- a/src/storage/seg/src/item/mod.rs
+++ b/src/storage/seg/src/item/mod.rs
@@ -91,11 +91,9 @@ impl std::fmt::Debug for Item {
 
 /// Items are the base unit of data stored within the cache.
 pub struct RichItem {
-    cas: u32,
-    age: u32,
+    item: Item,
     item_info: u64,
     item_info_ptr: *const u64,
-    raw: RawItem,
 }
 
 impl RichItem {
@@ -103,16 +101,15 @@ impl RichItem {
     pub(crate) fn new(
         raw: RawItem,
         age: u32,
+        cas: u32,
         item_info: u64,
         item_info_ptr: *const u64,
-        cas: u32,
     ) -> Self {
+        let item = Item::new(raw, age, cas);
         RichItem {
-            cas,
-            age,
+            item,
             item_info,
             item_info_ptr,
-            raw,
         }
     }
 
@@ -124,26 +121,34 @@ impl RichItem {
     /// Panics if the magic bytes are incorrect, indicating that the data has
     /// become corrupted or the item was loaded from the wrong offset.
     pub(crate) fn check_magic(&self) {
-        self.raw.check_magic()
+        self.item.raw.check_magic()
     }
 
     /// Borrow the item key
     pub fn key(&self) -> &[u8] {
-        self.raw.key()
+        self.item.raw.key()
     }
 
     /// Borrow the item value
     pub fn value(&self) -> Value {
-        self.raw.value()
+        self.item.raw.value()
     }
 
     /// CAS value for the item
     pub fn cas(&self) -> u32 {
-        self.cas
+        self.item.cas
     }
 
     pub fn age(&self) -> u32 {
-        self.age
+        self.item.age
+    }
+
+    pub fn item(&self) -> &Item {
+        &self.item
+    }
+
+    pub fn item_mut(&mut self) -> &mut Item {
+        &mut self.item
     }
 
     // used to support multi readers and single writer
@@ -155,19 +160,19 @@ impl RichItem {
 
     /// Borrow the optional data
     pub fn optional(&self) -> Option<&[u8]> {
-        self.raw.optional()
+        self.item.raw.optional()
     }
 
     /// Perform a wrapping addition on the value. Returns an error if the item
     /// is not a numeric type.
     pub fn wrapping_add(&mut self, rhs: u64) -> Result<(), SegError> {
-        self.raw.wrapping_add(rhs)
+        self.item.raw.wrapping_add(rhs)
     }
 
     /// Perform a saturating subtraction on the value. Returns an error if the
     /// item is not a numeric type.
     pub fn saturating_sub(&mut self, rhs: u64) -> Result<(), SegError> {
-        self.raw.saturating_sub(rhs)
+        self.item.raw.saturating_sub(rhs)
     }
 }
 
@@ -175,7 +180,7 @@ impl std::fmt::Debug for RichItem {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::result::Result<(), std::fmt::Error> {
         f.debug_struct("Item")
             .field("cas", &self.cas())
-            .field("raw", &self.raw)
+            .field("raw", &self.item.raw)
             .field("item_info", &self.item_info)
             .field("item_info_ptr", &self.item_info_ptr)
             .finish()
diff --git a/src/storage/seg/src/lib.rs b/src/storage/seg/src/lib.rs
index 34f64a300..c797cc042 100644
--- a/src/storage/seg/src/lib.rs
+++ b/src/storage/seg/src/lib.rs
@@ -58,7 +58,7 @@ pub use crate::seg::Seg;
 pub use builder::Builder;
 pub use error::SegError;
 pub use eviction::Policy;
-pub use item::Item;
+pub use item::{Item, RichItem};
 
 // publicly exported items from external crates
 pub use storage_types::Value;

From 2a5e5e69f3d8f1618d8d0c66af5a3d4c90652798 Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Thu, 28 Jul 2022 17:48:43 -0700
Subject: [PATCH 12/13] remove seg_mature_time, current value causes some
 benchmarks to fail to write new data

---
 src/storage/seg/src/segments/header.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/storage/seg/src/segments/header.rs b/src/storage/seg/src/segments/header.rs
index 5bbed133a..bb5db7b3b 100644
--- a/src/storage/seg/src/segments/header.rs
+++ b/src/storage/seg/src/segments/header.rs
@@ -33,7 +33,7 @@ use crate::*;
 
 // the minimum age of a segment before it is eligible for eviction
 // TODO(bmartin): this should be parameterized.
-const SEG_MATURE_TIME: Duration = Duration::from_secs(20);
+const SEG_MATURE_TIME: Duration = Duration::from_secs(0);
 
 #[derive(Debug)]
 #[repr(C)]

From c3c5dd6395ebbf364250463de4cc1bf36f406110 Mon Sep 17 00:00:00 2001
From: Juncheng Yang <junchengy@twitter.com>
Date: Thu, 28 Jul 2022 18:37:59 -0700
Subject: [PATCH 13/13] update

---
 src/storage/seg/src/hashtable/mod.rs | 2 --
 1 file changed, 2 deletions(-)

diff --git a/src/storage/seg/src/hashtable/mod.rs b/src/storage/seg/src/hashtable/mod.rs
index de2fcb1f9..8d4d4a010 100644
--- a/src/storage/seg/src/hashtable/mod.rs
+++ b/src/storage/seg/src/hashtable/mod.rs
@@ -398,8 +398,6 @@ impl HashTable {
                         } else {
                             freq = (freq | 0x80) << FREQ_BIT_SHIFT;
                         }
-                        // TODO: this needs to be atomic
-                        // worse case new item insert fails
                         *item_info = (*item_info & !FREQ_MASK) | freq;
                     }