From 6704acb87d4f3fc7bea5b69086cf762b9daa1227 Mon Sep 17 00:00:00 2001
From: sgdxbc <sung@comp.nus.edu.sg>
Date: Sat, 30 Mar 2024 05:18:05 +0000
Subject: [PATCH] Checkpoint before switching to Worker based solution

---
 src/cops.rs | 139 +++++++++++++++++++++++++++++++++++++++-------------
 1 file changed, 105 insertions(+), 34 deletions(-)

diff --git a/src/cops.rs b/src/cops.rs
index b08e5acd..e0cc3ea9 100644
--- a/src/cops.rs
+++ b/src/cops.rs
@@ -4,15 +4,18 @@ use std::{
 };
 
 use crate::{
-    event::{erased::OnEvent, Timer},
+    event::{erased::OnEvent, SendEvent, Timer},
     message::Payload,
     net::{events::Recv, SendMessage},
 };
 
+// "key" under COPS context, "id" under Boson's logical clock context
+pub type KeyId = u32;
+
 pub struct Put<V, A> {
-    key: Payload,
+    key: KeyId,
     value: Payload,
-    deps: BTreeMap<Payload, V>,
+    deps: BTreeMap<KeyId, V>,
     client_addr: A,
 }
 
@@ -21,7 +24,7 @@ pub struct PutOk<V> {
 }
 
 pub struct Get<A> {
-    key: Payload,
+    key: KeyId,
     client_addr: A,
 }
 
@@ -31,13 +34,13 @@ pub struct GetOk<V> {
 }
 
 pub struct Sync<V, A> {
-    key: Payload,
+    key: KeyId,
     version: V,
     server_addr: A,
 }
 
 pub struct SyncOk<V> {
-    key: Payload,
+    key: KeyId,
     value: Payload,
     version: V,
 }
@@ -64,28 +67,30 @@ impl<
 }
 
 pub struct CreateVersion<V> {
-    pub previous: V,
+    pub id: KeyId,
+    pub previous: Option<V>,
     pub deps: Vec<V>,
 }
 
-pub struct CreateVersionOk<V>(pub V);
+pub struct CreateVersionOk<V>(pub KeyId, pub V);
 
 // pub struct Client<V> {
 // }
 
-pub struct Server<N, CN, V, A> {
-    store: BTreeMap<Payload, (Payload, V)>,
+pub trait VersionService<V>: SendEvent<CreateVersion<V>> {}
+impl<T: SendEvent<CreateVersion<V>>, V> VersionService<V> for T {}
+
+pub struct Server<N, CN, S, V, A> {
+    store: BTreeMap<KeyId, (Payload, V)>,
+    version_zero: V,
+    pending_puts: Vec<Put<V, A>>,
+    pending_gets: Vec<Get<A>>,
     net: N,
     client_net: CN,
-    pending_puts: Vec<PendingPut<V, A>>,
-}
-
-struct PendingPut<V, A> {
-    message: Put<V, A>,
-    keys: HashSet<Payload>,
+    version_service: S,
 }
 
-impl<N, CN: ClientNet<A, V>, A, V: Clone> OnEvent<Recv<Get<A>>> for Server<N, CN, V, A> {
+impl<N, CN: ClientNet<A, V>, A, V: Clone, S> OnEvent<Recv<Get<A>>> for Server<N, CN, S, V, A> {
     fn on_event(&mut self, Recv(get): Recv<Get<A>>, _: &mut impl Timer) -> anyhow::Result<()> {
         if let Some((value, version)) = self.store.get(&get.key) {
             let get_ok = GetOk {
@@ -94,13 +99,14 @@ impl<N, CN: ClientNet<A, V>, A, V: Clone> OnEvent<Recv<Get<A>>> for Server<N, CN
             };
             return self.client_net.send(get.client_addr, get_ok);
         }
-        // TODO
+        // TODO figure out who to sync with and send Sync
+        self.pending_gets.push(get);
         Ok(())
     }
 }
 
-impl<N: ServerNet<A, V>, CN: ClientNet<A, V>, A, V: PartialOrd + Clone> OnEvent<Recv<Put<V, A>>>
-    for Server<N, CN, V, A>
+impl<N: ServerNet<A, V>, CN: ClientNet<A, V>, A, V: PartialOrd + Clone, S: VersionService<V>>
+    OnEvent<Recv<Put<V, A>>> for Server<N, CN, S, V, A>
 {
     fn on_event(&mut self, Recv(put): Recv<Put<V, A>>, _: &mut impl Timer) -> anyhow::Result<()> {
         if let Some((_, version)) = self.store.get(&put.key) {
@@ -124,23 +130,43 @@ impl<N: ServerNet<A, V>, CN: ClientNet<A, V>, A, V: PartialOrd + Clone> OnEvent<
                         return None;
                     }
                 }
-                Some(dep_key.clone())
+                Some(dep_key)
             })
             .collect::<HashSet<_>>();
         if pending_sync_keys.is_empty() {
-            // TODO
+            let deps = put
+                .deps
+                .keys()
+                .map(|dep_key| self.store.get(dep_key).unwrap().1.clone())
+                .collect::<Vec<_>>();
+            let previous = self.store.get(&put.key).map(|(_, version)| version.clone());
+            // shortcut only when both condition holds
+            // if there are already dependencies on the first version, the version should not start
+            // with `version_zero` anymore
+            // if there is no dependency but it's not the first version, the version still need to
+            // be incremented
+            if deps.is_empty() && previous.is_none() {
+                self.store
+                    .insert(put.key, (put.value, self.version_zero.clone()));
+                let put_ok = PutOk {
+                    version: self.version_zero.clone(),
+                };
+                return self.client_net.send(put.client_addr, put_ok);
+            }
+            self.version_service.send(CreateVersion {
+                id: put.key,
+                previous,
+                deps,
+            })?;
         }
-        self.pending_puts.push(PendingPut {
-            message: put,
-            keys: pending_sync_keys,
-        });
-        // TODO
+        // TODO figure out who to sync with and send Sync
+        self.pending_puts.push(put);
         Ok(())
     }
 }
 
-impl<N: ServerNet<A, V>, CN, A, V: PartialOrd + Clone> OnEvent<Recv<Sync<V, A>>>
-    for Server<N, CN, V, A>
+impl<N: ServerNet<A, V>, CN, A, V: PartialOrd + Clone, S> OnEvent<Recv<Sync<V, A>>>
+    for Server<N, CN, S, V, A>
 {
     fn on_event(&mut self, Recv(sync): Recv<Sync<V, A>>, _: &mut impl Timer) -> anyhow::Result<()> {
         if let Some((value, version)) = self.store.get(&sync.key) {
@@ -157,8 +183,8 @@ impl<N: ServerNet<A, V>, CN, A, V: PartialOrd + Clone> OnEvent<Recv<Sync<V, A>>>
     }
 }
 
-impl<N, CN: ClientNet<A, V>, A, V: PartialOrd + Clone> OnEvent<Recv<SyncOk<V>>>
-    for Server<N, CN, V, A>
+impl<N, CN: ClientNet<A, V>, A, V: PartialOrd + Clone, S: VersionService<V>>
+    OnEvent<Recv<SyncOk<V>>> for Server<N, CN, S, V, A>
 {
     fn on_event(
         &mut self,
@@ -170,9 +196,54 @@ impl<N, CN: ClientNet<A, V>, A, V: PartialOrd + Clone> OnEvent<Recv<SyncOk<V>>>
                 return Ok(());
             }
         }
-        self.store
-            .insert(sync_ok.key, (sync_ok.value, sync_ok.version));
-        // TODO
+        self.store.insert(
+            sync_ok.key,
+            (sync_ok.value.clone(), sync_ok.version.clone()),
+        );
+
+        let mut pending_gets = Vec::new();
+        for get in self.pending_gets.drain(..) {
+            if get.key != sync_ok.key {
+                pending_gets.push(get)
+            } else {
+                let get_ok = GetOk {
+                    value: sync_ok.value.clone(),
+                    version: sync_ok.version.clone(),
+                };
+                self.client_net.send(get.client_addr, get_ok)?
+            }
+        }
+        self.pending_gets = pending_gets;
+
+        for put in &self.pending_puts {
+            if put.deps.iter().all(|(dep_key, dep_version)| {
+                if let Some((_, version)) = self.store.get(dep_key) {
+                    matches!(version.partial_cmp(dep_version), Some(Greater | Equal))
+                } else {
+                    false
+                }
+            }) {
+                self.version_service.send(CreateVersion {
+                    id: put.key,
+                    previous: self.store.get(&put.key).map(|(_, version)| version.clone()),
+                    deps: put.deps.values().cloned().collect(),
+                })?
+            }
+        }
+
+        Ok(())
+    }
+}
+
+impl<N, CN: ClientNet<A, V>, A, V: PartialOrd + Clone, S: VersionService<V>>
+    OnEvent<CreateVersionOk<V>> for Server<N, CN, S, V, A>
+{
+    fn on_event(
+        &mut self,
+        CreateVersionOk(key, version): CreateVersionOk<V>,
+        _: &mut impl Timer,
+    ) -> anyhow::Result<()> {
+        //
         Ok(())
     }
 }