Added basic Rust request tracker implementation (#9743)

* Added basic single threaded request tracker implementation without invalidations

* slightly cleaner

Author: alshdavid

Cargo.lock

@@ -19,3 +19,5 @@ serde_repr = "0.1.19"
 serde-value = "0.7.0"
 xxhash-rust = { version = "0.8.2", features = ["xxh3"] }
 anyhow = "1.0.82"
+dyn-hash = "0.x"
+petgraph = "0.x"

crates/parcel_core/Cargo.toml

@@ -3,6 +3,7 @@
 
 pub mod hash;
 pub mod plugin;
+pub mod request_tracker;
 pub mod types;
 
 // pub use parcel::*;

crates/parcel_core/src/lib.rs

@@ -0,0 +1,139 @@
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+use std::sync::Mutex;
+
+use super::*;
+
+#[test]
+fn should_run_request() {
+  let rt = RequestTracker::<Vec<String>>::new();
+
+  let request_c = TestRequest::new("C", &[]);
+  let request_b = TestRequest::new("B", &[&request_c]);
+  let request_a = TestRequest::new("A", &[&request_b]);
+
+  let result = rt.run_request(Box::new(&request_a)).unwrap();
+
+  assert_eq!(result[0], "A");
+  assert_eq!(result[1], "B");
+  assert_eq!(result[2], "C");
+}
+
+#[test]
+fn should_reuse_previously_run_request() {
+  let rt = RequestTracker::<Vec<String>>::new();
+
+  let request_c = TestRequest::new("C", &[]);
+  let request_b = TestRequest::new("B", &[&request_c]);
+  let request_a = TestRequest::new("A", &[&request_b]);
+
+  let result = rt.run_request(Box::new(&request_a)).unwrap();
+
+  assert_eq!(result[0], "A");
+  assert_eq!(result[1], "B");
+  assert_eq!(result[2], "C");
+
+  let result = rt.run_request(Box::new(&request_a)).unwrap();
+  assert_eq!(result[0], "A");
+  assert_eq!(result[1], "B");
+  assert_eq!(result[2], "C");
+}
+
+#[test]
+fn should_run_request_once() {
+  let rt = RequestTracker::<Vec<String>>::new();
+
+  let request_a = TestRequest::new("A", &[]);
+
+  let result = rt.run_request(Box::new(&request_a)).unwrap();
+
+  assert_eq!(result[0], "A");
+  assert_eq!(request_a.run_count(), 1);
+
+  let result = rt.run_request(Box::new(&request_a)).unwrap();
+  assert_eq!(result[0], "A");
+  assert_eq!(request_a.run_count(), 1);
+}
+
+#[test]
+fn should_run_request_once_2() {
+  let rt = RequestTracker::<Vec<String>>::new();
+
+  let request_b = TestRequest::new("B", &[]);
+  let request_a = TestRequest::new("A", &[&request_b]);
+
+  let result = rt.run_request(Box::new(&request_a)).unwrap();
+
+  assert_eq!(result[0], "A");
+  assert_eq!(result[1], "B");
+  assert_eq!(request_a.run_count(), 1);
+  assert_eq!(request_b.run_count(), 1);
+
+  let result = rt.run_request(Box::new(&request_a)).unwrap();
+  assert_eq!(result[0], "A");
+  assert_eq!(result[1], "B");
+  assert_eq!(request_a.run_count(), 1);
+  assert_eq!(request_b.run_count(), 1);
+}
+
+/// This is a universal "Request" that can be instructed
+/// to run subrequests via the constructor
+#[derive(Clone, Default)]
+pub struct TestRequest<'a> {
+  pub runs: Arc<AtomicUsize>,
+  pub name: String,
+  pub subrequets: Arc<Mutex<Vec<&'a TestRequest<'a>>>>,
+}
+
+impl<'a> std::fmt::Debug for TestRequest<'a> {
+  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    f.debug_struct(&format!("TestRequest({})", self.name))
+      .finish()
+  }
+}
+
+impl<'a> TestRequest<'a> {
+  pub fn new<T: AsRef<str>>(name: T, subrequests: &[&'a TestRequest<'a>]) -> Self {
+    Self {
+      runs: Default::default(),
+      name: name.as_ref().to_string(),
+      subrequets: Arc::new(Mutex::new(subrequests.to_owned())),
+    }
+  }
+
+  pub fn run_count(&self) -> usize {
+    self.runs.load(Ordering::Relaxed)
+  }
+}
+
+impl<'a> std::hash::Hash for TestRequest<'a> {
+  fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+    self.name.hash(state);
+  }
+}
+
+impl<'a> Request<Vec<String>> for TestRequest<'a> {
+  fn run(
+    &self,
+    rt: RequestTracker<Vec<String>>,
+  ) -> Result<RequestResult<Vec<String>>, Vec<RequestError>> {
+    self.runs.fetch_add(1, Ordering::Relaxed);
+
+    let name = self.name.clone();
+    let mut subrequets = self.subrequets.lock().unwrap().clone();
+
+    let mut result = vec![name];
+
+    while let Some(subrequest) = subrequets.pop() {
+      let req = subrequest.clone();
+      let subrequest_result = rt.run_request(Box::new(&req))?;
+      result.extend(subrequest_result);
+    }
+
+    Ok(RequestResult {
+      result,
+      invalidations: vec![],
+    })
+  }
+}

crates/parcel_core/src/request_tracker/_test.rs

@@ -0,0 +1,10 @@
+mod request;
+mod request_graph;
+mod request_tracker;
+
+#[cfg(test)]
+mod _test;
+
+pub use self::request::*;
+pub use self::request_graph::*;
+pub use self::request_tracker::*;

crates/parcel_core/src/request_tracker/mod.rs

@@ -0,0 +1,34 @@
+use std::fmt::Debug;
+use std::hash::DefaultHasher;
+use std::hash::Hash;
+use std::hash::Hasher;
+
+use dyn_hash::DynHash;
+
+use super::RequestTracker;
+
+pub trait Request<T: Clone>: DynHash {
+  fn id(&self) -> u64 {
+    let mut hasher = DefaultHasher::default();
+    std::any::type_name::<Self>().hash(&mut hasher);
+    self.dyn_hash(&mut hasher);
+    hasher.finish()
+  }
+
+  fn run(&self, request_tracker: RequestTracker<T>) -> Result<RequestResult<T>, Vec<RequestError>>;
+}
+
+dyn_hash::hash_trait_object!(<T: Clone> Request<T>);
+
+pub struct RequestResult<Req> {
+  pub result: Req,
+  pub invalidations: Vec<Invalidation>,
+}
+
+#[derive(Debug, Clone)]
+pub enum RequestError {
+  Impossible,
+}
+
+#[derive(Debug)]
+pub enum Invalidation {}

crates/parcel_core/src/request_tracker/request.rs

@@ -0,0 +1,18 @@
+use petgraph::stable_graph::StableDiGraph;
+
+use super::RequestError;
+
+pub type RequestGraph<T> = StableDiGraph<RequestNode<T>, RequestEdgeType>;
+
+#[derive(Debug)]
+pub enum RequestNode<T> {
+  Error(Vec<RequestError>),
+  Root,
+  Incomplete,
+  Valid(T),
+}
+
+#[derive(Debug)]
+pub enum RequestEdgeType {
+  SubRequest,
+}

crates/parcel_core/src/request_tracker/request_graph.rs

@@ -0,0 +1,114 @@
+use std::cell::RefCell;
+use std::collections::HashMap;
+use std::rc::Rc;
+
+use petgraph::graph::NodeIndex;
+use petgraph::stable_graph::StableDiGraph;
+
+use super::Request;
+use super::RequestEdgeType;
+use super::RequestError;
+use super::RequestGraph;
+use super::RequestNode;
+use super::RequestResult;
+
+#[derive(Clone)]
+pub struct RequestTracker<T: Clone> {
+  parent_request_hash: Option<u64>,
+  graph: Rc<RefCell<RequestGraph<T>>>,
+  request_index: Rc<RefCell<HashMap<u64, NodeIndex>>>,
+}
+
+impl<T: Clone> RequestTracker<T> {
+  pub fn new() -> Self {
+    let mut graph = StableDiGraph::<RequestNode<T>, RequestEdgeType>::new();
+    graph.add_node(RequestNode::Root);
+    RequestTracker {
+      parent_request_hash: None,
+      graph: Rc::new(RefCell::new(graph)),
+      request_index: Rc::new(RefCell::new(HashMap::new())),
+    }
+  }
+
+  pub fn run_request(&self, request: Box<&dyn Request<T>>) -> Result<T, Vec<RequestError>> {
+    let request_id = request.id();
+
+    if self.prepare_request(request_id.clone()) {
+      let mut rt = self.clone();
+      rt.parent_request_hash.replace(request_id.clone());
+      self.store_request(&request_id, request.run(rt));
+    }
+
+    self.get_request(&request_id)
+  }
+
+  fn prepare_request(&self, request_id: u64) -> bool {
+    let mut graph = self.graph.borrow_mut();
+    let mut request_index = self.request_index.borrow_mut();
+
+    let node_index = request_index
+      .entry(request_id)
+      .or_insert_with(|| graph.add_node(RequestNode::Incomplete));
+
+    let request_node = graph.node_weight_mut(*node_index).unwrap();
+
+    // Don't run if already run
+    if let RequestNode::<T>::Valid(_) = request_node {
+      return false;
+    }
+
+    *request_node = RequestNode::Incomplete;
+    true
+  }
+
+  fn store_request(&self, request_id: &u64, result: Result<RequestResult<T>, Vec<RequestError>>) {
+    let request_index = self.request_index.borrow();
+    let mut graph = self.graph.borrow_mut();
+
+    let node_index = request_index.get(&request_id).unwrap();
+
+    let request_node = graph.node_weight_mut(*node_index).unwrap();
+    if let RequestNode::<T>::Valid(_) = request_node {
+      return;
+    }
+    *request_node = match result {
+      Ok(result) => RequestNode::Valid(result.result),
+      Err(error) => RequestNode::Error(error),
+    };
+  }
+
+  fn get_request(&self, request_id: &u64) -> Result<T, Vec<RequestError>> {
+    let mut graph = self.graph.borrow_mut();
+    let request_index = self.request_index.borrow();
+
+    let Some(node_index) = request_index.get(&request_id) else {
+      return Err(vec![RequestError::Impossible]);
+    };
+
+    if let Some(parent_request_id) = self.parent_request_hash {
+      let parent_node_index = request_index.get(&parent_request_id).unwrap();
+      graph.add_edge(
+        parent_node_index.clone(),
+        node_index.clone(),
+        RequestEdgeType::SubRequest,
+      );
+    } else {
+      graph.add_edge(
+        NodeIndex::new(0),
+        node_index.clone(),
+        RequestEdgeType::SubRequest,
+      );
+    }
+
+    let Some(request_node) = graph.node_weight(node_index.clone()) else {
+      return Err(vec![RequestError::Impossible]);
+    };
+
+    match request_node {
+      RequestNode::Root => Err(vec![RequestError::Impossible]),
+      RequestNode::Incomplete => Err(vec![RequestError::Impossible]),
+      RequestNode::Error(errors) => Err(errors.to_owned()),
+      RequestNode::Valid(value) => Ok(value.clone()),
+    }
+  }
+}