Add initial support for virtual time

safekeeper/src/lib.rs

@@ -13,13 +13,13 @@ pub mod control_file_upgrade;
 pub mod debug_dump;
 pub mod handler;
 pub mod http;
-pub mod sim;
 pub mod json_ctrl;
 pub mod metrics;
 pub mod receive_wal;
 pub mod remove_wal;
 pub mod safekeeper;
 pub mod send_wal;
+pub mod sim;
 pub mod timeline;
 pub mod wal_backup;
 pub mod wal_service;

safekeeper/src/sim/chan.rs

@@ -1,6 +1,6 @@
 use std::{collections::VecDeque, sync::Arc};
 
-use super::sync::{Mutex, Condvar, Park};
+use super::sync::{Condvar, Mutex, Park};
 
 /// FIFO channel with blocking send and receive. Can be cloned and shared between threads.
 #[derive(Clone)]
@@ -15,11 +15,11 @@ struct ChanState<T> {
 
 impl<T: Clone> Chan<T> {
     pub fn new() -> Chan<T> {
-        Chan{
-            shared: Arc::new(ChanState{
+        Chan {
+            shared: Arc::new(ChanState {
                 queue: Mutex::new(VecDeque::new()),
                 condvar: Condvar::new(),
-            })
+            }),
         }
     }
 

safekeeper/src/sim/client.rs

@@ -1,13 +1,17 @@
-use super::{node_os::NodeOs, world::NodeId, proto::AnyMessage};
+use super::{
+    node_os::NodeOs,
+    proto::{AnyMessage, ReplCell},
+    world::NodeId,
+};
 
 /// Copy all data from array to the remote node.
-pub fn run_client(os: NodeOs, data: &[u32], dst: NodeId) {
+pub fn run_client(os: NodeOs, data: &[ReplCell], dst: NodeId) {
     println!("started client");
-    
+
     let sock = os.open_tcp(dst);
     for num in data {
         println!("start send data from client");
-        sock.send(AnyMessage::Just32(num.clone()));
+        sock.send(AnyMessage::ReplCell(num.clone()));
         println!("finish send data from client");
     }
 

safekeeper/src/sim/disk.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use anyhow::Result;
 
-use super::sync::Mutex;
+use super::sync::{Mutex, Park};
 
 pub trait Storage<T> {
     fn flush_pos(&self) -> u32;
@@ -12,7 +12,7 @@ pub trait Storage<T> {
 
 #[derive(Clone)]
 pub struct SharedStorage<T> {
-    state: Arc<Mutex<InMemoryStorage<T>>>,
+    pub state: Arc<Mutex<InMemoryStorage<T>>>,
 }
 
 impl<T> SharedStorage<T> {
@@ -29,17 +29,19 @@ impl<T> Storage<T> for SharedStorage<T> {
     }
 
     fn flush(&mut self) -> Result<()> {
+        Park::yield_thread();
         self.state.lock().flush()
     }
 
     fn write(&mut self, t: T) {
+        Park::yield_thread();
         self.state.lock().write(t);
     }
 }
 
 pub struct InMemoryStorage<T> {
-    data: Vec<T>,
-    flush_pos: u32,
+    pub data: Vec<T>,
+    pub flush_pos: u32,
 }
 
 impl<T> InMemoryStorage<T> {

safekeeper/src/sim/disklog.rs

@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 
-use super::{node_os::NodeOs, disk::Storage, world::NetworkEvent, proto::AnyMessage};
+use super::{disk::Storage, node_os::NodeOs, proto::AnyMessage, world::NodeEvent};
 
 pub struct DiskLog {
     pub map: HashMap<String, u32>,
@@ -25,19 +25,18 @@ impl DiskLog {
 pub fn run_server(os: NodeOs, mut storage: Box<dyn Storage<u32>>) {
     println!("started server");
 
-    let epoll = os.network_epoll();
+    let epoll = os.epoll();
     loop {
         let event = epoll.recv();
         println!("got event: {:?}", event);
         match event {
-            NetworkEvent::Message(msg) => {
-                match msg {
-                    AnyMessage::Just32(num) => {
-                        storage.write(num);
-                    }
+            NodeEvent::Message(msg) => match msg {
+                AnyMessage::ReplCell(num) => {
+                    storage.write(num.value);
                 }
-            }
-            _ => {},
+                _ => {}
+            },
+            _ => {}
         }
     }
 }

safekeeper/src/sim/mod.rs

@@ -1,11 +1,13 @@
-pub mod client;
-pub mod disklog;
-pub mod proto;
-pub mod disk;
-pub mod node_os;
-pub mod world;
-pub mod tcp;
 pub mod chan;
-pub mod sync;
+pub mod client;
+pub mod disk;
+pub mod disklog;
+pub mod node_os;
+pub mod proto;
+#[cfg(test)]
 pub mod start_test;
+pub mod sync;
+pub mod tcp;
+pub mod time;
 pub mod wait_group;
+pub mod world;

safekeeper/src/sim/node_os.rs

@@ -1,6 +1,10 @@
 use std::sync::Arc;
 
-use super::{world::{World, NodeId, Node, NetworkEvent}, tcp::Tcp, chan::Chan};
+use super::{
+    chan::Chan,
+    tcp::Tcp,
+    world::{Node, NodeEvent, NodeId, World},
+};
 
 /// Abstraction with all functions (aka syscalls) available to the node.
 pub struct NodeOs {
@@ -10,10 +14,7 @@ pub struct NodeOs {
 
 impl NodeOs {
     pub fn new(world: Arc<World>, internal: Arc<Node>) -> NodeOs {
-        NodeOs{
-            world,
-            internal,
-        }
+        NodeOs { world, internal }
     }
 
     /// Get the node id.
@@ -27,8 +28,8 @@ impl NodeOs {
         self.world.open_tcp(&self.internal, dst)
     }
 
-    /// Returns a channel to receive events from the network.
-    pub fn network_epoll(&self) -> Chan<NetworkEvent> {
+    /// Returns a channel to receive timers and events from the network.
+    pub fn epoll(&self) -> Chan<NodeEvent> {
         self.internal.network_chan()
     }
 }

safekeeper/src/sim/proto.rs

@@ -3,4 +3,12 @@
 #[derive(Clone, Debug)]
 pub enum AnyMessage {
     Just32(u32),
+    ReplCell(ReplCell),
+}
+
+#[derive(Clone, Debug)]
+pub struct ReplCell {
+    pub value: u32,
+    pub client_id: u32,
+    pub seqno: u32,
 }

safekeeper/src/sim/start_test.rs

@@ -1,6 +1,8 @@
-use std::{sync::Arc, thread::sleep_ms, time::Duration};
+use std::{sync::Arc};
 
-use super::{world::World, client::run_client, disklog::run_server, disk::SharedStorage};
+use super::{
+    client::run_client, disk::SharedStorage, disklog::run_server, proto::ReplCell, world::World,
+};
 
 #[test]
 fn start_simulation() {
@@ -10,17 +12,14 @@ fn start_simulation() {
     let server_id = server_node.id;
 
     // start the client thread
-    let data = [1, 2, 3, 4, 5];
-    client_node.launch(move |os| {
-        run_client(os, &data, server_id)
-    });
+    let u32_data = &[1, 2, 3, 4, 5];
+    let data = u32_to_cells(u32_data, 1);
+    client_node.launch(move |os| run_client(os, &data, server_id));
 
     // start the server thread
     let shared_storage = SharedStorage::new();
     let server_storage = shared_storage.clone();
-    server_node.launch(move |os| {
-        run_server(os, Box::new(server_storage))
-    });
+    server_node.launch(move |os| run_server(os, Box::new(server_storage)));
 
     world.debug_print_state();
     world.await_all();
@@ -30,4 +29,31 @@ fn start_simulation() {
         println!("made a step!");
         world.debug_print_state();
     }
+
+    let disk_data = shared_storage.state.lock().data.clone();
+    assert!(verify_data(&disk_data, &u32_data[..]));
+}
+
+fn u32_to_cells(data: &[u32], client_id: u32) -> Vec<ReplCell> {
+    let mut res = Vec::new();
+    for i in 0..data.len() {
+        res.push(ReplCell {
+            client_id,
+            seqno: i as u32,
+            value: data[i],
+        });
+    }
+    res
+}
+
+fn verify_data(disk_data: &[u32], data: &[u32]) -> bool {
+    if disk_data.len() != data.len() {
+        return false;
+    }
+    for i in 0..data.len() {
+        if disk_data[i] != data[i] {
+            return false;
+        }
+    }
+    true
 }

safekeeper/src/sim/sync.rs

@@ -1,6 +1,9 @@
-use std::{sync::Arc, backtrace::Backtrace, io::{self, Write}};
+use std::{
+    backtrace::Backtrace,
+    sync::Arc,
+};
+
 
-use parking_lot::MutexGuard;
 
 use super::world::{Node, NodeId};
 
@@ -19,7 +22,9 @@ struct CondvarState {
 impl Condvar {
     pub fn new() -> Condvar {
         Condvar {
-            waiters: Mutex::new(CondvarState { waiters: Vec::new() }),
+            waiters: Mutex::new(CondvarState {
+                waiters: Vec::new(),
+            }),
         }
     }
 
@@ -83,17 +88,15 @@ struct ParkState {
 
 impl Park {
     pub fn new(can_continue: bool) -> Arc<Park> {
-        Arc::new(
-            Park {
-                lock: Mutex::new(ParkState {
-                    can_continue,
-                    finished: false,
-                    node_id: None,
-                    backtrace: None,
-                }),
-                cvar: parking_lot::Condvar::new(),
-            }
-        )
+        Arc::new(Park {
+            lock: Mutex::new(ParkState {
+                can_continue,
+                finished: false,
+                node_id: None,
+                backtrace: None,
+            }),
+            cvar: parking_lot::Condvar::new(),
+        })
     }
 
     fn init_state(state: &mut ParkState, node: &Arc<Node>) {
@@ -175,14 +178,17 @@ impl Park {
 
         let mut state = self.lock.lock();
         if state.can_continue {
-            println!("WARN wake() called on a thread that is already waked, node {:?}", state.node_id);
+            println!(
+                "WARN wake() called on a thread that is already waked, node {:?}",
+                state.node_id
+            );
             return;
         }
         state.can_continue = true;
         // and here we park the waiting thread
         self.cvar.notify_all();
         drop(state);
-        
+
         // and here we block the thread that called wake() by defer
         let node = Node::current();
         let mut state = self_park.lock.lock();
@@ -193,7 +199,10 @@ impl Park {
     pub fn internal_world_wake(&self) {
         let mut state = self.lock.lock();
         if state.finished {
-            println!("WARN internal_world_wake() called on a thread that is already waked, node {:?}", state.node_id);
+            println!(
+                "WARN internal_world_wake() called on a thread that is already waked, node {:?}",
+                state.node_id
+            );
             return;
         }
         state.finished = true;
@@ -202,7 +211,7 @@ impl Park {
 
     /// Print debug info about the parked thread.
     pub fn debug_print(&self) {
-        let state = self.lock.lock();
+        let _state = self.lock.lock();
         // println!("PARK: node {:?} wake1={} wake2={}", state.node_id, state.can_continue, state.finished);
         // println!("DEBUG: node {:?} wake1={} wake2={}, trace={:?}", state.node_id, state.can_continue, state.finished, state.backtrace);
     }

safekeeper/src/sim/tcp.rs

@@ -1,8 +1,13 @@
-use std::{sync::{Mutex, Arc}, collections::VecDeque};
+use std::{
+    sync::{Arc},
+};
 
-use anyhow::Result;
 
-use super::{world::{NodeId, Node, NetworkEvent}, proto::AnyMessage};
+
+use super::{
+    proto::AnyMessage,
+    world::{Node, NodeEvent},
+};
 
 /// Simplistic simulation of a bidirectional network stream without reordering (TCP).
 /// There are almost no errors, writes are always successful (but may end up in void).
@@ -15,15 +20,13 @@ pub struct Tcp {
 
 impl Tcp {
     pub fn new(dst: Arc<Node>) -> Tcp {
-        Tcp{
-            dst,
-        }
+        Tcp { dst }
     }
 
     /// Send a message to the other side. It's guaranteed that it will not arrive
     /// before the arrival of all messages sent earlier.
     pub fn send(&self, msg: AnyMessage) {
         // TODO: send to the internal TCP buffer
-        self.dst.network_chan().send(NetworkEvent::Message(msg));
+        self.dst.network_chan().send(NodeEvent::Message(msg));
     }
 }

safekeeper/src/sim/time.rs

@@ -0,0 +1,97 @@
+use std::{cmp::Ordering, collections::BinaryHeap};
+
+
+
+pub struct Timing {
+    /// Current world's time.
+    current_time: u64,
+    /// Pending timers.
+    timers: BinaryHeap<Pending>,
+    /// Global nonce.
+    nonce: u32,
+}
+
+impl Timing {
+    pub fn new() -> Timing {
+        Timing {
+            current_time: 0,
+            timers: BinaryHeap::new(),
+            nonce: 0,
+        }
+    }
+
+    /// Tick-tock the global clock. Return the event ready to be processed
+    /// or move the clock forward and then return the event.
+    pub fn step(&mut self) -> Option<Pending> {
+        if self.timers.len() == 0 {
+            // no future events
+            return None;
+        }
+
+        if !self.is_event_ready() {
+            let next_time = self.timers.peek().unwrap().time;
+            println!("Advancing time from {} to {}", self.current_time, next_time);
+            self.current_time = next_time;
+            assert!(self.is_event_ready());
+        }
+
+        self.timers.pop()
+    }
+
+    /// TODO: write docs
+    pub fn schedule(&mut self, time: u64, event: Event) {
+        self.nonce += 1;
+        let nonce = self.nonce;
+        self.timers.push(Pending { time, nonce, event })
+    }
+
+    /// Return true if there is a ready event.
+    fn is_event_ready(&self) -> bool {
+        self.timers
+            .peek()
+            .map_or(false, |x| x.time <= self.current_time)
+    }
+}
+
+pub struct Pending {
+    pub time: u64,
+    pub nonce: u32,
+    pub event: Event,
+}
+
+impl Pending {
+    pub fn process(&self) {
+        self.event.process();
+    }
+}
+
+// BinaryHeap is a max-heap, and we want a min-heap. Reverse the ordering here
+// to get that.
+impl PartialOrd for Pending {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        (other.time, other.nonce).partial_cmp(&(self.time, self.nonce))
+    }
+}
+
+impl Ord for Pending {
+    fn cmp(&self, other: &Self) -> Ordering {
+        (other.time, other.nonce).cmp(&(self.time, self.nonce))
+    }
+}
+
+impl PartialEq for Pending {
+    fn eq(&self, other: &Self) -> bool {
+        &(other.time, other.nonce) == &(self.time, self.nonce)
+    }
+}
+
+impl Eq for Pending {}
+
+#[derive(Debug)]
+pub enum Event {}
+
+impl Event {
+    fn process(&self) {
+        // TODO:
+    }
+}

safekeeper/src/sim/world.rs

@@ -1,7 +1,18 @@
-use std::{sync::{atomic::AtomicI32, Arc}, cell::RefCell};
-use rand::{rngs::StdRng, SeedableRng, Rng};
+use rand::{rngs::StdRng, Rng, SeedableRng};
+use std::{
+    cell::RefCell,
+    sync::{Arc},
+};
 
-use super::{tcp::Tcp, sync::{Mutex, Park}, chan::Chan, proto::AnyMessage, node_os::NodeOs, wait_group::WaitGroup};
+use super::{
+    chan::Chan,
+    node_os::NodeOs,
+    proto::AnyMessage,
+    sync::{Mutex, Park},
+    tcp::Tcp,
+    time::Timing,
+    wait_group::WaitGroup,
+};
 
 pub type NodeId = u32;
 
@@ -18,15 +29,19 @@ pub struct World {
 
     /// Random number generator.
     rng: Mutex<StdRng>,
+
+    /// Timers and stuff.
+    timing: Mutex<Timing>,
 }
 
 impl World {
     pub fn new() -> World {
-        World{
+        World {
             nodes: Mutex::new(Vec::new()),
             unconditional_parking: Mutex::new(Vec::new()),
             wait_group: WaitGroup::new(),
             rng: Mutex::new(StdRng::seed_from_u64(1337)),
+            timing: Mutex::new(Timing::new()),
         }
     }
 
@@ -52,7 +67,7 @@ impl World {
     }
 
     /// Returns a writable end of a TCP connection, to send src->dst messages.
-    pub fn open_tcp(&self, src: &Arc<Node>, dst: NodeId) -> Tcp {
+    pub fn open_tcp(&self, _src: &Arc<Node>, dst: NodeId) -> Tcp {
         // TODO: replace unwrap() with /dev/null socket.
         let dst = self.get_node(dst).unwrap();
 
@@ -64,37 +79,70 @@ impl World {
         self.wait_group.wait();
     }
 
-    pub fn step(&self) -> bool {
-        self.await_all();
-
+    /// Take a random unconditionally parked thread and return it.
+    fn thread_to_unpark(&self) -> Option<Arc<Park>> {
         let mut parking = self.unconditional_parking.lock();
         if parking.is_empty() {
             // nothing to do, all threads have finished
-            return false;
+            return None;
         }
 
         let chosen_one = self.rng.lock().gen_range(0..parking.len());
         let park = parking.swap_remove(chosen_one);
         drop(parking);
+        Some(park)
+    }
 
-        println!("Waking up park at node {:?}", park.node_id());
-
-        // Wake up the chosen thread. To do that:
-        // 1. Increment the counter of running threads.
-        // 2. Send a singal to continue the thread.
-        self.wait_group.add(1);
-        park.internal_world_wake();
-
-        // to have a clean state after each step, wait for all threads to finish
+    pub fn step(&self) -> bool {
         self.await_all();
-        return true;
+
+        // First try to wake up unconditional thread.
+        let to_resume = self.thread_to_unpark();
+        if let Some(park) = to_resume {
+            println!("Waking up park at node {:?}", park.node_id());
+
+            // Wake up the chosen thread. To do that:
+            // 1. Increment the counter of running threads.
+            // 2. Send a singal to continue the thread.
+            self.wait_group.add(1);
+            park.internal_world_wake();
+
+            // to have a clean state after each step, wait for all threads to finish
+            self.await_all();
+            return true;
+        }
+
+        // Otherwise, all threads are probably waiting for some event.
+        // We'll try to advance virtual time to the next available event.
+        //
+        // This way all code running in simulation is considered to be
+        // instant in terms of "virtual time", and time is advanced only
+        // when code is waiting for external events.
+        let mut timing = self.timing.lock();
+        if let Some(event) = timing.step() {
+            println!("Processing event: {:?}", event.event);
+            event.process();
+            // to have a clean state after each step, wait for all threads to finish
+            self.await_all();
+            return true;
+        }
+
+        false
     }
 
     /// Print full world state to stdout.
     pub fn debug_print_state(&self) {
-        println!("[DEBUG] World state, nodes.len()={:?}, parking.len()={:?}", self.nodes.lock().len(), self.unconditional_parking.lock().len());
+        println!(
+            "[DEBUG] World state, nodes.len()={:?}, parking.len()={:?}",
+            self.nodes.lock().len(),
+            self.unconditional_parking.lock().len()
+        );
         for node in self.nodes.lock().iter() {
-            println!("[DEBUG] node id={:?} status={:?}", node.id, node.status.lock());
+            println!(
+                "[DEBUG] node id={:?} status={:?}",
+                node.id,
+                node.status.lock()
+            );
         }
         for park in self.unconditional_parking.lock().iter() {
             park.debug_print();
@@ -109,7 +157,7 @@ thread_local! {
 /// Internal node state.
 pub struct Node {
     pub id: NodeId,
-    network: Chan<NetworkEvent>,
+    network: Chan<NodeEvent>,
     status: Mutex<NodeStatus>,
     world: Arc<World>,
     join_handle: Mutex<Option<std::thread::JoinHandle<()>>>,
@@ -127,7 +175,7 @@ pub enum NodeStatus {
 
 impl Node {
     pub fn new(id: NodeId, world: Arc<World>) -> Node {
-        Node{
+        Node {
             id,
             network: Chan::new(),
             status: Mutex::new(NodeStatus::NotStarted),
@@ -176,7 +224,7 @@ impl Node {
     }
 
     /// Returns a channel to receive events from the network.
-    pub fn network_chan(&self) -> Chan<NetworkEvent> {
+    pub fn network_chan(&self) -> Chan<NodeEvent> {
         self.network.clone()
     }
 
@@ -220,14 +268,13 @@ impl Node {
 
     /// Get the current node, panics if called from outside of a node thread.
     pub fn current() -> Arc<Node> {
-        CURRENT_NODE.with(|current_node| {
-            current_node.borrow().clone().unwrap()
-        })
+        CURRENT_NODE.with(|current_node| current_node.borrow().clone().unwrap())
     }
 }
 
+/// Network events and timers.
 #[derive(Clone, Debug)]
-pub enum NetworkEvent {
+pub enum NodeEvent {
     Accept,
     Message(AnyMessage),
     // TODO: close?