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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
// Copyright 2021 Twitter, Inc.
// Copyright 2023 Pelikan Cache contributors
// Licensed under the MIT and Apache-2.0 licenses
//! Core datastructure
use crate::Value;
use crate::*;
use core::hash::{BuildHasher, Hasher};
use std::cmp::min;
const RESERVE_RETRIES: usize = 3;
/// A pre-allocated key-value store with eager expiration. It uses a
/// segment-structured design that stores data in fixed-size segments, grouping
/// objects with nearby expiration time into the same segment, and lifting most
/// per-object metadata into the shared segment header.
pub struct Segcache {
pub(crate) hashtable: HashTable,
pub(crate) segments: Segments,
pub(crate) ttl_buckets: TtlBuckets,
pub(crate) time: Instant,
}
impl Segcache {
/// Returns a new `Builder` which is used to configure and construct a
/// `Segcache` instance.
///
/// ```
/// use segcache::{Policy, Segcache};
///
/// const MB: usize = 1024 * 1024;
///
/// // create a heap using 1MB segments
/// let cache = Segcache::builder()
/// .heap_size(64 * MB)
/// .segment_size(1 * MB as i32)
/// .hash_power(16)
/// .eviction(Policy::Random).build().expect("failed to create cache");
/// ```
pub fn builder() -> Builder {
Builder::default()
}
/// Gets a count of items in the `Segcache` instance. This is an expensive
/// operation and is only enabled for tests and builds with the `debug`
/// feature enabled.
///
/// ```
/// use segcache::{Policy, Segcache};
///
/// let mut cache = Segcache::builder().build().expect("failed to create cache");
/// assert_eq!(cache.items(), 0);
/// ```
#[cfg(any(test, feature = "debug"))]
pub fn items(&mut self) -> usize {
trace!("getting segment item counts");
self.segments.items()
}
/// Get the item in the `Segcache` with the provided key
///
/// ```
/// use segcache::{Policy, Segcache};
/// use std::time::Duration;
///
/// let mut cache = Segcache::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(b"coffee").expect("didn't get item back");
/// assert_eq!(item.value(), b"strong");
/// ```
pub fn get(&mut self, key: &[u8]) -> Option<Item> {
self.hashtable.get(key, self.time, &mut self.segments)
}
/// Get the item in the `Segcache` with the provided key without
/// increasing the item frequency - useful for combined operations that
/// check for presence - eg replace is a get + set
/// ```
/// use segcache::{Policy, Segcache};
///
/// let mut cache = Segcache::builder().build().expect("failed to create cache");
/// assert!(cache.get_no_freq_incr(b"coffee").is_none());
/// ```
pub fn get_no_freq_incr(&mut self, key: &[u8]) -> Option<Item> {
self.hashtable.get_no_freq_incr(key, &mut self.segments)
}
/// Insert a new item into the cache. May return an error indicating that
/// the insert was not successful.
/// ```
/// use segcache::{Policy, Segcache};
/// use std::time::Duration;
///
/// let mut cache = Segcache::builder().build().expect("failed to create cache");
/// assert!(cache.get(b"drink").is_none());
///
/// cache.insert(b"drink", b"coffee", None, Duration::ZERO);
/// let item = cache.get(b"drink").expect("didn't get item back");
/// assert_eq!(item.value(), b"coffee");
///
/// cache.insert(b"drink", b"whisky", None, Duration::ZERO);
/// let item = cache.get(b"drink").expect("didn't get item back");
/// assert_eq!(item.value(), b"whisky");
/// ```
pub fn insert<'a, T: Into<Value<'a>>>(
&mut self,
key: &'a [u8],
value: T,
optional: Option<&[u8]>,
ttl: std::time::Duration,
) -> Result<(), SegcacheError> {
let value: Value = value.into();
// default optional data is empty
let optional = optional.unwrap_or(&[]);
// calculate size for item
let size = (((ITEM_HDR_SIZE + key.len() + size_of(&value) + optional.len()) >> 3) + 1) << 3;
let ttl = Duration::from_secs(min(u32::MAX as u64, ttl.as_secs()) as u32);
// For S3-FIFO: determine target pool based on ghost queue
let target_pool = if matches!(self.segments.evict_policy(), Policy::S3Fifo { .. }) {
let hash = {
let mut hasher = self.hashtable.hash_builder().build_hasher();
hasher.write(key);
hasher.finish()
};
if self.segments.ghost_contains(hash) {
self.segments.ghost_remove(hash);
SegmentPool::Main
} else {
SegmentPool::Admission
}
} else {
SegmentPool::Main
};
// For S3-FIFO: ensure the target pool has room by evicting from it
// if it's at capacity. This enforces the small/main ratio computed
// at construction time.
if matches!(self.segments.evict_policy(), Policy::S3Fifo { .. })
&& !self.segments.pool_has_room(target_pool)
{
let _ = self
.segments
.evict(&mut self.ttl_buckets, &mut self.hashtable);
}
// try to get a `ReservedItem`
let mut retries = RESERVE_RETRIES;
let reserved;
loop {
match self
.ttl_buckets
.get_mut_bucket(ttl)
.reserve(size, &mut self.segments)
{
Ok(mut reserved_item) => {
reserved_item.define(key, value, optional);
// Set the segment pool for S3-FIFO (only transitions
// Main→Admission need a counter update; fresh segments
// default to Main)
if let Ok(mut seg) = self.segments.get_mut(reserved_item.seg()) {
if target_pool == SegmentPool::Admission
&& seg.pool() != SegmentPool::Admission
{
seg.set_pool(target_pool);
self.segments.incr_pool(SegmentPool::Admission);
}
}
reserved = reserved_item;
break;
}
Err(TtlBucketsError::ItemOversized { size }) => {
return Err(SegcacheError::ItemOversized { size });
}
Err(TtlBucketsError::NoFreeSegments) => {
if self
.segments
.evict(&mut self.ttl_buckets, &mut self.hashtable)
.is_err()
{
retries -= 1;
} else {
// we successfully evicted a segment, return to start of
// loop to reserve the item
continue;
}
}
}
if retries == 0 {
// segment acquire failed, increment the stats and return with
// an error
#[cfg(feature = "metrics")]
{
SEGMENT_REQUEST.increment();
SEGMENT_REQUEST_FAILURE.increment();
}
return Err(SegcacheError::NoFreeSegments);
}
retries -= 1;
}
// insert into the hashtable, or roll-back by removing the item
// TODO(bmartin): we can probably roll-back the offset and re-use the
// space in the segment, currently we consume the space even if the
// hashtable is overfull
if self
.hashtable
.insert(
reserved.item(),
reserved.seg(),
reserved.offset() as u64,
&mut self.ttl_buckets,
&mut self.segments,
)
.is_err()
{
// this just needs to alter the segment header and update stats
let _ = self.segments.remove_at(
reserved.seg(),
reserved.offset(),
&mut self.ttl_buckets,
&mut self.hashtable,
);
Err(SegcacheError::HashTableInsertEx)
} else {
Ok(())
}
}
/// Performs a CAS operation, inserting the item only if the CAS value
/// matches the current value for that item.
///
/// ```
/// use segcache::{Policy, Segcache, SegcacheError};
/// use std::time::Duration;
///
/// let mut cache = Segcache::builder().build().expect("failed to create cache");
///
/// // If the item is not in the cache, CAS will fail as 'NotFound'
/// assert_eq!(
/// cache.cas(b"drink", b"coffee", None, Duration::ZERO, 0),
/// Err(SegcacheError::NotFound)
/// );
///
/// // If a stale CAS value is provided, CAS will fail as 'Exists'
/// cache.insert(b"drink", b"coffee", None, Duration::ZERO);
/// assert_eq!(
/// cache.cas(b"drink", b"coffee", None, Duration::ZERO, 0),
/// Err(SegcacheError::Exists)
/// );
///
/// // Getting the CAS value and then performing the operation ensures
/// // success in absence of a race with another client
/// let current = cache.get(b"drink").expect("not found");
/// assert!(cache.cas(b"drink", b"whisky", None, Duration::ZERO, current.cas()).is_ok());
/// let item = cache.get(b"drink").expect("not found");
/// assert_eq!(item.value(), b"whisky"); // item is updated
/// ```
pub fn cas<'a, T: Into<Value<'a>>>(
&mut self,
key: &'a [u8],
value: T,
optional: Option<&[u8]>,
ttl: std::time::Duration,
cas: u32,
) -> Result<(), SegcacheError> {
match self.hashtable.try_update_cas(key, cas, &mut self.segments) {
Ok(()) => self.insert(key, value, optional, ttl),
Err(e) => Err(e),
}
}
/// Remove the item with the given key, returns a bool indicating if it was
/// removed.
/// ```
/// use segcache::{Policy, Segcache, SegcacheError};
/// use std::time::Duration;
///
/// let mut cache = Segcache::builder().build().expect("failed to create cache");
///
/// // If the item is not in the cache, delete will return false
/// assert_eq!(cache.delete(b"coffee"), false);
///
/// // And will return true on success
/// cache.insert(b"coffee", b"strong", None, Duration::ZERO);
/// assert!(cache.get(b"coffee").is_some());
/// assert_eq!(cache.delete(b"coffee"), true);
/// assert!(cache.get(b"coffee").is_none());
/// ```
// TODO(bmartin): a result would be better here
pub fn delete(&mut self, key: &[u8]) -> bool {
self.hashtable
.delete(key, &mut self.ttl_buckets, &mut self.segments)
}
/// Loops through the TTL Buckets to handle eager expiration, returns the
/// number of segments expired
/// ```
/// use segcache::{Policy, Segcache, SegcacheError};
/// use std::time::Duration;
///
/// let mut cache = Segcache::builder().build().expect("failed to create cache");
///
/// // Insert an item with a short ttl
/// cache.insert(b"coffee", b"strong", None, Duration::from_secs(5));
///
/// // The item is still in the cache
/// assert!(cache.get(b"coffee").is_some());
///
/// // Delay and then trigger expiration
/// std::thread::sleep(Duration::from_secs(6));
/// cache.expire();
///
/// // And the expired item is not in the cache
/// assert!(cache.get(b"coffee").is_none());
/// ```
pub fn expire(&mut self) -> usize {
self.time = Instant::now();
self.ttl_buckets
.expire(&mut self.hashtable, &mut self.segments)
}
pub fn clear(&mut self) -> usize {
self.time = Instant::now();
self.ttl_buckets
.clear(&mut self.hashtable, &mut self.segments)
}
/// Checks the integrity of all segments
/// *NOTE*: this operation is relatively expensive
#[cfg(feature = "debug")]
pub fn check_integrity(&mut self) -> Result<(), SegcacheError> {
if self.segments.check_integrity(&mut self.hashtable) {
Ok(())
} else {
Err(SegcacheError::DataCorrupted)
}
}
/// Perform a wrapping addition on the value stored at the supplied key.
/// Returns an error if the key is invalid, the item is not found, or the
/// stored value is not a numeric type.
pub fn wrapping_add(&mut self, key: &[u8], rhs: u64) -> Result<Item, SegcacheError> {
let mut item = self
.hashtable
.get(key, self.time, &mut self.segments)
.ok_or(SegcacheError::NotFound)?;
item.wrapping_add(rhs)?;
Ok(item)
}
/// Perform a saturating subtraction on the value stored at the supplied
/// key. Returns an error if the key is invalid, the item is not found, or
/// the stored value is not a numeric type.
pub fn saturating_sub(&mut self, key: &[u8], rhs: u64) -> Result<Item, SegcacheError> {
let mut item = self
.hashtable
.get(key, self.time, &mut self.segments)
.ok_or(SegcacheError::NotFound)?;
item.saturating_sub(rhs)?;
Ok(item)
}
}