armature_cache/
parallel.rs

1//! Parallel batch operations for cache stores.
2
3use crate::error::{CacheError, CacheResult};
4use crate::traits::CacheStore;
5use futures::future::{join_all, try_join_all};
6use serde::{Serialize, de::DeserializeOwned};
7use std::collections::HashMap;
8use std::time::Duration;
9
10/// Parallel batch operations for cache stores.
11///
12/// This module provides high-performance batch operations that execute
13/// multiple cache operations concurrently, significantly reducing total latency.
14///
15/// # Performance
16///
17/// - **get_many**: 10-100x faster than sequential gets (depending on network latency)
18/// - **set_many**: 10-100x faster than sequential sets
19/// - **delete_many**: Similar performance gains
20///
21/// # Examples
22///
23/// ```no_run
24/// use armature_cache::*;
25/// use armature_cache::parallel::*;
26///
27/// # async fn example() -> CacheResult<()> {
28/// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
29///
30/// // Get multiple keys in parallel
31/// let keys = vec!["user:1", "user:2", "user:3"];
32/// let values = get_many_json(&cache, &keys).await?;
33///
34/// // Set multiple keys in parallel
35/// let items = vec![
36///     ("key1", "value1".to_string()),
37///     ("key2", "value2".to_string()),
38/// ];
39/// set_many_json(&cache, &items, None).await?;
40/// # Ok(())
41/// # }
42/// ```
43pub struct ParallelCacheOps;
44
45impl ParallelCacheOps {
46    /// Get multiple JSON values in parallel.
47    ///
48    /// # Arguments
49    ///
50    /// * `store` - The cache store
51    /// * `keys` - Slice of keys to fetch
52    ///
53    /// # Returns
54    ///
55    /// A vector of optional values in the same order as keys.
56    ///
57    /// # Examples
58    ///
59    /// ```no_run
60    /// use armature_cache::*;
61    /// use armature_cache::parallel::ParallelCacheOps;
62    ///
63    /// # async fn example() -> CacheResult<()> {
64    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
65    ///
66    /// let keys = vec!["key1", "key2", "key3"];
67    /// let values = ParallelCacheOps::get_many_json(&cache, &keys).await?;
68    ///
69    /// for (key, value) in keys.iter().zip(values.iter()) {
70    ///     println!("{}: {:?}", key, value);
71    /// }
72    /// # Ok(())
73    /// # }
74    /// ```
75    pub async fn get_many_json<S: CacheStore>(
76        store: &S,
77        keys: &[&str],
78    ) -> CacheResult<Vec<Option<String>>> {
79        // Delegate to the store's native batch primitive. Backends like Redis
80        // collapse this into a single `MGET` round-trip; others fall back to the
81        // concurrent per-key loop. Order matches `keys` in both cases.
82        store.mget(keys).await
83    }
84
85    /// Get multiple typed values in parallel.
86    ///
87    /// # Type Parameters
88    ///
89    /// * `T` - The type to deserialize into
90    ///
91    /// # Examples
92    ///
93    /// ```no_run
94    /// use armature_cache::*;
95    /// use armature_cache::parallel::ParallelCacheOps;
96    /// use serde::{Deserialize, Serialize};
97    ///
98    /// #[derive(Serialize, Deserialize)]
99    /// struct User {
100    ///     id: u64,
101    ///     name: String,
102    /// }
103    ///
104    /// # async fn example() -> CacheResult<()> {
105    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
106    ///
107    /// let keys = vec!["user:1", "user:2", "user:3"];
108    /// let users: Vec<Option<User>> = ParallelCacheOps::get_many(&cache, &keys).await?;
109    /// # Ok(())
110    /// # }
111    /// ```
112    pub async fn get_many<S: CacheStore, T: DeserializeOwned>(
113        store: &S,
114        keys: &[&str],
115    ) -> CacheResult<Vec<Option<T>>> {
116        let json_values = Self::get_many_json(store, keys).await?;
117
118        json_values
119            .into_iter()
120            .map(|opt_json| {
121                opt_json
122                    .map(|json| {
123                        serde_json::from_str(&json)
124                            .map_err(|e| CacheError::Deserialization(e.to_string()))
125                    })
126                    .transpose()
127            })
128            .collect()
129    }
130
131    /// Set multiple JSON values in parallel.
132    ///
133    /// # Arguments
134    ///
135    /// * `store` - The cache store
136    /// * `items` - Slice of (key, value) tuples
137    /// * `ttl` - Optional time-to-live for all items
138    ///
139    /// # Examples
140    ///
141    /// ```no_run
142    /// use armature_cache::*;
143    /// use armature_cache::parallel::ParallelCacheOps;
144    /// use std::time::Duration;
145    ///
146    /// # async fn example() -> CacheResult<()> {
147    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
148    ///
149    /// let items = vec![
150    ///     ("key1", r#"{"value": 1}"#.to_string()),
151    ///     ("key2", r#"{"value": 2}"#.to_string()),
152    /// ];
153    ///
154    /// ParallelCacheOps::set_many_json(&cache, &items, Some(Duration::from_secs(3600))).await?;
155    /// # Ok(())
156    /// # }
157    /// ```
158    pub async fn set_many_json<S: CacheStore>(
159        store: &S,
160        items: &[(&str, String)],
161        ttl: Option<Duration>,
162    ) -> CacheResult<()> {
163        // Delegate to the store's native batch primitive (e.g. Redis `MSET` /
164        // pipelined `SET ... EX`), falling back to the per-key loop otherwise.
165        store.mset(items, ttl).await
166    }
167
168    /// Set multiple typed values in parallel.
169    ///
170    /// # Type Parameters
171    ///
172    /// * `T` - The type to serialize from
173    ///
174    /// # Examples
175    ///
176    /// ```no_run
177    /// use armature_cache::*;
178    /// use armature_cache::parallel::ParallelCacheOps;
179    /// use serde::{Deserialize, Serialize};
180    ///
181    /// #[derive(Serialize, Deserialize)]
182    /// struct Counter {
183    ///     count: u64,
184    /// }
185    ///
186    /// # async fn example() -> CacheResult<()> {
187    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
188    ///
189    /// let items = vec![
190    ///     ("counter:1", Counter { count: 10 }),
191    ///     ("counter:2", Counter { count: 20 }),
192    /// ];
193    ///
194    /// ParallelCacheOps::set_many(&cache, &items, None).await?;
195    /// # Ok(())
196    /// # }
197    /// ```
198    pub async fn set_many<S: CacheStore, T: Serialize>(
199        store: &S,
200        items: &[(&str, T)],
201        ttl: Option<Duration>,
202    ) -> CacheResult<()> {
203        let json_items: Result<Vec<_>, _> = items
204            .iter()
205            .map(|(key, value)| {
206                serde_json::to_string(value)
207                    .map(|json| (*key, json))
208                    .map_err(|e| CacheError::Serialization(e.to_string()))
209            })
210            .collect();
211
212        let json_items = json_items?;
213        let item_refs: Vec<_> = json_items.iter().map(|(k, v)| (*k, v.clone())).collect();
214
215        Self::set_many_json(store, &item_refs, ttl).await
216    }
217
218    /// Delete multiple keys in parallel.
219    ///
220    /// # Examples
221    ///
222    /// ```no_run
223    /// use armature_cache::*;
224    /// use armature_cache::parallel::ParallelCacheOps;
225    ///
226    /// # async fn example() -> CacheResult<()> {
227    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
228    ///
229    /// let keys = vec!["key1", "key2", "key3"];
230    /// ParallelCacheOps::delete_many(&cache, &keys).await?;
231    /// # Ok(())
232    /// # }
233    /// ```
234    pub async fn delete_many<S: CacheStore>(store: &S, keys: &[&str]) -> CacheResult<()> {
235        // Delegate to the store's native batch primitive (e.g. Redis variadic
236        // `DEL`), falling back to the concurrent per-key loop otherwise.
237        store.mdel(keys).await
238    }
239
240    /// Check if multiple keys exist in parallel.
241    ///
242    /// # Returns
243    ///
244    /// A vector of booleans indicating existence, in the same order as keys.
245    ///
246    /// # Examples
247    ///
248    /// ```no_run
249    /// use armature_cache::*;
250    /// use armature_cache::parallel::ParallelCacheOps;
251    ///
252    /// # async fn example() -> CacheResult<()> {
253    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
254    ///
255    /// let keys = vec!["key1", "key2", "key3"];
256    /// let exists = ParallelCacheOps::exists_many(&cache, &keys).await?;
257    ///
258    /// for (key, exists) in keys.iter().zip(exists.iter()) {
259    ///     println!("{}: {}", key, exists);
260    /// }
261    /// # Ok(())
262    /// # }
263    /// ```
264    pub async fn exists_many<S: CacheStore>(store: &S, keys: &[&str]) -> CacheResult<Vec<bool>> {
265        let futures = keys.iter().map(|key| store.exists(key));
266        let results: Vec<CacheResult<bool>> = join_all(futures).await;
267
268        results.into_iter().collect()
269    }
270
271    /// Get TTL for multiple keys in parallel.
272    ///
273    /// # Returns
274    ///
275    /// A vector of optional durations, in the same order as keys.
276    ///
277    /// # Examples
278    ///
279    /// ```no_run
280    /// use armature_cache::*;
281    /// use armature_cache::parallel::ParallelCacheOps;
282    ///
283    /// # async fn example() -> CacheResult<()> {
284    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
285    ///
286    /// let keys = vec!["key1", "key2", "key3"];
287    /// let ttls = ParallelCacheOps::ttl_many(&cache, &keys).await?;
288    ///
289    /// for (key, ttl) in keys.iter().zip(ttls.iter()) {
290    ///     println!("{}: {:?}", key, ttl);
291    /// }
292    /// # Ok(())
293    /// # }
294    /// ```
295    pub async fn ttl_many<S: CacheStore>(
296        store: &S,
297        keys: &[&str],
298    ) -> CacheResult<Vec<Option<Duration>>> {
299        let futures = keys.iter().map(|key| store.ttl(key));
300        let results: Vec<CacheResult<Option<Duration>>> = join_all(futures).await;
301
302        results.into_iter().collect()
303    }
304
305    /// Cache warming: preload multiple keys into cache.
306    ///
307    /// # Type Parameters
308    ///
309    /// * `T` - The type to serialize
310    /// * `F` - Factory function that returns data for a given key
311    ///
312    /// # Examples
313    ///
314    /// ```ignore
315    /// use armature_cache::*;
316    /// use armature_cache::parallel::ParallelCacheOps;
317    /// use std::time::Duration;
318    ///
319    /// # async fn example() -> CacheResult<()> {
320    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
321    ///
322    /// let keys = vec!["user:1", "user:2", "user:3"];
323    ///
324    /// ParallelCacheOps::warm_cache(
325    ///     &cache,
326    ///     &keys,
327    ///     Some(Duration::from_secs(3600)),
328    ///     |key: &str| async move {
329    ///         // Fetch from database
330    ///         let data = format!("Data for {}", key);
331    ///         Ok::<String, CacheError>(data)
332    ///     },
333    /// ).await?;
334    /// # Ok(())
335    /// # }
336    /// ```
337    pub async fn warm_cache<S, T, F, Fut>(
338        store: &S,
339        keys: &[&str],
340        ttl: Option<Duration>,
341        factory: F,
342    ) -> CacheResult<()>
343    where
344        S: CacheStore,
345        T: Serialize,
346        F: Fn(&str) -> Fut,
347        Fut: std::future::Future<Output = CacheResult<T>>,
348    {
349        let mut futures = Vec::new();
350
351        for key in keys {
352            let fut = async {
353                let value = factory(key).await?;
354                let json = serde_json::to_string(&value)
355                    .map_err(|e| CacheError::Serialization(e.to_string()))?;
356                store.set_json(key, json, ttl).await?;
357                Ok::<(), CacheError>(())
358            };
359            futures.push(fut);
360        }
361
362        try_join_all(futures).await?;
363        Ok(())
364    }
365}
366
367/// Helper functions for parallel cache operations.
368///
369/// These functions provide a more convenient API than `ParallelCacheOps` methods.
370/// Get multiple JSON values in parallel.
371pub async fn get_many_json<S: CacheStore>(
372    store: &S,
373    keys: &[&str],
374) -> CacheResult<Vec<Option<String>>> {
375    ParallelCacheOps::get_many_json(store, keys).await
376}
377
378/// Get multiple typed values in parallel.
379pub async fn get_many<S: CacheStore, T: DeserializeOwned>(
380    store: &S,
381    keys: &[&str],
382) -> CacheResult<Vec<Option<T>>> {
383    ParallelCacheOps::get_many(store, keys).await
384}
385
386/// Set multiple JSON values in parallel.
387pub async fn set_many_json<S: CacheStore>(
388    store: &S,
389    items: &[(&str, String)],
390    ttl: Option<Duration>,
391) -> CacheResult<()> {
392    ParallelCacheOps::set_many_json(store, items, ttl).await
393}
394
395/// Set multiple typed values in parallel.
396pub async fn set_many<S: CacheStore, T: Serialize>(
397    store: &S,
398    items: &[(&str, T)],
399    ttl: Option<Duration>,
400) -> CacheResult<()> {
401    ParallelCacheOps::set_many(store, items, ttl).await
402}
403
404/// Delete multiple keys in parallel.
405pub async fn delete_many<S: CacheStore>(store: &S, keys: &[&str]) -> CacheResult<()> {
406    ParallelCacheOps::delete_many(store, keys).await
407}
408
409/// Build a HashMap from multiple keys fetched in parallel.
410pub async fn get_many_as_map<S: CacheStore, T: DeserializeOwned>(
411    store: &S,
412    keys: &[&str],
413) -> CacheResult<HashMap<String, T>> {
414    let values = get_many(store, keys).await?;
415
416    let map: HashMap<String, T> = keys
417        .iter()
418        .zip(values)
419        .filter_map(|(key, opt_value)| opt_value.map(|value| (key.to_string(), value)))
420        .collect();
421
422    Ok(map)
423}
424
425#[cfg(test)]
426mod tests {
427    #[test]
428    fn test_parallel_ops_exist() {
429        // Ensure the module compiles - this test validates the module is correctly defined
430    }
431}
armature_cache/parallel.rs

armature_cache/
parallel.rs
