From 9ef63e8945794199d7b1a07a41efe9f4d3e1bd61 Mon Sep 17 00:00:00 2001 From: Andrey Ermilov Date: Sat, 29 Jul 2023 17:39:20 +0200 Subject: [PATCH] Stretto in-memory backend --- examples/examples/axum.rs | 10 ++- examples/examples/tower.rs | 6 +- hitbox-backend/Cargo.toml | 1 + hitbox-backend/src/serializer.rs | 44 ++++++++++++- hitbox-stretto/Cargo.toml | 3 + hitbox-stretto/src/backend.rs | 103 +++++++++++++++++++++---------- hitbox-stretto/src/builder.rs | 12 ++-- hitbox-stretto/src/lib.rs | 5 +- 8 files changed, 132 insertions(+), 52 deletions(-) diff --git a/examples/examples/axum.rs b/examples/examples/axum.rs index 9727b3f..3542c2d 100644 --- a/examples/examples/axum.rs +++ b/examples/examples/axum.rs @@ -1,6 +1,7 @@ use axum::{extract::Path, routing::get, Json, Router}; use hitbox_redis::RedisBackend; +use hitbox_stretto::StrettoBackend; use hitbox_tower::Cache; use http::StatusCode; use tower::ServiceBuilder; @@ -38,20 +39,17 @@ async fn main() { tracing::subscriber::set_global_default(subscriber).unwrap(); let backend = RedisBackend::new().unwrap(); - let inmemory = hitbox_stretto::StrettoBackendBuilder::new(12960, 1e6 as i64) + let inmemory = StrettoBackend::builder(2 ^ 16) .finalize() .unwrap(); // build our application with a single route let app = Router::new() .route("/greet/:name/", get(handler_result)) .route("/", get(handler)) - .route( - "/json/", - get(handler_json), // .layer(Cache::builder().backend(inmemory.clone()).build()), - ) + .route("/json/", get(handler_json)) .layer( ServiceBuilder::new() - // .layer(Cache::builder().backend(inmemory).build()) + .layer(Cache::builder().backend(inmemory).build()) .layer(Cache::builder().backend(backend).build()), ); diff --git a/examples/examples/tower.rs b/examples/examples/tower.rs index de434e3..e789393 100644 --- a/examples/examples/tower.rs +++ b/examples/examples/tower.rs @@ -1,4 +1,4 @@ -use hitbox_stretto::builder::StrettoBackendBuilder; +use hitbox_stretto::StrettoBackend; use hitbox_tower::Cache; use hyper::{Body, Server}; use std::{convert::Infallible, net::SocketAddr}; @@ -18,9 +18,7 @@ async fn main() { .finish(); tracing::subscriber::set_global_default(subscriber).unwrap(); - let inmemory = StrettoBackendBuilder::new(12960, 1e6 as i64) - .finalize() - .unwrap(); + let inmemory = StrettoBackend::builder(2 ^ 16).finalize().unwrap(); let service = tower::ServiceBuilder::new() .layer(tower_http::trace::TraceLayer::new_for_http()) .layer(Cache::builder().backend(inmemory).build()) diff --git a/hitbox-backend/Cargo.toml b/hitbox-backend/Cargo.toml index e8f348b..eb4814d 100644 --- a/hitbox-backend/Cargo.toml +++ b/hitbox-backend/Cargo.toml @@ -16,4 +16,5 @@ futures = { version = "0.3", default-features = false } chrono = { version = "0.4", features = ["serde"] } serde = { version = "1", features = ["derive"] } serde_json = "1" +bincode = "1" thiserror = "1" diff --git a/hitbox-backend/src/serializer.rs b/hitbox-backend/src/serializer.rs index 0dafb8c..97df3f7 100644 --- a/hitbox-backend/src/serializer.rs +++ b/hitbox-backend/src/serializer.rs @@ -108,6 +108,40 @@ impl Serializer for JsonSerializer { } } +#[derive(Default)] +pub struct BinSerializer> { + _raw: PhantomData, +} + +impl Serializer for BinSerializer> { + type Raw = Vec; + + fn deserialize(data: Self::Raw) -> Result, SerializerError> + where + T: DeserializeOwned, + { + let deserialized: SerializableCachedValue = bincode::deserialize(&data) + .map_err(|err| SerializerError::Deserialize(Box::new(err)))?; + let cached_value = deserialized.into_cached_value(); + Ok(CachedValue::new( + cached_value.data.into(), + cached_value.expired, + )) + } + + fn serialize(value: &CachedValue) -> Result + where + T: Serialize, + { + let serializable_value: SerializableCachedValue<&T> = SerializableCachedValue { + data: &value.data, + expired: value.expired, + }; + Ok(bincode::serialize(&serializable_value) + .map_err(|err| SerializerError::Serialize(Box::new(err)))?) + } +} + #[cfg(test)] mod test { use std::convert::Infallible; @@ -155,7 +189,15 @@ mod test { fn test_json_string_serializer() { let value = CachedValue::new(Test::new(), Utc::now()); let raw = JsonSerializer::::serialize(&value).unwrap(); - dbg!(&raw); + assert_eq!(raw.len(), 71); assert_eq!(value, JsonSerializer::::deserialize(raw).unwrap()); } + + #[test] + fn test_bincode_serializer() { + let value = CachedValue::new(Test::new(), Utc::now()); + let raw = ::serialize(&value).unwrap(); + assert_eq!(raw.len(), 54); + assert_eq!(value, BinSerializer::>::deserialize(raw).unwrap()); + } } diff --git a/hitbox-stretto/Cargo.toml b/hitbox-stretto/Cargo.toml index 9f7b782..c8fab52 100644 --- a/hitbox-stretto/Cargo.toml +++ b/hitbox-stretto/Cargo.toml @@ -30,3 +30,6 @@ tower = { version = "0.4", features = ["full"] } tower-http = { version = "0.4", features = ["full"] } http = "0.2" lazy_static = "1" + +[dev-dependencies] +chrono = { version = "0.4", features = ["serde"] } diff --git a/hitbox-stretto/src/backend.rs b/hitbox-stretto/src/backend.rs index 356b01a..4ba1fc0 100644 --- a/hitbox-stretto/src/backend.rs +++ b/hitbox-stretto/src/backend.rs @@ -1,21 +1,20 @@ +use crate::builder::StrettoBackendBuilder; use axum::async_trait; use hitbox_backend::{ - serializer::{JsonSerializer, Serializer}, + serializer::{BinSerializer, Serializer}, BackendError, BackendResult, CacheBackend, CacheableResponse, CachedValue, DeleteStatus, }; use std::time::Duration; use stretto::AsyncCache; -const COST: i64 = 0; - #[derive(Clone)] pub struct StrettoBackend { - cache: AsyncCache>, + pub(crate) cache: AsyncCache>, } impl StrettoBackend { - pub fn new(cache: AsyncCache>) -> Self { - Self { cache } + pub fn builder(max_size: i64) -> StrettoBackendBuilder { + StrettoBackendBuilder::new(max_size) } } @@ -26,12 +25,18 @@ impl CacheBackend for StrettoBackend { T: CacheableResponse, ::Cached: serde::de::DeserializeOwned, { + let () = self + .cache + .wait() + .await + .map_err(crate::error::Error::from) + .map_err(BackendError::from)?; + match self.cache.get(&key).await { - Some(cached) => Ok(Some( - JsonSerializer::>::deserialize(cached.value().to_owned()) - .map_err(BackendError::from) - .unwrap(), - )), + Some(cached) => BinSerializer::>::deserialize(cached.value().to_owned()) + .map_err(BackendError::from) + .map(Some), + None => Ok(None), } } @@ -46,15 +51,20 @@ impl CacheBackend for StrettoBackend { T: CacheableResponse + Send, T::Cached: serde::Serialize + Send + Sync, { - let serialized = - JsonSerializer::>::serialize(&value).map_err(BackendError::from)?; + let serialized = BinSerializer::>::serialize(&value).map_err(BackendError::from)?; + let cost = serialized.len(); let inserted = match ttl { Some(ttl) => { self.cache - .insert_with_ttl(key, serialized, COST, Duration::from_secs(ttl as u64)) + .insert_with_ttl( + key, + serialized, + cost as i64, + Duration::from_secs(ttl as u64), + ) .await } - None => self.cache.insert(key, serialized, COST).await, + None => self.cache.insert(key, serialized, cost as i64).await, }; if inserted { Ok(()) @@ -65,7 +75,7 @@ impl CacheBackend for StrettoBackend { async fn delete(&self, key: String) -> BackendResult { self.cache.remove(&key).await; - Ok(DeleteStatus::Deleted(0)) + Ok(DeleteStatus::Deleted(1)) } async fn start(&self) -> BackendResult<()> { @@ -73,24 +83,55 @@ impl CacheBackend for StrettoBackend { } } -#[tokio::test] -async fn test() { - let c: AsyncCache = AsyncCache::new(1000, 100, tokio::spawn).unwrap(); +#[cfg(test)] +mod test { + use axum::async_trait; + use chrono::Utc; + use serde::{Deserialize, Serialize}; + + use super::*; + use hitbox_backend::CacheableResponse; - for i in 0..100 { - let key = format!("key-{}", i); - let r = c.insert(key, "value".to_string(), 1).await; - dbg!(r); + #[derive(Serialize, Deserialize, Clone, PartialEq, Debug)] + struct Test { + a: i32, + b: String, } - c.wait().await.unwrap(); + #[async_trait] + impl CacheableResponse for Test { + type Cached = Self; - for i in 0..100 { - let key = format!("key-{}", i); - let value = c.get(&key).await; - match value { - Some(v) => dbg!(v.to_string()), - None => dbg!("None".to_string()), - }; + async fn into_cached(self) -> Self::Cached { + self + } + async fn from_cached(cached: Self::Cached) -> Self { + cached + } + } + + impl Test { + pub fn new() -> Self { + Self { + a: 42, + b: "nope".to_owned(), + } + } + } + + #[tokio::test] + async fn test_set_and_get() { + let cache = crate::StrettoBackend::builder(100).finalize().unwrap(); + let value = CachedValue::new(Test::new(), Utc::now()); + let res = cache.set::("key-1".to_string(), &value, None).await; + assert!(res.is_ok()); + let value = cache + .get::("key-1".to_string()) + .await + .unwrap() + .unwrap() + .into_inner(); + assert_eq!(value.a, 42); + assert_eq!(value.b, "nope".to_owned()); } } diff --git a/hitbox-stretto/src/builder.rs b/hitbox-stretto/src/builder.rs index 4357353..6909621 100644 --- a/hitbox-stretto/src/builder.rs +++ b/hitbox-stretto/src/builder.rs @@ -7,8 +7,9 @@ type Cache = AsyncCacheBuilder>; pub struct StrettoBackendBuilder(Cache); impl StrettoBackendBuilder { - pub fn new(num_counters: usize, max_cost: i64) -> Self { - Self(AsyncCacheBuilder::new(num_counters, max_cost)) + pub fn new(max_size: i64) -> Self { + let num_counters = max_size * 10; + Self(AsyncCacheBuilder::new(num_counters as usize, max_size)) } pub fn set_buffer_size(self, sz: usize) -> Self { @@ -19,18 +20,15 @@ impl StrettoBackendBuilder { Self(self.0.set_buffer_items(sz)) } - pub fn set_ingore_internal_cost(self, val: bool) -> Self { - Self(self.0.set_ignore_internal_cost(val)) - } - pub fn set_cleanup_duration(self, d: Duration) -> Self { Self(self.0.set_cleanup_duration(d)) } pub fn finalize(self) -> Result { self.0 + .set_ignore_internal_cost(true) .finalize(tokio::spawn) - .map(crate::backend::StrettoBackend::new) + .map(|cache| crate::backend::StrettoBackend { cache }) .map_err(Error::from) } } diff --git a/hitbox-stretto/src/lib.rs b/hitbox-stretto/src/lib.rs index 92da684..d179e98 100644 --- a/hitbox-stretto/src/lib.rs +++ b/hitbox-stretto/src/lib.rs @@ -1,6 +1,5 @@ -pub mod backend; -pub mod builder; +mod backend; +mod builder; pub mod error; pub use crate::backend::StrettoBackend; -pub use crate::builder::StrettoBackendBuilder;