ishan/distributed-systems-flyio
1
0
Fork 0
Code

g-counter complete

Browse Source
This commit is contained in:
Ishan Jain 2024-05-08 18:09:13 +05:30
parent 90fe99f0c6
commit 80b702155f
Signed by: ishan
GPG Key ID: 0506DB2A1CC75C27
3 changed files with 143 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

152
src/lib.rs
View File

@ -4,16 +4,18 @@
pub mod types; pub mod types;
use crate::types::{Message, MessageBody, SeqKvInput}; use crate::types::{Message, MessageBody};
use futures::future::BoxFuture; use futures::future::BoxFuture;
use rand::{rngs::SmallRng, Rng, SeedableRng}; use rand::{rngs::SmallRng, Rng, SeedableRng};
use std::{ use std::{
collections::HashMap, collections::HashMap,
future::Future, future::Future,
io::{stdin, stdout, BufRead, Error as IoError, Stdout, Write}, io::{stdin, stdout, BufRead, Error as IoError, Stdout, Write},
ops::{Deref, DerefMut},
sync::Arc, sync::Arc,
time::{Duration, Instant, SystemTime, UNIX_EPOCH}, time::{Duration, SystemTime, UNIX_EPOCH},
}; };
use types::NodeType;
use tokio::sync::{ use tokio::sync::{
oneshot::{self, Receiver, Sender}, oneshot::{self, Receiver, Sender},
@ -27,6 +29,7 @@ type Handler = Arc<
Message, Message,
Arc<RwLock<Maelstorm>>, Arc<RwLock<Maelstorm>>,
Arc<Mutex<MaelstormIo>>, Arc<Mutex<MaelstormIo>>,
Arc<RwLock<MaelstormCallbacks>>,
) -> BoxFuture<'static, Result<(), String>> ) -> BoxFuture<'static, Result<(), String>>
+ Send + Send
+ Sync, + Sync,
@ -35,13 +38,11 @@ type Handler = Arc<
pub struct Maelstorm { pub struct Maelstorm {
pub node: Node, pub node: Node,
pub handlers: HashMap<String, Handler>, pub handlers: HashMap<String, Handler>,
callbacks: HashMap<u64, Callback>,
} }
pub struct Node { pub struct Node {
pub id: String, pub id: String,
pub counter: u64, pub counter: u64,
pub other_counter: u64,
pub nodes: Vec<String>, pub nodes: Vec<String>,
rng: SmallRng, rng: SmallRng,
} }
@ -49,6 +50,27 @@ pub struct MaelstormIo {
stdout: Stdout, stdout: Stdout,
} }
pub struct MaelstormCallbacks(HashMap<u64, Callback>);
impl Default for MaelstormCallbacks {
fn default() -> Self {
Self(HashMap::new())
}
}
impl Deref for MaelstormCallbacks {
type Target = HashMap<u64, Callback>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl DerefMut for MaelstormCallbacks {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
impl Default for Maelstorm { impl Default for Maelstorm {
fn default() -> Self { fn default() -> Self {
let seed = SystemTime::now() let seed = SystemTime::now()
@ -58,12 +80,10 @@ impl Default for Maelstorm {
node: Node { node: Node {
id: String::new(), id: String::new(),
counter: 0, counter: 0,
other_counter: 0,
nodes: Vec::new(), nodes: Vec::new(),
rng: SmallRng::seed_from_u64(seed.as_nanos() as u64), rng: SmallRng::seed_from_u64(seed.as_nanos() as u64),
}, },
handlers: HashMap::new(), handlers: HashMap::new(),
callbacks: HashMap::new(),
} }
} }
} }
@ -84,8 +104,9 @@ impl Default for MaelstormIo {
} }
impl Maelstorm { impl Maelstorm {
pub async fn run(self, io: MaelstormIo) { pub async fn run(self, io: MaelstormIo, callbacks: MaelstormCallbacks) {
let program = Arc::new(RwLock::new(self)); let program = Arc::new(RwLock::new(self));
let callbacks = Arc::new(RwLock::new(callbacks));
let io = Arc::new(Mutex::new(io)); let io = Arc::new(Mutex::new(io));
let stdin = stdin(); let stdin = stdin();
@ -115,13 +136,9 @@ impl Maelstorm {
} }
}; };
eprintln!("INCOMING MESSAGE = {:?}", message);
if let Some(reply_msg_id) = message.body.in_reply_to { if let Some(reply_msg_id) = message.body.in_reply_to {
eprintln!("REPLY ID FOR CALLBACK FOUND = {:?}", message); let mut callbacks = callbacks.write().await;
let mut program = program.write().await; let callback = match callbacks.remove(&reply_msg_id) {
let callback = match program.callbacks.remove(&reply_msg_id) {
Some(v) => v, Some(v) => v,
None => { None => {
eprintln!("no callback for msg with reply id {}", reply_msg_id); eprintln!("no callback for msg with reply id {}", reply_msg_id);
@ -131,8 +148,6 @@ impl Maelstorm {
if let Err(e) = callback(message.clone()).await { if let Err(e) = callback(message.clone()).await {
eprintln!("error in callback: {e}"); eprintln!("error in callback: {e}");
} else {
eprintln!("SUCCESSFULLY CALLED CALLBACK FOR = {:?}", message);
} }
continue; continue;
@ -152,8 +167,9 @@ impl Maelstorm {
let pc = program.clone(); let pc = program.clone();
let io = io.clone(); let io = io.clone();
let c = callbacks.clone();
tokio::spawn(async move { tokio::spawn(async move {
if let Err(e) = handler(message, pc, io).await { if let Err(e) = handler(message, pc, io, c).await {
eprintln!("error in serving request: {}", e); eprintln!("error in serving request: {}", e);
} }
}); });
@ -167,7 +183,12 @@ impl Maelstorm {
pub fn register<F, Fut>(&mut self, name: &str, func: F) pub fn register<F, Fut>(&mut self, name: &str, func: F)
where where
F: Fn(Message, Arc<RwLock<Maelstorm>>, Arc<Mutex<MaelstormIo>>) -> Fut F: Fn(
Message,
Arc<RwLock<Maelstorm>>,
Arc<Mutex<MaelstormIo>>,
Arc<RwLock<MaelstormCallbacks>>,
) -> Fut
+ Send + Send
+ Sync + Sync
+ 'static, + 'static,
@ -175,55 +196,48 @@ impl Maelstorm {
{ {
self.handlers.insert( self.handlers.insert(
name.to_string(), name.to_string(),
Arc::new(move |a, b, c| Box::pin(func(a, b, c))), Arc::new(move |a, b, c, d| Box::pin(func(a, b, c, d))),
); );
} }
pub async fn sync_rpc( pub async fn sync_rpc(
&mut self, &self,
io: Arc<Mutex<MaelstormIo>>, io: Arc<Mutex<MaelstormIo>>,
callbacks: Arc<RwLock<MaelstormCallbacks>>,
msg: Message, msg: Message,
) -> Result<Message, String> { ) -> Result<Message, String> {
let (tx, rx): (Sender<Message>, Receiver<Message>) = oneshot::channel(); let (tx, rx): (Sender<Message>, Receiver<Message>) = oneshot::channel();
let m = msg.clone(); let m = msg.clone();
self.rpc(io, msg, async |msg: Message| match tx.send(msg.clone()) { self.rpc(
Ok(v) => { io.clone(),
eprintln!("SENT RESPONSE INTO CHANNEL = {:?}", msg); callbacks.clone(),
msg,
Ok(()) async |msg: Message| match tx.send(msg.clone()) {
} Ok(_) => Ok(()),
Err(e) => { Err(e) => {
format!("error in sending to tx chan: {:?}", e); format!("error in sending to tx chan: {:?}", e);
Ok(()) Ok(())
} }
}) },
)
.await .await
.map_err(|e| e.to_string())?; .map_err(|e| e.to_string())?;
let t = Instant::now();
match tokio::time::timeout(Duration::from_secs(1), rx).await { match tokio::time::timeout(Duration::from_secs(1), rx).await {
Ok(result) => { Ok(result) => match result {
eprintln!( Ok(v) => Ok(v),
"GOT RESULT BEFORE TIMEOUT = {:?} elapsed = {:?}", Err(e) => {
result, eprintln!("sync callback error: {}", e);
t.elapsed() Err(e.to_string())
);
match result {
Ok(v) => Ok(v),
Err(e) => {
eprintln!("sync callback error: {}", e);
Err(e.to_string())
}
} }
} },
Err(e) => { Err(e) => {
eprintln!( eprintln!(
"sync callback timeout: {} msg = {:?} callback queue {:?}", "sync callback timeout: {} msg = {:?} callback queue {:?}",
e, e,
m, m,
self.callbacks.keys() callbacks.read().await.keys()
); );
Err(e.to_string()) Err(e.to_string())
} }
@ -231,8 +245,9 @@ impl Maelstorm {
} }
async fn rpc<F, Fut>( async fn rpc<F, Fut>(
&mut self, &self,
io: Arc<Mutex<MaelstormIo>>, io: Arc<Mutex<MaelstormIo>>,
callbacks: Arc<RwLock<MaelstormCallbacks>>,
mut msg: Message, mut msg: Message,
handler: F, handler: F,
) -> Result<(), IoError> ) -> Result<(), IoError>
@ -242,9 +257,8 @@ impl Maelstorm {
{ {
let next_msg_id = msg.body.msg_id.unwrap(); let next_msg_id = msg.body.msg_id.unwrap();
eprintln!("INSERTING {} {:?}", next_msg_id, msg); let mut callbacks = callbacks.write().await;
self.callbacks callbacks.insert(next_msg_id, Box::new(move |a| Box::pin(handler(a))));
.insert(next_msg_id, Box::new(move |a| Box::pin(handler(a))));
msg.body.msg_id = Some(next_msg_id); msg.body.msg_id = Some(next_msg_id);
@ -279,7 +293,7 @@ impl Maelstorm {
io.write(&out).map(|_| Ok(()))? io.write(&out).map(|_| Ok(()))?
} }
pub fn id(&self) -> String { pub async fn id(&self) -> String {
self.node.id.clone() self.node.id.clone()
} }
} }
@ -288,10 +302,14 @@ impl Maelstorm {
pub async fn read_counter( pub async fn read_counter(
&mut self, &mut self,
io: Arc<Mutex<MaelstormIo>>, io: Arc<Mutex<MaelstormIo>>,
callbacks: Arc<RwLock<MaelstormCallbacks>>,
node_to_read: String, node_to_read: String,
) -> Result<u64, String> { ) -> Result<u64, String> {
let msg_id: u64 = self.node.rng.gen(); let msg_id: u64 = self.node.rng.gen();
let mut value = HashMap::new();
value.insert("key".to_string(), NodeType::String(node_to_read));
let msg = Message { let msg = Message {
id: None, id: None,
src: self.node.id.clone(), src: self.node.id.clone(),
@ -300,31 +318,31 @@ impl Maelstorm {
msg_id: Some(msg_id), msg_id: Some(msg_id),
in_reply_to: None, in_reply_to: None,
message_type: "read".to_string(), message_type: "read".to_string(),
message_body: serde_json::to_value(SeqKvInput { message_body: value,
key: node_to_read,
value: None,
})
.unwrap(),
}, },
}; };
let mut result = match self.sync_rpc(io, msg).await { let result = match self.sync_rpc(io, callbacks, msg).await {
Ok(v) => v, Ok(v) => v,
Err(e) if e.to_string() == "deadline has elapsed" => return Ok(0), Err(e) if e.to_string() == "deadline has elapsed" => return Ok(0),
Err(e) => return Err(e), Err(e) => return Err(e),
}; };
let body: SeqKvInput = serde_json::from_value(result.body.message_body.take()) match result.body.message_body.get("value") {
.map_err(|e| format!("error in parsing response body: {}", e))?; None => Ok(0),
Some(v) => Ok(v.u64()),
Ok(body.value.unwrap()) }
} }
pub async fn write_counter( pub async fn write_counter(
&mut self, &mut self,
io: Arc<Mutex<MaelstormIo>>, io: Arc<Mutex<MaelstormIo>>,
callbacks: Arc<RwLock<MaelstormCallbacks>>,
val: u64, val: u64,
) -> Result<(), String> { ) -> Result<(), String> {
let msg_id: u64 = self.node.rng.gen(); let msg_id: u64 = self.node.rng.gen();
let mut value = HashMap::new();
value.insert("key".to_string(), NodeType::String(self.node.id.clone()));
value.insert("value".to_string(), NodeType::U64(val));
let msg = Message { let msg = Message {
id: None, id: None,
@ -334,23 +352,15 @@ impl Maelstorm {
msg_id: Some(msg_id), msg_id: Some(msg_id),
in_reply_to: None, in_reply_to: None,
message_type: "write".to_string(), message_type: "write".to_string(),
message_body: serde_json::to_value(SeqKvInput { message_body: value,
key: self.node.id.clone(),
value: Some(val),
})
.unwrap(),
}, },
}; };
let mut result = match self.sync_rpc(io, msg).await { let _result = match self.sync_rpc(io, callbacks, msg).await {
Ok(v) => v, Ok(v) => v,
Err(e) if e.to_string() == "deadline has elapsed" => return Ok(()), Err(e) if e.to_string() == "deadline has elapsed" => return Ok(()),
Err(e) => return Err(e), Err(e) => return Err(e),
}; };
// TODO: could return the parsed confirmation from here
serde_json::from_value(result.body.message_body.take())
.map_err(|e| format!("error in parsing response body: {}", e))?;
Ok(()) Ok(())
} }
} }

69
src/main.rs
View File

@ -1,10 +1,10 @@
#![feature(async_closure)] #![feature(async_closure)]
use distributed_systems_flyio::{ use distributed_systems_flyio::{
types::{GrowCounterReadMessage, GrowCounterUpdateMessage, InitMessage, Message}, types::{Message, NodeType},
Maelstorm, MaelstormIo, Maelstorm, MaelstormCallbacks, MaelstormIo,
}; };
use std::sync::Arc; use std::{collections::HashMap, sync::Arc};
use tokio::sync::{Mutex, RwLock}; use tokio::sync::{Mutex, RwLock};
#[tokio::main] #[tokio::main]
@ -16,13 +16,14 @@ async fn main() {
// TODO: Replace error string with a type // TODO: Replace error string with a type
async |mut msg: Message, async |mut msg: Message,
program: Arc<RwLock<Maelstorm>>, program: Arc<RwLock<Maelstorm>>,
io: Arc<Mutex<MaelstormIo>>| io: Arc<Mutex<MaelstormIo>>,
_: Arc<RwLock<MaelstormCallbacks>>|
-> Result<(), String> { -> Result<(), String> {
let message_body: InitMessage = serde_json::from_value(msg.body.message_body.take())
.map_err(|e| format!("error in parsing response body: {}", e))?;
let mut program = program.write().await; let mut program = program.write().await;
program.init(message_body.node_id, message_body.nodes);
let node_id = msg.body.message_body.get("node_id").unwrap().string();
let nodes = msg.body.message_body.get("node_ids").unwrap().vec_string();
program.init(node_id, nodes);
msg.body.message_type = "init_ok".into(); msg.body.message_type = "init_ok".into();
program program
@ -36,31 +37,20 @@ async fn main() {
"add", "add",
async |mut msg: Message, async |mut msg: Message,
program: Arc<RwLock<Maelstorm>>, program: Arc<RwLock<Maelstorm>>,
io: Arc<Mutex<MaelstormIo>>| io: Arc<Mutex<MaelstormIo>>,
callbacks: Arc<RwLock<MaelstormCallbacks>>|
-> Result<(), String> { -> Result<(), String> {
let body: GrowCounterUpdateMessage =
serde_json::from_value(msg.body.message_body.take())
.map_err(|e| format!("error in parsing response body: {}", e))?;
msg.body.message_type = "add_ok".into(); msg.body.message_type = "add_ok".into();
let mut program = program.write().await; let mut program = program.write().await;
program.node.counter += body.delta; program.node.counter += msg.body.message_body.get("delta").unwrap().u64();
let current = program.node.counter; let current = program.node.counter;
program.write_counter(io.clone(), current).await?; program
.write_counter(io.clone(), callbacks.clone(), current)
let mut other_counter = 0; .await?;
for node in program.node.nodes.clone() {
if node == msg.dest {
continue;
}
let resp = program.read_counter(io.clone(), node).await?;
other_counter += resp;
}
program.node.other_counter = other_counter;
msg.body.message_body = HashMap::new();
program program
.reply(io, msg) .reply(io, msg)
.await .await
@ -72,30 +62,32 @@ async fn main() {
"read", "read",
async |mut msg: Message, async |mut msg: Message,
program: Arc<RwLock<Maelstorm>>, program: Arc<RwLock<Maelstorm>>,
io: Arc<Mutex<MaelstormIo>>| io: Arc<Mutex<MaelstormIo>>,
callbacks: Arc<RwLock<MaelstormCallbacks>>|
-> Result<(), String> { -> Result<(), String> {
msg.body.message_type = "read_ok".into(); msg.body.message_type = "read_ok".into();
let mut program = program.write().await;
let mut other_counter = 0; let mut other_counter = 0;
let mut program = program.write().await;
for node in program.node.nodes.clone() { for node in program.node.nodes.clone() {
if node == msg.dest { if node == msg.dest {
continue; continue;
} }
let resp = program.read_counter(io.clone(), node).await?; let resp = program
.read_counter(io.clone(), callbacks.clone(), node)
.await?;
other_counter += resp; other_counter += resp;
} }
program.node.other_counter = other_counter;
eprintln!( let mut resp = HashMap::new();
"READ OP {} {}", resp.insert(
program.node.counter, program.node.other_counter "value".to_string(),
NodeType::U64(program.node.counter + other_counter),
); );
let resp = GrowCounterReadMessage {
value: program.node.counter + program.node.other_counter, msg.body.message_body = resp;
};
msg.body.message_body = serde_json::to_value(resp).unwrap();
program program
.reply(io, msg) .reply(io, msg)
@ -104,7 +96,8 @@ async fn main() {
}, },
); );
let callbacks = MaelstormCallbacks::default();
let io = MaelstormIo::default(); let io = MaelstormIo::default();
program.run(io).await; program.run(io, callbacks).await;
} }

54
src/types.rs
View File

@ -1,3 +1,5 @@
use std::collections::HashMap;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize, Clone, Deserialize)] #[derive(Debug, Serialize, Clone, Deserialize)]
@ -21,33 +23,39 @@ pub struct MessageBody {
pub message_type: String, pub message_type: String,
#[serde(flatten)] #[serde(flatten)]
pub message_body: serde_json::Value, pub message_body: HashMap<String, NodeType>,
} }
#[derive(Debug, Serialize, Deserialize)] #[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(deny_unknown_fields)] #[serde(untagged)]
pub struct InitMessage { pub enum NodeType {
pub node_id: String, String(String),
#[serde(rename = "node_ids")] U64(u64),
pub nodes: Vec<String>, VecString(Vec<String>),
} }
#[derive(Debug, Serialize, Deserialize)] impl NodeType {
#[serde(deny_unknown_fields)] pub fn u64(&self) -> u64 {
pub struct GrowCounterUpdateMessage { if let NodeType::U64(v) = self {
pub delta: u64, return *v;
} } else {
unreachable!()
}
}
#[derive(Debug, Serialize, Deserialize)] pub fn string(&self) -> String {
#[serde(deny_unknown_fields)] if let NodeType::String(s) = self {
pub struct GrowCounterReadMessage { return s.clone();
pub value: u64, } else {
} unreachable!()
}
}
#[derive(Debug, Serialize, Deserialize)] pub fn vec_string(&self) -> Vec<String> {
#[serde(deny_unknown_fields)] if let NodeType::VecString(s) = self {
pub struct SeqKvInput { return s.clone();
pub key: String, } else {
#[serde(skip_serializing_if = "Option::is_none")] unreachable!()
pub value: Option<u64>, }
}
} }