gg-broadcast/src/main.rs

@@ -42,7 +42,7 @@ fn main() {
         tx.send(msg).unwrap();
     });
 
-    runner.run(rx, None);
+    runner.run(rx);
     let _ = i.join();
     let _ = g.join();
 }

gg-echo/src/main.rs

@@ -39,6 +39,6 @@ fn main() {
 
     let runner = Runner::new(node, out);
 
-    runner.run(rx, None);
+    runner.run(rx);
     i.join().unwrap();
 }

gg-g_counter/src/main.rs

@@ -1,32 +1,32 @@
 use std::{
-    rc::Rc,
     sync::{Arc, Mutex},
+    thread,
+    time::Duration,
 };
 
 use nebkor_maelstrom::{mk_payload, mk_stdin, Body, Message, Node, Runner};
 
 fn main() {
+    let out = std::io::stdout();
+
     let node = Counter;
     let node = Arc::new(Mutex::new(node));
 
-    let runner = Rc::new(Runner::new(node));
+    let runner = Runner::new(node, out);
 
-    let (i, _, rx) = mk_stdin();
+    let (i, extra_input, rx) = mk_stdin();
+    let init = thread::spawn(move || {
+        thread::sleep(Duration::from_millis(10));
+        extra_input
+            .send(Message {
+                body: Body::from_type("kv_init"),
+                ..Default::default()
+            })
+            .unwrap();
+    });
 
-    let on_init = |rnr: &Runner| {
+    runner.run(rx);
-        let payload = mk_payload(&[
+    init.join().unwrap();
-            ("key", KEY.into()),
-            ("from", 0i64.into()),
-            ("to", 0i64.into()),
-            ("create_if_not_exists", true.into()),
-        ]);
-        let body = Body::from_type("cas").with_payload(payload);
-        rnr.send("seq-kv", body);
-    };
-
-    let on_init = Box::new(on_init);
-
-    runner.run(rx, Some(on_init));
     i.join().unwrap();
 }
 
@@ -36,36 +36,24 @@ const KEY: &str = "COUNTER";
 struct Counter;
 
 impl Node for Counter {
-    fn handle<'slf>(&'slf mut self, runner: &'slf Rc<Runner>, req: Message) {
+    fn handle<'slf>(&'slf mut self, runner: &'slf Runner, req: Message) {
         let typ = req.body.typ.as_str();
         let frm = req.src.clone();
         let msg_id = req.body.msg_id.to_owned();
+
         match typ {
-            "add" => {
+            "kv_init" => {
-                let read_runner = runner.clone();
-                let cas_runner = runner.clone();
-                let add_req = req.clone();
-
-                let cas_handler = move |_msg: Message| {
-                    let req = add_req;
-                    cas_runner.reply(&req, Body::from_type("add_ok"));
-                };
-
-                let delta = req.body.payload.get("delta").unwrap().as_i64().unwrap();
-                let read_handler = move |msg: Message| {
-                    let value = msg.body.payload.get("value").unwrap().as_i64().unwrap();
                 let payload = mk_payload(&[
-                        ("from", value.into()),
-                        ("to", (value + delta).into()),
                     ("key", KEY.into()),
+                    ("from", 0i64.into()),
+                    ("to", 0i64.into()),
+                    ("create_if_not_exists", true.into()),
                 ]);
                 let body = Body::from_type("cas").with_payload(payload);
-                    read_runner.rpc("seq-kv", body, Box::new(cas_handler));
+                runner.send("seq-kv", body);
-                };
+            }
-                // kick it off by calling "read" on seq-kv:
+            "add" => {
-                let payload = mk_payload(&[("key", KEY.into())]);
+                runner.reply(&req, Body::from_type("add_ok"));
-                let body = Body::from_type("read").with_payload(payload);
-                runner.rpc("seq-kv", body, Box::new(read_handler));
             }
             "read" => {
                 let rn = runner.clone();

gg-uid/src/main.rs

@@ -26,7 +26,7 @@ fn main() {
 
     let runner = Runner::new(node, out);
 
-    runner.run(rx, None);
+    runner.run(rx);
     i.join().unwrap();
 }
 

nebkor-maelstrom/src/kv.rs

@@ -1,9 +1,7 @@
 /*
-use std::{rc::Rc, sync::Arc};
-
 use serde_json::Value;
 
-use crate::{mk_payload, Body, Message, Result, Runner};
+use crate::Body;
 
 #[derive(Debug, Default, Clone)]
 pub struct Kv {
@@ -11,40 +9,20 @@ pub struct Kv {
 }
 
 impl Kv {
-    pub fn seq() -> Self {
+    pub fn new(service: &str) -> Self {
         Kv {
-            service: "seq-kv".to_string(),
+            service: service.to_string(),
         }
     }
-
+    pub fn read(&self, key: &str) -> Body {
-    pub fn lin() -> Self {
-        Kv {
-            service: "lin-kv".to_string(),
-        }
-    }
-
-    pub fn lww() -> Self {
-        Kv {
-            service: "lww-kv".to_string(),
-        }
-    }
-
-    pub fn read(&self, runner: &Runner, key: &str) -> Result<Value> {
         todo!()
     }
 
-    pub fn write(&self, runner: &Runner, key: &str, val: Value, create: bool) -> Result<()> {
+    pub fn write(&self, key: &str, val: Value, create: bool) -> Body {
         todo!()
     }
 
-    pub fn cas(
+    pub fn cas(&self, key: &str, from: Value, to: Value, create: bool) -> Body {
-        &self,
-        runner: &Runner,
-        key: &str,
-        from: Value,
-        to: Value,
-        create: bool,
-    ) -> Result<()> {
         todo!()
     }
 }

nebkor-maelstrom/src/lib.rs

@@ -1,70 +1,63 @@
 use std::{
     collections::HashMap,
     io::{BufRead, Stdout, Write},
-    rc::Rc,
     sync::{
         atomic::{AtomicU64, AtomicUsize, Ordering},
         mpsc::{channel, Receiver, Sender},
-        Arc, LazyLock, Mutex, OnceLock,
+        Arc, Mutex, OnceLock,
     },
     thread::{self, JoinHandle},
 };
 
 pub mod protocol;
-use protocol::ErrorCode;
 pub use protocol::{Body, Message, Payload};
 use serde_json::Value;
 
 pub mod kv;
 
 pub type DynNode = Arc<Mutex<dyn Node>>;
-pub type Handler = Box<dyn FnOnce(Message)>; // -> Result<Option<Value>>>;
+pub type Handler = Box<dyn FnMut(Message)>;
-pub type OnInit = Box<dyn Fn(&Runner)>;
-
-pub type Result<T> = std::result::Result<T, ErrorCode>;
-
-static MSG_ID: AtomicU64 = AtomicU64::new(0);
-static OUTPUT: LazyLock<Stdout> = LazyLock::new(std::io::stdout);
 
 pub trait Node {
-    fn handle(&mut self, runner: &Rc<Runner>, msg: Message);
+    fn handle(&mut self, runner: &Runner, msg: Message);
 }
 
 #[derive(Clone)]
 pub struct Runner {
+    msg_id: Arc<AtomicU64>,
     node: DynNode,
     node_id: OnceLock<String>,
     nodes: OnceLock<Vec<String>>,
     steps: Arc<AtomicUsize>,
+    output: Arc<Mutex<Stdout>>,
     handlers: Arc<Mutex<HashMap<u64, Handler>>>,
 }
 
 impl Runner {
-    pub fn new(node: DynNode) -> Self {
+    pub fn new(node: DynNode, output: Stdout) -> Self {
         Runner {
             node,
+            msg_id: Arc::new(AtomicU64::new(1)),
             nodes: OnceLock::new(),
             node_id: OnceLock::new(),
             steps: Arc::new(AtomicUsize::new(0)),
+            output: Arc::new(Mutex::new(output)),
             handlers: Default::default(),
         }
     }
 
-    pub fn run(self: &Rc<Self>, input: Receiver<Message>, on_init: Option<OnInit>) {
+    pub fn run(&self, input: Receiver<Message>) {
         for msg in input.iter() {
             let typ = &msg.body.typ;
             if let "init" = typ.as_str() {
                 self.init(&msg);
                 let body = Body::from_type("init_ok");
                 self.reply(&msg, body);
-                if let Some(ref h) = on_init {
-                    h(self);
-                }
             } else {
                 let irt = msg.body.in_reply_to;
                 {
                     let mut g = self.handlers.lock().unwrap();
-                    if let Some(h) = g.remove(&irt) {
+                    if let Some(mut h) = g.remove(&irt) {
                         h(msg.clone());
                     }
                 }
@@ -90,11 +83,11 @@ impl Runner {
     }
 
     pub fn next_msg_id(&self) -> u64 {
-        MSG_ID.fetch_add(1, Ordering::SeqCst)
+        self.msg_id.fetch_add(1, Ordering::SeqCst)
     }
 
     pub fn cur_msg_id(&self) -> u64 {
-        MSG_ID.load(Ordering::SeqCst)
+        self.msg_id.load(Ordering::SeqCst)
     }
 
     pub fn nodes(&self) -> &[String] {
@@ -153,7 +146,7 @@ impl Runner {
     }
 
     fn writeln(&self, msg: &str) {
-        let mut out = OUTPUT.lock();
+        let mut out = self.output.lock().unwrap();
         let msg = format!("{msg}\n");
         out.write_all(msg.as_bytes()).unwrap();
         out.flush().unwrap();
@@ -170,7 +163,7 @@ pub fn mk_stdin() -> (JoinHandle<()>, Sender<Message>, Receiver<Message>) {
 
     let join = thread::spawn(move || {
         let g = stdin.lock();
-        for line in g.lines().map_while(std::result::Result::ok) {
+        for line in g.lines().map_while(Result::ok) {
             if let Ok(msg) = serde_json::from_str(&line) {
                 tx.send(msg).unwrap();
             }