stretto 0.9.0

Stretto is a high performance thread-safe memory-bound Rust cache.
Documentation 
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

use crate::{
  CacheError, MetricType, Metrics,
  axync::{Receiver, Sender, select, unbounded},
  policy::PolicyInner,
};
use agnostic_lite::RuntimeLite;
use futures::future::FutureExt;
use parking_lot::Mutex;
use std::{collections::hash_map::RandomState, hash::BuildHasher, sync::Arc};

pub(crate) struct AsyncLFUPolicy<S = RandomState> {
  pub(crate) inner: Arc<Mutex<PolicyInner<S>>>,
  pub(crate) items_tx: Sender<Vec<u64>>,
  pub(crate) metrics: Arc<Metrics>,
}

impl AsyncLFUPolicy {
  #[cfg_attr(not(tarpaulin), inline(always))]
  pub(crate) fn new<RT: RuntimeLite>(
    ctrs: usize,
    max_cost: i64,
    stop_rx: Receiver<()>,
  ) -> Result<Self, CacheError> {
    Self::with_hasher::<RT>(ctrs, max_cost, RandomState::new(), stop_rx)
  }
}

impl<S: BuildHasher + Clone + 'static + Send> AsyncLFUPolicy<S> {
  #[cfg_attr(not(tarpaulin), inline(always))]
  pub fn with_hasher<RT: RuntimeLite>(
    ctrs: usize,
    max_cost: i64,
    hasher: S,
    stop_rx: Receiver<()>,
  ) -> Result<Self, CacheError> {
    let inner = PolicyInner::with_hasher(ctrs, max_cost, hasher)?;

    let (items_tx, items_rx) = unbounded();

    PolicyProcessor::new(inner.clone(), items_rx, stop_rx).spawn::<RT>();

    let this = Self {
      inner,
      items_tx,
      metrics: Arc::new(Metrics::new()),
    };

    Ok(this)
  }

  pub fn push(&self, keys: Vec<u64>) -> Option<Vec<u64>> {
    let num_of_keys = keys.len() as u64;
    if num_of_keys == 0 {
      return Some(keys);
    }
    let first = keys[0];
    match self.items_tx.try_send(keys) {
      Ok(_) => {
        self.metrics.add(MetricType::KeepGets, first, num_of_keys);
        None
      }
      Err(err) => {
        self.metrics.add(MetricType::DropGets, first, num_of_keys);
        Some(err.into_inner())
      }
    }
  }
}

pub(crate) struct PolicyProcessor<S> {
  inner: Arc<Mutex<PolicyInner<S>>>,
  items_rx: Receiver<Vec<u64>>,
  stop_rx: Receiver<()>,
}

impl<S: BuildHasher + Clone + 'static + Send> PolicyProcessor<S> {
  #[cfg_attr(not(tarpaulin), inline(always))]
  fn new(
    inner: Arc<Mutex<PolicyInner<S>>>,
    items_rx: Receiver<Vec<u64>>,
    stop_rx: Receiver<()>,
  ) -> Self {
    Self {
      inner,
      items_rx,
      stop_rx,
    }
  }

  #[cfg_attr(not(tarpaulin), inline(always))]
  fn spawn<RT: RuntimeLite>(self) {
    RT::spawn_detach(async move {
      loop {
        select! {
          items = self.items_rx.recv().fuse() => {
            if let Ok(items) = items {
              self.handle_items(items);
            } else {
              // Channel closed, so no more items will be received. Exit the loop.
              return;
            }
          },
          _ = self.stop_rx.recv().fuse() => return,
        }
      }
    });
  }

  #[cfg_attr(not(tarpaulin), inline(always))]
  fn handle_items(&self, items: Vec<u64>) {
    let mut inner = self.inner.lock();
    inner.admit.increments(items);
  }
}

impl_policy!(AsyncLFUPolicy);