1use anyhow::Result;
2use byte_unit::Byte;
3use downcast_rs::{DowncastSync, impl_downcast};
4use fxhash::FxBuildHasher;
5use log::error;
6use moka::{future::Cache, notification::RemovalCause};
7use std::str::FromStr;
8use std::{
9 fs,
10 path::Path,
11 sync::{
12 Arc,
13 atomic::{AtomicU64, Ordering},
14 },
15};
16
17use crate::ShardId;
18use crate::{cache::fast_cache::FastCache, cache::storage_cache::StorageCache};
19
20use super::msec::{MSEC_SHR, MSec, get_cache_time};
21
22const MOKA_BASE_MEMORY: u32 = 400;
23const DEFAULT_FAST_CACHE_INDEX_SIZE: &str = "1MiB";
24const DEFAULT_SHORT_CACHE_CAPACITY: &str = "8MiB";
25const DEFAULT_SHORT_CACHE_TIME: &str = "60";
26const DEFAULT_LONG_CACHE_CAPACITY: &str = "32MiB";
27const DEFAULT_LONG_CACHE_TIME: &str = "86400";
28const DEFAULT_LONG_CACHE_IDLE_TIME: &str = "86400";
29const DEFAULT_DISK_CACHE_INDEX_SIZE: &str = "8MiB";
30const DEFAULT_DISK_CACHE_FILE_NUM: &str = "1";
31const DEFAULT_DISK_CACHE_FILE_SIZE: &str = "100MiB";
32const DEFAULT_CACHE_TTL: &str = "86400";
33const DISK_CACHE_FILE_NAME: &str = "cache-%Y%m%d%H%M%S";
34
35pub trait CacheVal: DowncastSync + std::fmt::Debug {
36 fn _size(&self) -> u32;
37 fn _type_id(&self) -> u64;
38 fn __type_id() -> u64
39 where
40 Self: Sized;
41 fn _shard_id(&self) -> ShardId;
42 fn _time(&self) -> MSec;
43 fn _estimate() -> usize
44 where
45 Self: Sized;
46 fn _encode(&self) -> Result<Vec<u8>>;
47 fn _decode(v: &[u8]) -> Result<Self>
48 where
49 Self: Sized;
50}
51impl_downcast!(sync CacheVal);
52
53pub trait HashVal: Send + Sync {
54 fn hash_val(&self, shard_id: ShardId) -> u128;
55}
56
57fn get_fast_cache(name: &str, time_to_live: u64) -> FastCache {
58 let index_size = Byte::from_str(
59 &std::env::var(format!("{}_FAST_CACHE_INDEX_SIZE", name))
60 .unwrap_or_else(|_| DEFAULT_FAST_CACHE_INDEX_SIZE.to_owned()),
61 )
62 .unwrap_or_else(|e| panic!("{}_FAST_CACHE_INDEX_SIZE has an error:{:?}", name, e))
63 .as_u64();
64 FastCache::new(index_size, time_to_live)
65}
66
67fn get_short_cache(
68 name: &str,
69 short_cache_evicted: Arc<AtomicU64>,
70) -> Cache<u128, Arc<dyn CacheVal>, FxBuildHasher> {
71 let capacity = Byte::from_str(
72 &std::env::var(format!("{}_SHORT_CACHE_CAPACITY", name))
73 .unwrap_or_else(|_| DEFAULT_SHORT_CACHE_CAPACITY.to_owned()),
74 )
75 .unwrap_or_else(|e| panic!("{}_SHORT_CACHE_CAPACITY has an error:{:?}", name, e))
76 .as_u64();
77
78 let time_to_live = std::env::var(format!("{}_SHORT_CACHE_TIME", name))
79 .unwrap_or_else(|_| DEFAULT_SHORT_CACHE_TIME.to_owned())
80 .parse::<u64>()
81 .unwrap_or_else(|e| panic!("{}_SHORT_CACHE_TIME has an error:{:?}", name, e));
82
83 Cache::builder()
84 .weigher(|_key, value: &Arc<dyn CacheVal>| -> u32 {
85 value._size().saturating_add(MOKA_BASE_MEMORY)
86 })
87 .max_capacity(capacity)
88 .time_to_live(std::time::Duration::from_secs(time_to_live))
89 .support_invalidation_closures()
90 .eviction_listener(move |_k, _v, cause| {
91 if cause == RemovalCause::Size {
92 short_cache_evicted.fetch_add(1, Ordering::Relaxed);
93 }
94 })
95 .build_with_hasher(FxBuildHasher::default())
96}
97
98fn get_long_cache(
99 name: &str,
100 long_cache_evicted: Arc<AtomicU64>,
101 storage_cache: Option<Arc<StorageCache>>,
102) -> Cache<u128, Arc<dyn CacheVal>, FxBuildHasher> {
103 let capacity = Byte::from_str(
104 &std::env::var(format!("{}_LONG_CACHE_CAPACITY", name))
105 .unwrap_or_else(|_| DEFAULT_LONG_CACHE_CAPACITY.to_owned()),
106 )
107 .unwrap_or_else(|e| panic!("{}_LONG_CACHE_CAPACITY has an error:{:?}", name, e))
108 .as_u64();
109
110 let time_to_live = std::env::var(format!("{}_LONG_CACHE_TIME", name))
111 .unwrap_or_else(|_| DEFAULT_LONG_CACHE_TIME.to_owned())
112 .parse::<u64>()
113 .unwrap_or_else(|e| panic!("{}_LONG_CACHE_TIME has an error:{:?}", name, e));
114
115 let time_to_idle = std::env::var(format!("{}_LONG_CACHE_IDLE_TIME", name))
116 .unwrap_or_else(|_| DEFAULT_LONG_CACHE_IDLE_TIME.to_owned())
117 .parse::<u64>()
118 .unwrap_or_else(|e| panic!("{}_LONG_CACHE_IDLE_TIME has an error:{:?}", name, e));
119
120 Cache::builder()
121 .weigher(|_key, value: &Arc<dyn CacheVal>| -> u32 {
122 value._size().saturating_add(MOKA_BASE_MEMORY)
123 })
124 .max_capacity(capacity)
125 .time_to_live(std::time::Duration::from_secs(time_to_live))
126 .time_to_idle(std::time::Duration::from_secs(time_to_idle))
127 .support_invalidation_closures()
128 .eviction_listener(move |k, v, cause| {
129 if cause == RemovalCause::Size {
130 long_cache_evicted.fetch_add(1, Ordering::Relaxed);
131 }
132 if cause.was_evicted()
133 && let Some(ref storage_cache) = storage_cache
134 && let Ok(buf) = v._encode()
135 {
136 storage_cache.write(*k, v._type_id(), &buf, v._time());
137 }
138 })
139 .build_with_hasher(FxBuildHasher::default())
140}
141
142fn get_storage_cache(
143 name: &str,
144 is_hot_deploy: bool,
145 path: &Path,
146 time_to_live: u64,
147) -> Result<StorageCache> {
148 let index_size = Byte::from_str(
149 &std::env::var(format!("{}_DISK_CACHE_INDEX_SIZE", name))
150 .unwrap_or_else(|_| DEFAULT_DISK_CACHE_INDEX_SIZE.to_owned()),
151 )
152 .unwrap_or_else(|e| panic!("{}_DISK_CACHE_INDEX_SIZE has an error:{:?}", name, e))
153 .as_u64();
154
155 let file_num = std::env::var(format!("{}_DISK_CACHE_FILE_NUM", name))
156 .unwrap_or_else(|_| DEFAULT_DISK_CACHE_FILE_NUM.to_owned())
157 .parse::<usize>()
158 .unwrap_or_else(|e| panic!("{}_DISK_CACHE_FILE_NUM has an error:{:?}", name, e));
159
160 let file_size = Byte::from_str(
161 &std::env::var(format!("{}_DISK_CACHE_FILE_SIZE", name))
162 .unwrap_or_else(|_| DEFAULT_DISK_CACHE_FILE_SIZE.to_owned()),
163 )
164 .unwrap_or_else(|e| panic!("{}_DISK_CACHE_FILE_SIZE has an error:{:?}", name, e))
165 .as_u64();
166
167 if !is_hot_deploy && path.is_dir() {
168 for entry in path.read_dir()? {
169 let entry = entry?;
170 if entry.metadata()?.is_file() {
171 fs::remove_file(entry.path())?;
172 }
173 }
174 }
175 fs::create_dir_all(path)?;
176 let path = path.join(
177 chrono::Local::now()
178 .format(DISK_CACHE_FILE_NAME)
179 .to_string(),
180 );
181 StorageCache::start(path, index_size, file_num, file_size, time_to_live)
182}
183
184pub struct DbCache {
185 fast_cache: Option<FastCache>,
186 short_cache: Cache<u128, Arc<dyn CacheVal>, FxBuildHasher>,
187 version_cache: Cache<u128, Arc<dyn CacheVal>, FxBuildHasher>,
188 long_cache: Cache<u128, Arc<dyn CacheVal>, FxBuildHasher>,
189 storage_cache: Option<Arc<StorageCache>>,
190 fast_cache_hit: AtomicU64,
191 long_cache_hit: AtomicU64,
192 short_cache_hit: AtomicU64,
193 version_cache_hit: AtomicU64,
194 storage_cache_hit: AtomicU64,
195 cache_request_count: AtomicU64,
196 long_cache_evicted: Arc<AtomicU64>,
197 short_cache_evicted: Arc<AtomicU64>,
198 version_cache_evicted: Arc<AtomicU64>,
199 ttl: u64,
200}
201
202impl DbCache {
203 pub fn start(
204 name: &str,
205 is_hot_deploy: bool,
206 path: Option<&Path>,
207 use_fast_cache: bool,
208 use_storage_cache: bool,
209 ) -> Result<DbCache> {
210 let ttl = std::env::var(format!("{}_CACHE_TTL", name))
211 .unwrap_or_else(|_| DEFAULT_CACHE_TTL.to_owned())
212 .parse::<u64>()
213 .unwrap_or_else(|e| panic!("{}_CACHE_TTL has an error:{:?}", name, e));
214 let ttl = ttl.saturating_mul(1_000_000_000 / (1 << MSEC_SHR));
215
216 let fast_cache = if use_fast_cache {
217 Some(get_fast_cache(name, ttl))
218 } else {
219 None
220 };
221 let storage_cache = if use_storage_cache && let Some(path) = path {
222 Some(Arc::new(get_storage_cache(
223 name,
224 is_hot_deploy,
225 path,
226 ttl,
227 )?))
228 } else {
229 None
230 };
231 let short_cache_evicted = Arc::new(AtomicU64::new(0));
232 let short_cache = get_short_cache(name, Arc::clone(&short_cache_evicted));
233 let version_cache_evicted = Arc::new(AtomicU64::new(0));
234 let version_cache = get_short_cache(name, Arc::clone(&version_cache_evicted));
235 let long_cache_evicted = Arc::new(AtomicU64::new(0));
236 let long_cache =
237 get_long_cache(name, Arc::clone(&long_cache_evicted), storage_cache.clone());
238 Ok(DbCache {
239 fast_cache,
240 short_cache,
241 version_cache,
242 long_cache,
243 storage_cache,
244 fast_cache_hit: AtomicU64::new(0),
245 long_cache_hit: AtomicU64::new(0),
246 short_cache_hit: AtomicU64::new(0),
247 version_cache_hit: AtomicU64::new(0),
248 storage_cache_hit: AtomicU64::new(0),
249 cache_request_count: AtomicU64::new(0),
250 long_cache_evicted,
251 short_cache_evicted,
252 version_cache_evicted,
253 ttl,
254 })
255 }
256
257 pub fn stop(&self) {
258 if let Some(ref storage_cache) = self.storage_cache {
259 storage_cache.stop();
260 }
261 }
262
263 pub async fn insert_short(&self, id: &dyn HashVal, value: Arc<dyn CacheVal>) {
264 let hash = id.hash_val(value._shard_id());
265 self.short_cache.insert(hash, value).await
266 }
267
268 pub async fn insert_version(&self, id: &dyn HashVal, value: Arc<dyn CacheVal>) {
269 let hash = id.hash_val(value._shard_id());
270 self.version_cache.insert(hash, value).await
271 }
272
273 pub async fn insert_long(
274 &self,
275 id: &dyn HashVal,
276 value: Arc<dyn CacheVal>,
277 use_fast_cache: bool,
278 ) {
279 let hash = id.hash_val(value._shard_id());
280 if use_fast_cache && let Some(ref fast_cache) = self.fast_cache {
281 let old = fast_cache.insert(hash, value);
282 if let Some(old) = old {
283 self.long_cache.insert(old.0, old.1).await;
284 }
285 return;
286 }
287 self.long_cache.insert(hash, value).await;
288 }
289
290 pub async fn get<T>(
291 &self,
292 hash: u128,
293 shard_id: ShardId,
294 use_fast_cache: bool,
295 from_memory: bool,
296 ) -> Option<Arc<T>>
297 where
298 T: CacheVal,
299 {
300 let (now, msec) = get_cache_time();
301 self.cache_request_count.fetch_add(1, Ordering::Relaxed);
302
303 if use_fast_cache && let Some(ref fast_cache) = self.fast_cache {
304 let val = fast_cache
305 .get(hash, now, msec)
306 .filter(|v| v._shard_id() == shard_id)
307 .map(|v| v.downcast_arc::<T>().ok())
308 .unwrap_or(None);
309 if val.is_some() {
310 self.fast_cache_hit.fetch_add(1, Ordering::Relaxed);
311 return val;
312 }
313 }
314
315 let val = self
316 .long_cache
317 .get(&hash)
318 .await
319 .filter(|v| v._shard_id() == shard_id)
320 .map(|v| v.downcast_arc::<T>().ok())
321 .unwrap_or(None);
322 if let Some(val) = val {
323 if val._time().less_than_ttl(msec, self.ttl) {
324 return None;
325 }
326 if use_fast_cache && let Some(ref fast_cache) = self.fast_cache {
327 fast_cache.insert(hash, val.clone());
328 }
329 self.long_cache_hit.fetch_add(1, Ordering::Relaxed);
330 return Some(val);
331 }
332
333 let val = self
334 .short_cache
335 .get(&hash)
336 .await
337 .filter(|v| v._shard_id() == shard_id)
338 .map(|v| v.downcast_arc::<T>().ok())
339 .unwrap_or(None);
340 if let Some(val) = val {
341 self.short_cache_hit.fetch_add(1, Ordering::Relaxed);
342 self.long_cache.insert(hash, val.clone()).await;
343 return Some(val);
344 }
345
346 if from_memory {
347 return None;
348 }
349
350 if let Some(ref storage_cache) = self.storage_cache
351 && let Some(buf) = storage_cache
352 .read(hash, T::__type_id(), T::_estimate())
353 .await
354 {
355 match T::_decode(&buf) {
356 Ok(v) => {
357 if v._shard_id() == shard_id {
358 let val = Arc::new(v);
359 self.storage_cache_hit.fetch_add(1, Ordering::Relaxed);
360 self.long_cache.insert(hash, val.clone()).await;
361 return Some(val);
362 }
363 }
364 Err(e) => error!("{}", e),
365 }
366 }
367 None
368 }
369
370 pub async fn get_version<T>(&self, hash: u128, shard_id: ShardId) -> Option<Arc<T>>
371 where
372 T: CacheVal,
373 {
374 self.version_cache
375 .get(&hash)
376 .await
377 .filter(|v| v._shard_id() == shard_id)
378 .map(|v| v.downcast_arc::<T>().ok())
379 .unwrap_or(None)
380 }
381
382 pub async fn invalidate(&self, id: &dyn HashVal, shard_id: ShardId) {
383 if let Some(ref fast_cache) = self.fast_cache {
384 fast_cache.invalidate(id.hash_val(shard_id));
385 }
386 self.short_cache.invalidate(&id.hash_val(shard_id)).await;
387 self.long_cache.invalidate(&id.hash_val(shard_id)).await;
388 }
389
390 pub async fn invalidate_version(&self, id: &dyn HashVal, shard_id: ShardId) {
391 self.version_cache.invalidate(&id.hash_val(shard_id)).await;
392 }
393
394 pub fn invalidate_all_of<T>(&self)
395 where
396 T: CacheVal,
397 {
398 self.short_cache
399 .invalidate_entries_if(|_k, v| v.clone().downcast_arc::<T>().is_ok())
400 .unwrap();
401 self.long_cache
402 .invalidate_entries_if(|_k, v| v.clone().downcast_arc::<T>().is_ok())
403 .unwrap();
404 if let Some(ref storage_cache) = self.storage_cache {
405 storage_cache.invalidate_all_of(T::__type_id());
406 }
407 if let Some(ref fast_cache) = self.fast_cache {
408 fast_cache.invalidate_all_of(T::__type_id());
409 }
410 }
411
412 pub fn invalidate_all_of_version<T>(&self)
413 where
414 T: CacheVal,
415 {
416 self.version_cache
417 .invalidate_entries_if(|_k, v| v.clone().downcast_arc::<T>().is_ok())
418 .unwrap();
419 }
420
421 pub fn invalidate_all(&self) {
422 self.short_cache.invalidate_all();
423 self.version_cache.invalidate_all();
424 self.long_cache.invalidate_all();
425 if let Some(ref storage_cache) = self.storage_cache {
426 storage_cache.invalidate_all();
427 }
428 if let Some(ref fast_cache) = self.fast_cache {
429 fast_cache.invalidate_all();
430 }
431 }
432
433 pub fn fast_cache_hit(&self) -> u64 {
434 self.fast_cache_hit.load(Ordering::Relaxed)
435 }
436 pub fn long_cache_hit(&self) -> u64 {
437 self.long_cache_hit.load(Ordering::Relaxed)
438 }
439 pub fn short_cache_hit(&self) -> u64 {
440 self.short_cache_hit.load(Ordering::Relaxed)
441 }
442 pub fn version_cache_hit(&self) -> u64 {
443 self.version_cache_hit.load(Ordering::Relaxed)
444 }
445 pub fn storage_cache_hit(&self) -> u64 {
446 self.storage_cache_hit.load(Ordering::Relaxed)
447 }
448 pub fn cache_request_count(&self) -> u64 {
449 self.cache_request_count.load(Ordering::Relaxed)
450 }
451 pub fn long_cache_evicted(&self) -> u64 {
452 self.long_cache_evicted.load(Ordering::Relaxed)
453 }
454 pub fn short_cache_evicted(&self) -> u64 {
455 self.short_cache_evicted.load(Ordering::Relaxed)
456 }
457 pub fn version_cache_evicted(&self) -> u64 {
458 self.version_cache_evicted.load(Ordering::Relaxed)
459 }
460}