💥🎨🗑️Simplified module hierarchy

cfpyo3_rs_core/src/io.rs

@@ -1,6 +1,6 @@
 pub mod temporal;
 
 #[cfg(feature = "io-mem-redis")]
-pub use temporal::mem::fetchers::redis::RedisClient;
+pub use temporal::mem::redis::RedisClient;
 #[cfg(feature = "io-mem-redis")]
-pub use temporal::mem::fetchers::redis::REDIS_KEY_NBYTES;
+pub use temporal::mem::redis::REDIS_KEY_NBYTES;

cfpyo3_rs_core/src/io/temporal/mem.rs

@@ -17,23 +17,79 @@
 //! > the 'slice' here indicates a continuous piece of data laid out in memory
 //! 3. create a `Fetcher` for each tiny task
 //!
-//! this module focuses on the first two steps, and the final step is implemented in `./fetchers`.
+//! this module focuses on the first two steps, and the final step is implemented in sub-modules.
 
 use crate::toolkit::array::{AFloat, UnsafeSlice};
-use fetchers::{
-    shm::{SHMFetcher, SlicedSHMFetcher},
-    AsyncFetcher, Fetcher, FetcherArgs,
+use numpy::{
+    ndarray::{s, Array1, ArrayView1, CowArray},
+    Ix1,
 };
-use numpy::ndarray::{s, Array1, ArrayView1};
-use std::{collections::HashMap, iter::zip};
-
 #[cfg(feature = "io-mem-redis")]
-use fetchers::redis::{RedisClient, RedisFetcher, RedisGroupedFetcher, RedisKey};
+use redis::{RedisClient, RedisFetcher, RedisGroupedFetcher, RedisKey};
+use shm::{SHMFetcher, SlicedSHMFetcher};
+use std::{collections::HashMap, future::Future, iter::zip};
 
-pub mod fetchers;
+#[cfg(feature = "io-mem-redis")]
+pub mod redis;
+pub mod shm;
 
 // core implementations
 
+/// arguments for the `Fetcher::fetch` method
+///
+/// # concepts
+///
+/// - `compact data`: we assume that the data of each date is flattened
+/// and then concatenated into a single array, and this array is called `compact data`
+/// - `date data`: the flattened data of the specific date
+///
+/// # fields
+///
+/// - `c`: the index of the current fetcher among all fetchers in a batch get mode
+/// > if not in batch get mode, this field should be `None`
+/// - `start_idx`: the start index of the `compact data` to fetch
+/// - `end_idx`: the end index of the `compact data` to fetch
+/// - `date_idx`: the index of the date among all dates
+/// - `date_col_idx`: the index of the column in the `date data` to fetch
+/// - `date_start_idx`: the index where the `date data` starts in the `compact data`
+/// - `time_start_idx`: the start index of the `date data` to fetch
+/// - `time_end_idx`: the end index of the `date data` to fetch
+/// - `num_ticks_per_day`: the number of ticks per day
+/// - `data_len`: the length of the data to fetch, i.e. `end_idx - start_idx` or `time_end_idx - time_start_idx`
+#[derive(Debug)]
+pub struct FetcherArgs {
+    pub c: Option<usize>,
+    pub start_idx: i64,
+    pub end_idx: i64,
+    pub date_idx: i64,
+    pub date_col_idx: i64,
+    pub date_start_idx: i64,
+    pub time_start_idx: i64,
+    pub time_end_idx: i64,
+    pub num_ticks_per_day: i64,
+    pub data_len: i64,
+}
+
+pub trait Fetcher<T: AFloat> {
+    fn can_batch_fetch(&self) -> bool {
+        false
+    }
+    fn fetch(&self, args: FetcherArgs) -> CowArray<T, Ix1> {
+        if self.can_batch_fetch() {
+            self.batch_fetch(vec![args]).pop().unwrap()
+        } else {
+            unreachable!("should implement `fetch` when `can_batch_fetch` returns false");
+        }
+    }
+    fn batch_fetch(&self, _args: Vec<FetcherArgs>) -> Vec<CowArray<T, Ix1>> {
+        unreachable!("`batch_fetch` should be implemented when `can_batch_fetch` returns true");
+    }
+}
+
+pub trait AsyncFetcher<T: AFloat> {
+    fn fetch(&self, args: FetcherArgs) -> impl Future<Output = CowArray<T, Ix1>>;
+}
+
 fn searchsorted(arr: &ArrayView1<i64>, value: i64) -> i64 {
     arr.as_slice()
         .unwrap()