starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-10 08:07:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(cache): refurbish fs backed cache (#4863)

Browse Source 
* feat(cache): refurbish fs backed cache

* add `cached` exemple
This commit is contained in:
Lucas Abel
2024-12-06 12:26:36 +01:00
committed by GitHub
parent c0b93f9c97
commit b66a31a623
3 changed files with 213 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/cache.rs
+201 -64
View File
@@ -1,25 +1,149 @@
use crate::error;
use std::future::Future;
use std::hash::Hash;
use std::path::{Path, PathBuf};
use quick_cache::sync::Cache;
use quick_cache::Equivalent;
use serde::{Deserialize, Serialize};
use sqlx::PgExecutor;
pub use const_format::concatcp;
pub use lazy_static::lazy_static;
pub use quick_cache::sync::Cache;
/// Cache directory for windmill server/worker(s).
pub const CACHE_DIR: &str = "/tmp/windmill/cache/";
/// A file-system backed concurrent cache.
pub struct FsBackedCache<Key, Val> {
cache: Cache<Key, Val>,
root: &'static str,
}
impl<Key: Eq + Hash, Val: Clone + fs::Bundle> FsBackedCache<Key, Val> {
/// Create a new file-system backed cache with `items_capacity` capacity.
/// The cache will be stored in the `root` directory.
pub fn new(root: &'static str, items_capacity: usize) -> Self {
Self { cache: Cache::new(items_capacity), root }
}
/// Gets or inserts an item in the cache with key `key`.
pub async fn get_or_insert_async<'a, Q, F>(&'a self, key: &Q, with: F) -> error::Result<Val>
where
Q: Hash + Equivalent<Key> + ToOwned<Owned = Key> + Copy + Into<u64>,
F: Future<Output = error::Result<Val>>,
{
self.cache
.get_or_insert_async(
key,
fs::import_or_insert_with(self.root, (*key).into(), with),
)
.await
}
}
/// Like [`lazy_static`]`, but for file-system backed caches.
///
/// # Example
/// ```rust
/// use windmill_common::make_static;
///
/// make_static! {
/// /// String cache with a maximum capacity of 1000 items stored in the
/// /// "subdirectory" directory.
/// static ref CACHE: { u64 => String } in "subdirectory" <= 1000;
/// /// Another cache.
/// static ref ANOTHER_CACHE: { u64 => Vec<String> } in "another" <= 100;
/// }
/// ```
#[macro_export]
macro_rules! make_static {
{ $( $(#[$attr:meta])* static ref $name:ident: { $Key:ty => $Val:ty } in $root:literal <= $cap:literal; )+ } => {
$crate::cache::lazy_static! {
$(
$(#[$attr])*
static ref $name: $crate::cache::FsBackedCache<$Key, $Val> =
$crate::cache::FsBackedCache::new(
$crate::cache::concatcp!($crate::cache::CACHE_DIR, $root),
$cap
);
)+
}
};
}
// re-export:
pub use make_static;
/// Create an anonymous file-system backed cache for one-time use.
///
/// # Example
/// ```rust
/// use windmill_common::anon;
/// let cache = anon!({ u64 => String } in "subdirectory" <= 1000);
/// ```
#[macro_export]
macro_rules! anon {
({ $Key:ty => $Val:ty } in $root:literal <= $cap:literal) => {{
$crate::cache::make_static! {
static ref __ANON__: { $Key => $Val } in $root <= $cap;
}
&__ANON__
}};
}
// re-export:
pub use anon;
pub mod future {
use super::*;
/// Extension trait for futures that can be cached.
pub trait FutureCachedExt<Val: Clone + fs::Bundle>:
Future<Output = error::Result<Val>> + Sized
{
/// Get or insert the future result in the cache.
///
/// # Example
/// ```rust
/// use windmill_common::cache::{self, future::FutureCachedExt};
///
/// async {
/// let result = std::future::ready(Ok(42))
/// .cached(cache::anon!({ u64 => u64 } in "test" <= 1), &42)
/// .await;
///
/// assert_eq!(result.unwrap(), 42);
/// };
/// ```
fn cached<Key: Eq + Hash, Q>(
self,
cache: &FsBackedCache<Key, Val>,
key: &Q,
) -> impl Future<Output = error::Result<Val>>
where
Q: Hash + Equivalent<Key> + ToOwned<Owned = Key> + Copy + Into<u64>,
{
cache.get_or_insert_async(key, self)
}
}
impl<Val: Clone + fs::Bundle, F: Future<Output = error::Result<Val>> + Sized>
FutureCachedExt<Val> for F
{
}
}
pub mod flow {
use super::*;
use crate::flows::{FlowNodeId, FlowValue};
/// Cache directory for windmill server/worker(s) flow nodes.
pub const CACHE_DIR: &str = const_format::concatcp!(super::CACHE_DIR, "flow");
lazy_static::lazy_static! {
make_static! {
/// Flow node cache.
/// FIXME: This should be a static but [`Cache`] does not have a const constructor.
/// FIXME: Use `Arc<Val>` for cheap cloning.
static ref CACHE: Cache<FlowNodeId, Val> = Cache::new(1000);
static ref CACHE: { FlowNodeId => Val } in "flow" <= 1000;
}
/// Flow node cache value.
@@ -75,38 +199,35 @@ pub mod flow {
// so only one thread will be able to fetch the data from the database and write it to
// the file system and cache, hence no race on the file system.
CACHE
.get_or_insert_async(
&node,
fs::import_or_insert_with(CACHE_DIR, node.0 as u64, async {
sqlx::query!(
"SELECT \
.get_or_insert_async(&node, async {
sqlx::query!(
"SELECT \
lock AS \"lock: String\", \
code AS \"code: String\", \
flow::text AS \"flow: Box<str>\" \
FROM flow_node WHERE id = $1 LIMIT 1",
node.0,
)
.fetch_one(e)
.await
.map_err(Into::into)
.and_then(|r| {
Ok(Val {
lock: r
.lock
.and_then(|x| if x.is_empty() { None } else { Some(x) }),
code: r.code,
flow: match r.flow {
None => None,
Some(flow) => serde_json::from_str(&flow).map_err(|err| {
error::Error::InternalErr(format!(
"Unable to parse flow value: {err:?}"
))
})?,
},
})
node.0,
)
.fetch_one(e)
.await
.map_err(Into::into)
.and_then(|r| {
Ok(Val {
lock: r
.lock
.and_then(|x| if x.is_empty() { None } else { Some(x) }),
code: r.code,
flow: match r.flow {
None => None,
Some(flow) => serde_json::from_str(&flow).map_err(|err| {
error::Error::InternalErr(format!(
"Unable to parse flow value: {err:?}"
))
})?,
},
})
}),
)
})
})
.await
}
@@ -164,14 +285,10 @@ pub mod script {
use super::*;
use crate::scripts::{ScriptHash, ScriptLang};
/// Cache directory for windmill server/worker(s) scripts.
pub const CACHE_DIR: &str = const_format::concatcp!(super::CACHE_DIR, "script");
lazy_static::lazy_static! {
make_static! {
/// Scripts cache.
/// FIXME: This should be a static but [`Cache`] does not have a const constructor.
/// FIXME: Use `Arc<Val>` for cheap cloning.
static ref CACHE: Cache<ScriptHash, Val> = Cache::new(1000);
static ref CACHE: { ScriptHash => Val } in "script" <= 1000;
}
/// Script cache value.
@@ -197,34 +314,31 @@ pub mod script {
// so only one thread will be able to fetch the data from the database and write it to
// the file system and cache, hence no race on the file system.
CACHE
.get_or_insert_async(
&hash,
fs::import_or_insert_with(CACHE_DIR, hash.0 as u64, async {
sqlx::query!(
"SELECT \
.get_or_insert_async(&hash, async {
sqlx::query!(
"SELECT \
lock AS \"lock: String\", \
content AS \"code!: String\",
language AS \"language: Option<ScriptLang>\", \
envs AS \"envs: Vec<String>\", \
codebase AS \"codebase: String\" \
FROM script WHERE hash = $1 AND workspace_id = $2 LIMIT 1",
hash.0,
workspace_id,
)
.fetch_one(e)
.await
.map_err(Into::into)
.map(|r| Val {
lock: r
.lock
.and_then(|x| if x.is_empty() { None } else { Some(x) }),
code: r.code,
language: r.language,
envs: r.envs,
codebase: r.codebase,
})
}),
)
hash.0,
workspace_id,
)
.fetch_one(e)
.await
.map_err(Into::into)
.map(|r| Val {
lock: r
.lock
.and_then(|x| if x.is_empty() { None } else { Some(x) }),
code: r.code,
language: r.language,
envs: r.envs,
codebase: r.codebase,
})
})
.await
}
@@ -283,8 +397,6 @@ pub mod script {
mod fs {
use super::*;
use std::future::Future;
use std::fs::{self, OpenOptions};
use std::io::{Read, Write};
@@ -363,4 +475,29 @@ mod fs {
}
Ok(data)
}
// Auto-implement `Bundle` for all `serde` serializable types.
impl Item for () {
fn path(&self, root: &Path) -> PathBuf {
root.join("self.json")
}
}
impl<T: for<'de> Deserialize<'de> + Serialize + Default> Bundle for T {
type Item = ();
fn items() -> &'static [Self::Item] {
&[()]
}
fn import(&mut self, _: Self::Item, data: Vec<u8>) -> error::Result<()> {
*self = serde_json::from_slice(&data)?;
Ok(())
}
fn export(&self, _: Self::Item) -> error::Result<Option<Vec<u8>>> {
Ok(Some(serde_json::to_vec(self)?))
}
}
}
backend/windmill-common/src/flows.rs
+6
View File
@@ -407,6 +407,12 @@ pub enum InputTransform {
#[serde(transparent)]
pub struct FlowNodeId(pub i64);
impl Into<u64> for FlowNodeId {
fn into(self) -> u64 {
self.0 as u64
}
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Branch {
#[serde(skip_serializing_if = "Option::is_none")]
backend/windmill-common/src/scripts.rs
+6
View File
@@ -75,6 +75,12 @@ impl ScriptLang {
#[sqlx(transparent)]
pub struct ScriptHash(pub i64);
impl Into<u64> for ScriptHash {
fn into(self) -> u64 {
self.0 as u64
}
}
#[derive(PartialEq, sqlx::Type)]
#[sqlx(transparent, no_pg_array)]
pub struct ScriptHashes(pub Vec<i64>);