Merge branch 'main' into emlautarom1/expbackoff

Author: emlautarom1

Cargo.lock

@@ -24,21 +24,23 @@ publish = false
 
 [workspace.dependencies]
 axum = "0.8.6"
+blst = "0.3.13"
+cancellation = "0.1.0"
 chrono = { version = "0.4", features = ["serde"] }
+crossbeam = "0.8.4"
 hex = { version = "^0.4.3" }
+prost = "0.14"
+prost-build = "0.14"
+prost-types = "0.14"
+rand = { version = "0.8", features = ["std_rng"] }
+rand_core = "0.6"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "^1.0" }
+thiserror = "2.0.12"
 tokio = { version = "1", features = ["full"] }
 tower = { version = "0.5.2", features = ["full"] }
 libp2p = { version = "0.56", features = ["full", "secp256k1"] }
-prost = "0.14"
-prost-types = "0.14"
-prost-build = "0.14"
-blst = "0.3.13"
-rand_core = "0.6"
-thiserror = "2.0.12"
-rand = {version = "0.8", features = ["std_rng"]}
-uuid = {version = "1.16", features = ["serde", "v4"] }
+uuid = { version = "1.16", features = ["serde", "v4"] }
 serde_with = { version = "3", features = ["hex", "base64"] }
 base64 = { version = "0.22.1" }
 sha3 = { version = "0.10.8" }

Cargo.toml

@@ -7,8 +7,19 @@ license.workspace = true
 publish.workspace = true
 
 [dependencies]
+cancellation.workspace = true
+chrono.workspace = true
+crossbeam.workspace = true
+hex.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+thiserror.workspace = true
+libp2p.workspace = true
+prost.workspace = true
+prost-types.workspace = true
+
+[dev-dependencies]
+rand.workspace = true
 libp2p.workspace = true
 prost.workspace = true
 prost-types.workspace = true
@@ -19,4 +30,4 @@ chrono.workspace = true
 charon-build-proto.workspace = true
 
 [lints]
-workspace = true
+workspace = true

crates/charon-core/Cargo.toml

@@ -4,6 +4,7 @@
 //! This crate provides the fundamental building blocks, data structures, and
 //! core algorithms used throughout the Charon system.
 
+pub mod qbft;
 /// Types for the Charon core.
 pub mod types;
 

crates/charon-core/src/lib.rs

@@ -0,0 +1,153 @@
+use crossbeam::channel as mpmc;
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+    thread,
+    time::{Duration, Instant},
+};
+
+#[derive(Clone)]
+pub struct FakeClock {
+    inner: Arc<Mutex<FakeClockInner>>,
+}
+
+struct FakeClockInner {
+    start: Instant,
+    now: Instant,
+    last_id: usize,
+    clients: HashMap<usize, (mpmc::Sender<Instant>, Instant)>,
+}
+
+impl FakeClock {
+    pub fn new(now: Instant) -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(FakeClockInner {
+                start: now,
+                now,
+                last_id: 1,
+                clients: Default::default(),
+            })),
+        }
+    }
+
+    pub fn new_timer(
+        &self,
+        duration: Duration,
+    ) -> (
+        mpmc::Receiver<Instant>,
+        Box<dyn Fn() + Send + Sync + 'static>,
+    ) {
+        let (tx, rx) = mpmc::bounded::<Instant>(1);
+
+        let client_id = {
+            let mut inner = self.inner.lock().unwrap();
+            let id = inner.last_id;
+            let deadline = inner.now + duration;
+
+            inner.last_id += 1;
+            inner.clients.insert(id, (tx, deadline));
+
+            id
+        };
+
+        let inner = Arc::clone(&self.inner);
+        let cancel = Box::new(move || {
+            let mut inner = inner.lock().unwrap();
+            inner.clients.remove(&client_id);
+        });
+
+        (rx, cancel)
+    }
+
+    pub fn advance(&self, duration: Duration) {
+        // Advance time and collect expired senders under lock, but perform sends
+        // without holding lock.
+        let mut expired = vec![];
+
+        let now = {
+            let mut inner = self.inner.lock().unwrap();
+            inner.now += duration;
+            let now = inner.now;
+
+            for (&id, (ch, deadline)) in inner.clients.iter() {
+                if *deadline <= now {
+                    expired.push((id, ch.clone()));
+                }
+            }
+
+            for (id, _) in expired.iter() {
+                inner.clients.remove(id);
+            }
+
+            now
+        };
+
+        for (_, ch) in expired {
+            let _ = ch.send(now);
+        }
+    }
+
+    pub fn elapsed(&self) -> Duration {
+        let inner = self.inner.lock().unwrap();
+        inner.now - inner.start
+    }
+
+    pub fn cancel(&self) {
+        let mut inner = self.inner.lock().unwrap();
+        inner.clients.clear();
+    }
+}
+
+impl Drop for FakeClock {
+    fn drop(&mut self) {
+        self.cancel();
+    }
+}
+
+#[test]
+fn multiple_threads_timers() {
+    let clock = FakeClock::new(Instant::now());
+
+    let start = Instant::now();
+    thread::scope(|s| {
+        let c1 = clock.clone();
+        let (ch_1, _) = c1.new_timer(Duration::from_secs(5));
+        s.spawn(move || {
+            let _ = ch_1.recv();
+        });
+
+        let c2 = clock.clone();
+        let (ch_2, _) = c2.new_timer(Duration::from_secs(5));
+        s.spawn(move || {
+            let _ = ch_2.recv();
+        });
+
+        clock.advance(Duration::from_secs(6));
+    });
+
+    println!("start={:?}, clock={:?}", start.elapsed(), clock.elapsed());
+}
+
+#[test]
+fn multiple_threads_cancellation() {
+    let clock = FakeClock::new(Instant::now());
+
+    let start = Instant::now();
+    thread::scope(|s| {
+        let c1 = clock.clone();
+        let (ch_1, _) = c1.new_timer(Duration::from_secs(5));
+        s.spawn(move || {
+            let _ = ch_1.recv();
+        });
+
+        let c2 = clock.clone();
+        let (ch_2, _) = c2.new_timer(Duration::from_secs(5));
+        s.spawn(move || {
+            let _ = ch_2.recv();
+        });
+
+        clock.cancel();
+    });
+
+    println!("start={:?}, clock={:?}", start.elapsed(), clock.elapsed());
+}