Gossip Glomers
is a distributed systems challenge created by fly.io
, where participants are expected to work around a simulating library called maelstrom that simulates failures and latencies in a distributed system. The library provided works with go, but being the rustacean that I am, I decided to rewrite the library in Rust, and implement solutions to the challenges. The specifications
are publicly available (and well-documented) so it was a breeze to go through. In this post, I will only explain the solutions to the challenge upto the efficient broadcast challenge.
About this post.
This post is quite code-heavy as I expect the posts in this series to be. The challenge encompasses a lot of concepts in distributed systems, and frankly, squeezing all solutions into a single post will not make any sense. Instead, they will be spread out over posts in this series.
Setting up the project
The ‘maelstrom’ binary can be downloaded and it unsurprisingly works right out of the box. For starters, I created the following project structure:
1.
2├── Cargo.lock
3├── Cargo.toml
4├── maelstrom
5 ├── maelstrom
6 └── ..
7└── src
8 ├── main.rs
9 └── message.rsThe main.rs file is responsible to run the entire application, and message.rs describes the types of messages we support.
The message.rs file has the following contents:
1use serde::{Deserialize, Serialize};
2
3#[derive(Debug, Serialize, Deserialize, Clone)]
4#[serde(rename_all = "snake_case")]
5pub struct Message {
6 pub(crate) src: String,
7 pub(crate) dest: String,
8 pub(crate) body: MessageBody,
9}
10
11#[derive(Debug, Serialize, Deserialize, Clone)]
12#[serde(tag = "type", rename_all = "snake_case")]
13pub enum MessageBody {
14 Echo {
15 msg_id: i32,
16 echo: String,
17 },
18 EchoOk {
19 msg_id: i32,
20 in_reply_to: i32,
21 echo: String,
22 },
23 Init {
24 msg_id: i32,
25 node_id: String,
26 node_ids: Vec<String>,
27 },
28 InitOk {
29 in_reply_to: i32,
30 },
31}One thing that I missed in the beginning was to return an InitOk message (which was outlined in the spec). Since the go library does that automatically, and we are not using go, we need to implement everything from scratch. Using serde makes this whole thing a breeze. The common fields go into Message struct, and fields specific to each type of message go to MessageBody. The body is named type in the incoming json message, and it seamlessly works with the following message for deserialization:
1{
2 "src": "n0",
3 "dest": "n1",
4 "body": {
5 "type": "echo_ok",
6 "msg_id": 123,
7 "echo": "Echo echo"
8 }
9}The next thing would be to implement methods to read/write from/to stdin/stdout.
1fn main() -> Result<()> {
2 let mut buffer = String::new();
3 let mut node: Node = Default::default();
4
5 loop {
6 buffer.clear();
7 let len = stdin().read_line(&mut buffer)?;
8 if len == 0 {
9 break;
10 }
11
12 let message: Message = match serde_json::from_str(&buffer) {
13 Ok(msg) => msg,
14 Err(e) => {
15 eprintln!("Error parsing message: {}", e);
16 continue;
17 }
18 };
19
20 if let Err(e) = node.handle_message(message) {
21 eprintln!("Error handling message: {}", e);
22 }
23 }
24
25 Ok(())
26}The node does not need to be concurrent - we process messages serially. The code keeps reading from stdin, creating a message (after deserializing), and passing the deserialized messages to the node. Next, the implementation for Node type.
1#[derive(Debug, Default, Clone)]
2struct Node {
3 id: String,
4 node_ids: Vec<String>,
5 current_msg_id: i32,
6}
7
8impl Node {
9 fn new(id: String, node_ids: Vec<String>) -> Self {
10 Node {
11 id,
12 node_ids,
13 current_msg_id: 0,
14 }
15 }
16
17 fn get_next_message_id(&mut self) -> i32 {
18 }
19
20 fn handle_echo(&mut self, echo_in: Message) -> Result<()> {
21 }
22
23 fn handle_message(&mut self, message: Message) -> Result<()> {
24 }
25}The Node struct itself is very simple - it stores the node’s id, the node_ids of all nodes in the distributed system, and stores the current message id to use (no need to use atomic ints since we are just processing stuff in a single-thread). Next would be the implementation of each of these handlers. The get_next_message_id itself is very simple:
1fn get_next_message_id(&mut self) -> i32 {
2 let current_id = self.current_msg_id;
3 self.current_msg_id += 1;
4 current_id
5}A very basic echo
In order to handle messages in the Node, we dispatch it to the handle_message function from our main function:
1fn handle_message(&mut self, message: Message) -> Result<()> {
2 match message.body.clone() {
3 MessageBody::Echo { .. } => self.handle_echo(message),
4 MessageBody::Init { node_id, msg_id, node_ids } => {
5 // Initialize the node
6 self.id = node_id;
7 self.node_ids = node_ids;
8
9 // Send init_ok response
10 let reply_message = Message {
11 src: self.id.clone(),
12 dest: message.clone().src,
13 body: MessageBody::InitOk {
14 in_reply_to: msg_id,
15 },
16 };
17 println!("{}", serde_json::to_string(&reply_message)?);
18 eprintln!("{}", serde_json::to_string(&reply_message)?);
19
20 Ok(())
21 }
22 _ => {
23 eprintln!("Unhandled message type");
24 Ok(())
25 }
26 }
27}The handle_message message is what brings it all together. We check the message type. If it’s the init message, we send a ok response for init and initialize our node. Otherwise, we dispatch the message to a relevant message handler. As we can see in the case of the echo message:
1fn handle_echo(&mut self, echo_in: Message) -> Result<()> {
2 if let MessageBody::Echo { msg_id, echo } = echo_in.body {
3 let reply_message = Message {
4 src: self.id.clone(),
5 dest: echo_in.src,
6 body: MessageBody::EchoOk {
7 msg_id: self.get_next_message_id(),
8 in_reply_to: msg_id,
9 echo,
10 },
11 };
12 println!("{}", serde_json::to_string(&reply_message)?);
13 eprintln!("{}", serde_json::to_string(&reply_message)?);
14 }
15 Ok(())
16}Even though the message type can never be anything other that MessageBody::Echo, we still use it to cast the message (I guess this would’ve been better in C++, since we have two vtable lookups, one in handle_message and another in handle_echo).
Testing this with maelstrom library, we can see everything works:
1$ ./maelstrom/maelstrom test -w echo --bin target/release/gossiprs \
2 --node-count 1 --time-limit 10
3
4...
5... bunch of output
6...
7
8Everything looks good! ヽ(‘ー`)ノImproving message handling:
Since the first message is always the init message, we can check to see if that is actually the case, and only process messages after the first one (since we might miss a init message, and reach an invalid node state).
1diff --git a/src/main.rs b/src/main.rs
2index 8cb2685..434d943 100644
3--- a/src/main.rs
4+++ b/src/main.rs
5@@ -6,6 +6,7 @@ use std::io::stdin;
6 mod message;
7
8 #[derive(Debug, Default, Clone)]
9+#[allow(unused)]
10 struct Node {
11 id: String,
12 node_ids: Vec<String>,
13@@ -44,27 +45,26 @@ impl Node {
14 Ok(())
15 }
16
17+ fn handle_init(&mut self, msg_id: i32, src: String) -> Result<()> {
18+ // Send init_ok response
19+ let reply_message = Message {
20+ src: self.id.clone(),
21+ dest: src,
22+ body: MessageBody::InitOk {
23+ in_reply_to: msg_id,
24+ },
25+ };
26+ println!("{}", serde_json::to_string(&reply_message)?);
27+ eprintln!("{}", serde_json::to_string(&reply_message)?);
28+
29+ Ok(())
30+ }
31+
32 fn handle_message(&mut self, message: Message) -> Result<()> {
33 match message.body.clone() {
34 MessageBody::Echo { .. } => self.handle_echo(message),
35- MessageBody::Init { node_id, msg_id, node_ids } => {
36- // Initialize the node
37- self.id = node_id;
38- self.node_ids = node_ids;
39-
40- // Send init_ok response
41- let reply_message = Message {
42- src: self.id.clone(),
43- dest: message.clone().src,
44- body: MessageBody::InitOk {
45- in_reply_to: msg_id,
46- },
47- };
48- println!("{}", serde_json::to_string(&reply_message)?);
49- eprintln!("{}", serde_json::to_string(&reply_message)?);
50+ MessageBody::Init { msg_id, .. } => self.handle_init(msg_id, message.src.clone()),
51
52- Ok(())
53- }
54 _ => {
55 eprintln!("Unhandled message type");
56 Ok(())
57@@ -76,26 +76,38 @@ impl Node {
58 #[tokio::main]
59 async fn main() -> Result<()> {
60 let mut buffer = String::new();
61- let mut node: Node = Default::default();
62
63- loop {
64- buffer.clear();
65- let len = stdin().read_line(&mut buffer)?;
66- if len == 0 {
67- break;
68- }
69+ stdin().read_line(&mut buffer)?;
70+ let message: Message = serde_json::from_str(&buffer)?;
71+
72+ if let MessageBody::Init {
73+ node_id, node_ids, ..
74+ } = message.body
75+ {
76+ let mut node: Node = Node::new(node_id, node_ids);
77+ node.handle_message(serde_json::from_str(&buffer)?)?;
78
79- let message: Message = match serde_json::from_str(&buffer) {
80- Ok(msg) => msg,
81- Err(e) => {
82- eprintln!("Error parsing message: {}", e);
83- continue;
84+ loop {
85+ buffer.clear();
86+ let len = stdin().read_line(&mut buffer)?;
87+ if len == 0 {
88+ break;
89 }
90- };
91
92- if let Err(e) = node.handle_message(message) {
93- eprintln!("Error handling message: {}", e);
94+ let message: Message = match serde_json::from_str(&buffer) {
95+ Ok(msg) => msg,
96+ Err(e) => {
97+ eprintln!("Error parsing message: {}", e);
98+ continue;
99+ }
100+ };
101+
102+ if let Err(e) = node.handle_message(message) {
103+ eprintln!("Error handling message: {}", e);
104+ }
105 }
106+ } else {
107+ panic!("Expected first messaage to be a init message");
108 }
109
110 Ok(())
Handling unique IDs
The next part of the challenge is to handle unique ID generation. For that, we will introduce the required message types to our Message struct, and store an internal counter in the Node. Also, we improve “responding” by sending out messages on stdout and logging our progress on stderr. The majority of the work is done in the handle_generate function as listed below:
1diff --git a/src/main.rs b/src/main.rs
2index 434d943..772c545 100644
3--- a/src/main.rs
4+++ b/src/main.rs
5@@ -1,14 +1,18 @@
6 use anyhow::Result;
7 use message::{Message, MessageBody};
8 use serde_json;
9-use std::io::stdin;
10+use std::{
11+ io::{stdin, Write},
12+ sync::atomic::AtomicUsize,
13+};
14
15 mod message;
16
17-#[derive(Debug, Default, Clone)]
18+#[derive(Debug)]
19 #[allow(unused)]
20 struct Node {
21 id: String,
22+ counter: AtomicUsize,
23 node_ids: Vec<String>,
24 current_msg_id: i32,
25 }
26@@ -16,6 +20,7 @@ struct Node {
27 impl Node {
28 fn new(id: String, node_ids: Vec<String>) -> Self {
29 Node {
30+ counter: AtomicUsize::new(0),
31 id,
32 node_ids,
33 current_msg_id: 0,
34@@ -54,16 +59,45 @@ impl Node {
35 in_reply_to: msg_id,
36 },
37 };
38- println!("{}", serde_json::to_string(&reply_message)?);
39- eprintln!("{}", serde_json::to_string(&reply_message)?);
40
41- Ok(())
42+ self.write_and_log_response(reply_message)
43+ }
44+
45+ fn handle_generate(&mut self, message: Message) -> Result<()> {
46+ // Generate a unique ID and send back to the receiver.
47+ // To generate a unique ID, we will take the instant value in self,
48+ // and just return the number of millis passed.
49+
50+ Ok(if let MessageBody::Generate { msg_id } = message.body {
51+ let reply_message = Message {
52+ src: self.id.clone(),
53+ dest: message.src,
54+ body: MessageBody::GenerateOk {
55+ id: format!(
56+ "{}-{}",
57+ self.id,
58+ self.counter
59+ .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
60+ ),
61+ msg_id: self.get_next_message_id(),
62+ in_reply_to: msg_id,
63+ },
64+ };
65+
66+ self.write_and_log_response(reply_message)?
67+ } else {
68+ ()
69+ })
70 }
71
72 fn handle_message(&mut self, message: Message) -> Result<()> {
73+ let converted = serde_json::to_string(&message)?;
74+ eprintln!("Received request: {}", converted);
75+
76 match message.body.clone() {
77 MessageBody::Echo { .. } => self.handle_echo(message),
78 MessageBody::Init { msg_id, .. } => self.handle_init(msg_id, message.src.clone()),
79+ MessageBody::Generate { .. } => self.handle_generate(message),
80
81 _ => {
82 eprintln!("Unhandled message type");
83@@ -71,6 +105,13 @@ impl Node {
84 }
85 }
86 }
87+
88+ fn write_and_log_response(&self, message: Message) -> Result<()> {
89+ let converted = serde_json::to_string(&message)?;
90+ println!("{}", converted);
91+ eprint!("Responding: {}", converted);
92+ Ok(std::io::stdout().flush()?)
93+ }
94 }
95
96 #[tokio::main]
97diff --git a/src/message.rs b/src/message.rs
98index 6c79550..01adec1 100644
99--- a/src/message.rs
100+++ b/src/message.rs
101@@ -20,6 +20,7 @@ pub enum MessageBody {
102 in_reply_to: i32,
103 echo: String,
104 },
105+
106 Init {
107 msg_id: i32,
108 node_id: String,
109@@ -28,4 +29,14 @@ pub enum MessageBody {
110 InitOk {
111 in_reply_to: i32,
112 },
113+
114+ Generate {
115+ msg_id: i32
116+ },
117+
118+ GenerateOk {
119+ id: String,
120+ msg_id: i32,
121+ in_reply_to: i32
122+ },
123 }
The fun begins!
Now the actual fun begins. We will implement methods to handle broadcast, and forward it to our peers.
1diff --git a/src/main.rs b/src/main.rs
2index 772c545..7b19426 100644
3--- a/src/main.rs
4+++ b/src/main.rs
5@@ -2,6 +2,7 @@ use anyhow::Result;
6 use message::{Message, MessageBody};
7 use serde_json;
8 use std::{
9+ collections::HashSet,
10 };
11@@ -14,6 +15,7 @@ struct Node {
12 id: String,
13+ values: HashSet<i32>,
14 current_msg_id: i32,
15 }
16
17@@ -24,6 +26,7 @@ impl Node {
18 id,
19+ values: HashSet::new(),
20 }
21 }
22
23+ fn handle_broadcast(&mut self, message: Message) -> Result<()> {
24+ if let MessageBody::Broadcast {
25+ msg_id,
26+ message: value,
27+ } = message.body
28+ {
29+ self.values.insert(value);
30+ let reply = Message {
31+ src: self.id.clone(),
32+ dest: message.src,
33+ body: MessageBody::BroadcastOk {
34+ in_reply_to: msg_id,
35+ msg_id: self.get_next_message_id(),
36+ },
37+ };
38+
39+ self.write_and_log_response(reply)
40+ } else {
41+ Ok(())
42+ }
43+ }
44+
45+ fn handle_read(&mut self, message: Message) -> Result<()> {
46+ if let MessageBody::Read { msg_id } = message.body {
47+ let messages: HashSet<i32> = self.values.clone();
48+ let reply = Message {
49+ src: self.id.clone(),
50+ dest: message.src,
51+ body: MessageBody::ReadOk {
52+ messages,
53+ in_reply_to: msg_id,
54+ },
55+ };
56+
57+ self.write_and_log_response(reply)
58+ } else {
59+ Ok(())
60+ }
61+ }
62+
63+ fn handle_topology(&mut self, message: Message) -> Result<()> {
64+ if let MessageBody::Topology { msg_id, .. } = message.body {
65+ let reply = Message {
66+ src: self.id.clone(),
67+ dest: message.src,
68+ body: MessageBody::TopologyOk {
69+ in_reply_to: msg_id,
70+ },
71+ };
72+
73+ self.write_and_log_response(reply)
74+ } else {
75+ Ok(())
76+ }
77+ }
78+
79 fn handle_message(&mut self, message: Message) -> Result<()> {
80 let converted = serde_json::to_string(&message)?;
81 eprintln!("Received request: {}", converted);
82@@ -98,6 +157,9 @@ impl Node {
83 MessageBody::Echo { .. } => self.handle_echo(message),
84+ MessageBody::Broadcast { .. } => self.handle_broadcast(message),
85+ MessageBody::Read { .. } => self.handle_read(message),
86+ MessageBody::Topology { .. } => self.handle_topology(message),
87
88 _ => {
89 eprintln!("Unhandled message type");
90diff --git a/src/message.rs b/src/message.rs
91index 01adec1..7dd52b7 100644
92--- a/src/message.rs
93+++ b/src/message.rs
94@@ -1,3 +1,5 @@
95+use std::collections::{HashMap, HashSet};
96+
97 #[derive(Debug, Serialize, Deserialize, Clone)]
98@@ -39,4 +41,32 @@ pub enum MessageBody {
99+ Broadcast {
100+ msg_id: i32,
101+ message: i32
102+ },
103+
104+ BroadcastOk {
105+ in_reply_to: i32,
106+ msg_id: i32
107+ },
108+
109+ Read {
110+ msg_id: i32,
111+ },
112+
113+ ReadOk {
114+ messages: HashSet<i32>,
115+ in_reply_to: i32
116+ },
117+
118+ Topology {
119+ msg_id: i32,
120+ topology: HashMap<String, Vec<String>>
121+ },
122+
123+ TopologyOk {
124+ in_reply_to: i32
125+ }
126 }
For this, we need to implement handlers for read, topology, and broadcast messages as listed in the outline. The logic itself is super simple; we store all the IDs we’ve seen in a HashSet, and return the hashset when read is invoked on our node.
Talking to other nodes
The next patch will implement a way to talk to other nodes, whatever is connected to ours.
1diff --git a/src/main.rs b/src/main.rs
2index 7b19426..0f955b8 100644
3--- a/src/main.rs
4+++ b/src/main.rs
5@@ -17,6 +17,8 @@ struct Node {
6 node_ids: Vec<String>,
7 values: HashSet<i32>,
8 current_msg_id: i32,
9+ neighbors: Vec<String>,
10+ downstream: Vec<String>,
11 }
12
13 impl Node {
14@@ -27,6 +29,8 @@ impl Node {
15 node_ids,
16 current_msg_id: 0,
17 values: HashSet::new(),
18+ neighbors: Vec::new(),
19+ downstream: Vec::new(),
20 }
21 }
22
23@@ -99,7 +103,7 @@ impl Node {
24 message: value,
25 } = message.body
26 {
27- self.values.insert(value);
28+ // First, respond that I got this value.
29 let reply = Message {
30 src: self.id.clone(),
31 dest: message.src,
32@@ -109,7 +113,65 @@ impl Node {
33 },
34 };
35
36- self.write_and_log_response(reply)
37+ self.write_and_log_response(reply)?;
38+
39+ let i_have_this = self.values.contains(&value);
40+ if !i_have_this {
41+ self.values.insert(value);
42+
43+ // Tell my friends about this too!. Write Gossip messages.
44+ let neighbors = self.neighbors.clone();
45+ for neighbor in neighbors {
46+ let new_msg_id = self.get_next_message_id();
47+
48+ let gossip = Message {
49+ src: self.id.clone(),
50+ dest: neighbor.clone(),
51+ body: MessageBody::Gossip {
52+ msg_id: new_msg_id,
53+ gossip_value: value,
54+ },
55+ };
56+
57+ self.write_and_log_response(gossip)?;
58+ }
59+ }
60+
61+ Ok(())
62+ } else {
63+ Ok(())
64+ }
65+ }
66+
67+ fn handle_gossip(&mut self, message: Message) -> Result<()> {
68+ if let MessageBody::Gossip {
69+ gossip_value: value,
70+ ..
71+ } = message.body
72+ {
73+ let i_have_this = self.values.contains(&value);
74+ if !i_have_this {
75+ self.values.insert(value);
76+
77+ // Tell my friends about this too!. Write Gossip messages.
78+ let neighbors = self.neighbors.clone();
79+ for neighbor in neighbors {
80+ let new_msg_id = self.get_next_message_id();
81+
82+ let gossip = Message {
83+ src: self.id.clone(),
84+ dest: neighbor.clone(),
85+ body: MessageBody::Gossip {
86+ msg_id: new_msg_id,
87+ gossip_value: value,
88+ },
89+ };
90+
91+ self.write_and_log_response(gossip)?;
92+ }
93+ }
94+
95+ Ok(())
96 } else {
97 Ok(())
98 }
99@@ -134,7 +196,21 @@ impl Node {
100 }
101
102 fn handle_topology(&mut self, message: Message) -> Result<()> {
103- if let MessageBody::Topology { msg_id, .. } = message.body {
104+ if let MessageBody::Topology {
105+ msg_id,
106+ topology: graph,
107+ } = message.body
108+ {
109+ self.neighbors = graph
110+ .get(&self.id)
111+ .expect("No neighbors defined for me? Empty??")
112+ .clone();
113+
114+ // Something fancy. Do some graph computation to find downstream nodes? Nah.
115+ // Maybe in 3.d.
116+
117+ self.downstream.extend(self.neighbors.clone());
118+
119 let reply = Message {
120 src: self.id.clone(),
121 dest: message.src,
122@@ -160,6 +236,7 @@ impl Node {
123 MessageBody::Broadcast { .. } => self.handle_broadcast(message),
124 MessageBody::Read { .. } => self.handle_read(message),
125 MessageBody::Topology { .. } => self.handle_topology(message),
126+ MessageBody::Gossip { .. } => self.handle_gossip(message),
127
128 _ => {
129 eprintln!("Unhandled message type");
130diff --git a/src/message.rs b/src/message.rs
131index 7dd52b7..1748ee8 100644
132--- a/src/message.rs
133+++ b/src/message.rs
134@@ -68,5 +68,12 @@ pub enum MessageBody {
135
136 TopologyOk {
137 in_reply_to: i32
138- }
139+ },
140+
141+ // Custom message types:
142+
143+ Gossip {
144+ msg_id: i32,
145+ gossip_value: i32
146+ },
147 }
The logic is simple:
- We
Gossipour message to our peers when we receive a gossip message. Before doing that, we acknowledge the gossip. - The
GossipAckmessage is sent to the sender to make sure they don’t re-send the gossip again (this comes in a later part).
If we do not respond before updating our state, we will fall into an infinite loop where we forward the message, receive it, and keep forwarding it. So, we first ack the message, and then forward it to the neighbors.
Sending message with Retries
The current method implements ways to send message, but we do not retry if it fails. Since we don’t have the privilege of using SyncRPC that is in the go library, we need to implement our own. The go library uses Context with Cancel, and something similar that can be implemented in Rust is a oneshot_channel. For every message we expect to receive a reply to, we store a oneshot channel, which is keyed on the message id. If we receive any message whose in_reply_to field has that particular message id, we pop the channel, and cancel our timeout.
1diff --git a/src/main.rs b/src/main.rs
2index 0f955b8..b61487b 100644
3--- a/src/main.rs
4+++ b/src/main.rs
5@@ -2,10 +2,12 @@ use anyhow::Result;
6 use message::{Message, MessageBody};
7 use serde_json;
8 use std::{
9- collections::HashSet,
10+ collections::{HashMap, HashSet},
11 io::{stdin, Write},
12 sync::atomic::AtomicUsize,
13+ time::Duration,
14 };
15+use tokio::{runtime::Runtime, time};
16
17 mod message;
18
19@@ -13,24 +15,28 @@ mod message;
20 #[allow(unused)]
21 struct Node {
22 id: String,
23+ runtime: Runtime,
24 counter: AtomicUsize,
25 node_ids: Vec<String>,
26 values: HashSet<i32>,
27 current_msg_id: i32,
28 neighbors: Vec<String>,
29 downstream: Vec<String>,
30+ ack_channels: HashMap<i32, tokio::sync::oneshot::Sender<i32>>,
31 }
32
33 impl Node {
34 fn new(id: String, node_ids: Vec<String>) -> Self {
35 Node {
36 counter: AtomicUsize::new(0),
37+ runtime: Runtime::new().unwrap(),
38 id,
39 node_ids,
40 current_msg_id: 0,
41 values: HashSet::new(),
42 neighbors: Vec::new(),
43 downstream: Vec::new(),
44+ ack_channels: HashMap::new(),
45 }
46 }
47
48@@ -51,8 +57,8 @@ impl Node {
49 echo,
50 },
51 };
52- println!("{}", serde_json::to_string(&reply_message)?);
53- eprintln!("{}", serde_json::to_string(&reply_message)?);
54+
55+ self.write_and_log_response(reply_message)?;
56 }
57 Ok(())
58 }
59@@ -238,6 +244,19 @@ impl Node {
60 MessageBody::Topology { .. } => self.handle_topology(message),
61 MessageBody::Gossip { .. } => self.handle_gossip(message),
62
63+ MessageBody::GossipAck {
64+ in_reply_to,
65+ msg_id,
66+ } => {
67+ // Check that we have the channel to send the ack.
68+ if let Some(sender) = self.ack_channels.remove(&in_reply_to) {
69+ Ok(sender.send(msg_id).unwrap())
70+ } else {
71+ eprintln!("No channel found to send ack");
72+ Ok(())
73+ }
74+ }
75+
76 _ => {
77 eprintln!("Unhandled message type");
78 Ok(())
79@@ -245,11 +264,39 @@ impl Node {
80 }
81 }
82
83- fn write_and_log_response(&self, message: Message) -> Result<()> {
84- let converted = serde_json::to_string(&message)?;
85- println!("{}", converted);
86- eprint!("Responding: {}", converted);
87- Ok(std::io::stdout().flush()?)
88+ fn write_and_log_response(&mut self, message: Message) -> Result<()> {
89+ if let MessageBody::Gossip { msg_id, .. } = &message.body {
90+ let (sender, mut receiver) = tokio::sync::oneshot::channel::<i32>();
91+ self.ack_channels.insert(*msg_id, sender);
92+
93+ // Try to send the message to the receiver, wait for a certain time, and send the message
94+ // again if we do not get the response back. Do all this within a tokio runtime.
95+ let fut = async move {
96+ let converted = serde_json::to_string(&message).unwrap();
97+ println!("{}", converted);
98+ let mut ticker = time::interval(Duration::from_millis(100));
99+
100+ loop {
101+ tokio::select! {
102+ _ = ticker.tick() => {
103+ // Resend the message; we have no acks yet.
104+ println!("{}", converted.clone());
105+ }
106+ _ = &mut receiver => {
107+ // We got an ack! Break the loop.
108+ break
109+ }
110+ }
111+ }
112+ };
113+
114+ self.runtime.spawn(fut);
115+ Ok(())
116+ } else {
117+ println!("{}", serde_json::to_string(&message)?);
118+ eprint!("Responding: {}", serde_json::to_string(&message)?);
119+ Ok(std::io::stdout().flush()?)
120+ }
121 }
122 }
123
124diff --git a/src/message.rs b/src/message.rs
125index 1748ee8..d331470 100644
126--- a/src/message.rs
127+++ b/src/message.rs
128@@ -76,4 +76,9 @@ pub enum MessageBody {
129 msg_id: i32,
130 gossip_value: i32
131 },
132+
133+ GossipAck {
134+ in_reply_to: i32,
135+ msg_id: i32
136+ }
137 }
In order to implement this, we create a Tokio runtime, and push futures that are polling the stdin, looking for messages we have replies to, or waiting on a timer, which fires every 100ms. Either the message reply arrives, or the timer expires, when we resend the message. Since we are only looking for replies (confirmation) to the Gossip messages, we match the message type on message send. The future is created only if the message type matches Gossip message. The modification to handling gossip is simple enough:
1diff --git a/src/main.rs b/src/main.rs
2index b61487b..ab1b253 100644
3--- a/src/main.rs
4+++ b/src/main.rs
5@@ -152,9 +152,21 @@ impl Node {
6 fn handle_gossip(&mut self, message: Message) -> Result<()> {
7 if let MessageBody::Gossip {
8 gossip_value: value,
9- ..
10+ msg_id
11 } = message.body
12 {
13+ // First, respond that I got this value.
14+ let reply = Message {
15+ src: self.id.clone(),
16+ dest: message.src,
17+ body: MessageBody::GossipAck {
18+ in_reply_to: msg_id,
19+ msg_id: self.get_next_message_id(),
20+ },
21+ };
22+
23+ self.write_and_log_response(reply)?;
24+
25 let i_have_this = self.values.contains(&value);
26 if !i_have_this {
27 self.values.insert(value);
Latency constraints.
This is the part I had the most fun with while completing the first part of this challenge. You are supposed to send messages with 400ms latency medians (600ms max latency). The constraints are:
- 30 message-per-op
- 400ms median latency
- 600ms max latency
given that every message has 100ms latency within the system. This means, we need to optimize our graph in some way. In a overkill fashion, I decided to write a genetic algorithm to generate the graph.
Genetic Algorithm for graph generation.
The graph will be encoded in the following way:
1import random
2from typing import Dict, List, Self
3import networkx as nx
4import matplotlib.pyplot as plt
5
6NUM_NODES_FIXED = 25
7POPULATION_SIZE = 100
8MAX_ITERATIONS = 500
9MAX_PERMISSIBLE_PATH_LENGTH = 6
10
11class Graph(object):
12 def __init__(self, num_nodes = 25, max_permissible_path_length = 0) -> None:
13 self.vertices = range(0, num_nodes)
14 self.max_permissible_path_length = max_permissible_path_length
15 self.edges = {}
16
17 def add_edge(self, from_vertex: int, to_vertex: int) -> None:
18 if (from_vertex) not in self.vertices or to_vertex not in self.vertices:
19 return
20
21 self.edges[from_vertex] = self.edges.get(from_vertex, []) or []
22 self.edges[to_vertex] = self.edges.get(to_vertex, []) or []
23
24 if to_vertex not in self.edges[from_vertex]:
25 self.edges[from_vertex].append(to_vertex)
26 self.edges[to_vertex].append(from_vertex)
27
28 def get_distances(self, from_vertex: int) -> Dict:
29 distances = {}
30 visited = set()
31
32 queue = [(from_vertex, 0)]
33 distances[from_vertex] = 0
34 visited.add(from_vertex)
35
36 while queue:
37 first, depth = queue.pop(0)
38 neighbors = self.edges.get(first, [])
39 for neighbor in neighbors:
40 if neighbor not in visited:
41 visited.add(neighbor)
42 queue.append((neighbor, depth + 1))
43 distances[neighbor] = depth + 1
44
45 return distances
46
47 def get_nodes_count(self) -> int:
48 return len(self.vertices)
49
50 def check_average_propagation_cost(self) -> float:
51 score = 0
52
53 disconnected_nodes = 0
54 nodes_with_maxpath_violations = 0
55
56 max_edges = max ([ len(self.edges.get(node, [])) for node in self.vertices ])
57 total_edges = sum([ len(self.edges.get(node, [])) for node in self.vertices ])
58
59 for node in self.vertices:
60 distances = self.get_distances(node)
61 disconnected_nodes += len(self.vertices) - len(distances) # Calculate the number of nodes unreachable from this
62 nodes_with_maxpath_violations += len([tonode for tonode in distances if distances.get(tonode) > MAX_PERMISSIBLE_PATH_LENGTH ])
63
64 score += -100000 * disconnected_nodes # All nodes MUST be connected
65 score += -100000 * nodes_with_maxpath_violations
66 score += -2000 * max_edges
67 score += -250 * total_edges
68
69 return scoreThe functions are pretty self-explanatory:
get_nodes_countreturns the number of nodes in the graph.check_average_propagation_costreturns the score of the graph. The score is calculated as:- 100000 * disconnected_nodes: All nodes must be connected.
- 100000 * nodes_with_maxpath_violations: No node should have a path length greater than 6.
- 2000 * max_edges: The maximum number of edges in the graph.
- 250 * total_edges: The total number of edges in the graph.
get_distancesreturns the distances from a node to all other nodes in the graph.add_edgeadds an edge between two nodes. The edges are bidirectional.__init__initializes the graph.
Using this, and implementing a simple genetic algorithm, we land with a graph that looks like this:
1"n17" => &["n19", "n0"],
2"n19" => &["n17", "n14"],
3"n10" => &["n2", "n22", "n20"],
4"n2" => &["n10", "n13", "n11"],
5"n13" => &["n2", "n15"],
6"n7" => &["n14", "n3"],
7"n14" => &["n7", "n19", "n12"],
8"n24" => &["n22", "n1", "n23"],
9"n22" => &["n24", "n10", "n8"],
10"n18" => &["n5", "n15"],
11"n5" => &["n18", "n0", "n3"],
12"n0" => &["n4", "n17", "n5"],
13"n4" => &["n0", "n8"],
14"n1" => &["n24", "n6", "n11"],
15"n23" => &["n24", "n6", "n16"],
16"n21" => &["n20", "n12", "n9"],
17"n20" => &["n21", "n9", "n10"],
18"n15" => &["n13", "n12", "n18"],
19"n6" => &["n1", "n23", "n3"],
20"n12" => &["n21", "n14", "n15"],
21"n9" => &["n21", "n20"],
22"n11" => &["n2", "n1"],
23"n3" => &["n7", "n6", "n5"],
24"n8" => &["n4", "n16", "n22"],
25"n16" => &["n23", "n8"],Instead of using the given topology, this is the topology we can use. With this, we get the following results on median:
1latencies:
20.5=434, 1:599
3
4msg-per-op:
529The genetic algorithm is as follows:
1class Chromosome:
2 def __init__(self) -> None:
3 self.graph = Graph(num_nodes=NUM_NODES_FIXED, max_permissible_path_length=MAX_PERMISSIBLE_PATH_LENGTH)
4 # Initialize some random connections
5 for _ in range(random.randint(10, 50)):
6 from_node = random.randint(0, NUM_NODES_FIXED - 1)
7 to_node = random.randint(0, NUM_NODES_FIXED - 1)
8 if from_node != to_node:
9 self.graph.add_edge(from_node, to_node)
10
11 def mutate(self) -> Self:
12 """
13 Mutation strategies:
14 1. Add a random edge
15 2. Remove a random edge
16 3. Rewire an existing edge
17 """
18 mutation_type = random.choice(['add', 'remove', 'rewire'])
19
20 if mutation_type == 'add':
21 from_node = random.randint(0, NUM_NODES_FIXED - 1)
22 to_node = random.randint(0, NUM_NODES_FIXED - 1)
23 if from_node != to_node and to_node not in self.graph.edges.get(from_node, []):
24 self.graph.add_edge(from_node, to_node)
25
26 elif mutation_type == 'remove':
27 if self.graph.edges:
28 from_node = random.choice(list(self.graph.edges.keys()))
29 to_node = random.choice(self.graph.edges[from_node])
30 self.graph.edges[from_node].remove(to_node)
31 self.graph.edges[to_node].remove(from_node)
32
33 if not self.graph.edges[from_node]:
34 del self.graph.edges[from_node]
35 if not self.graph.edges[to_node]:
36 del self.graph.edges[to_node]
37
38 elif mutation_type == 'rewire':
39 if self.graph.edges:
40 from_node = random.choice(list(self.graph.edges.keys()))
41 old_to_node = random.choice(self.graph.edges[from_node])
42 new_to_node = random.randint(0, NUM_NODES_FIXED - 1)
43
44 self.graph.edges[from_node].remove(old_to_node)
45 self.graph.edges[old_to_node].remove(from_node)
46
47 if new_to_node != from_node and new_to_node not in self.graph.edges.get(from_node, []):
48 self.graph.add_edge(from_node, new_to_node)
49
50 return self
51
52 def crossover(self, other: "Chromosome") -> List["Chromosome"]:
53 """
54 Crossover strategies:
55 1. Randomly combine edges from both parent chromosomes
56 2. Create two offspring with mixed edge sets
57 """
58 offspring1 = Chromosome()
59 offspring2 = Chromosome()
60
61 all_edges_parent1 = set()
62 all_edges_parent2 = set()
63
64 for from_node, to_nodes in self.graph.edges.items():
65 for to_node in to_nodes:
66 all_edges_parent1.add((from_node, to_node))
67
68 for from_node, to_nodes in other.graph.edges.items():
69 for to_node in to_nodes:
70 all_edges_parent2.add((from_node, to_node))
71
72 combined_edges = list(all_edges_parent1.union(all_edges_parent2))
73 random.shuffle(combined_edges)
74
75 offspring1_edges = combined_edges[:len(combined_edges)//2]
76 offspring2_edges = combined_edges[len(combined_edges)//2:]
77
78 offspring1.graph.edges = {}
79 offspring2.graph.edges = {}
80
81 for from_node, to_node in offspring1_edges:
82 offspring1.graph.add_edge(from_node, to_node)
83
84 for from_node, to_node in offspring2_edges:
85 offspring2.graph.add_edge(from_node, to_node)
86
87 return [offspring1, offspring2]
88
89def genetic_algorithm():
90 # Initialize population
91 population = [Chromosome() for _ in range(POPULATION_SIZE)]
92
93 best_fitness_history = []
94
95 # Genetic Algorithm main loop
96 for generation in range(MAX_ITERATIONS):
97 # Evaluate fitness for each chromosome
98 fitness_scores = [chromosome.graph.check_average_propagation_cost() for chromosome in population]
99
100 # Track best fitness in this generation
101 best_fitness = max(fitness_scores)
102 best_fitness_history.append(best_fitness)
103
104 print(f"Generation {generation}: Best Fitness = {best_fitness}")
105
106 # Select top 20% chromosomes for reproduction
107 sorted_population = sorted(zip(population, fitness_scores), key=lambda x: x[1], reverse=True)
108 top_population = [chrom for chrom, _ in sorted_population[:POPULATION_SIZE//5]]
109
110 # Create new population through crossover and mutation
111 new_population = top_population.copy()
112
113 while len(new_population) < POPULATION_SIZE:
114 # Select parents for crossover
115 parent1 = random.choice(top_population)
116 parent2 = random.choice(top_population)
117
118 # Perform crossover
119 offspring = parent1.crossover(parent2)
120
121 # Optional mutation
122 for child in offspring:
123 if random.random() < 0.8: # 30% mutation rate
124 child.mutate()
125
126 new_population.extend(offspring)
127
128 # Truncate population to original size
129 population = new_population[:POPULATION_SIZE]
130
131 # Return the best chromosome
132 best_chromosome = max(population, key=lambda x: x.graph.check_average_propagation_cost())
133 return best_chromosomeI tried with starting out with a fully-connected graph and removing edges, but it did not work out as good as this implementation (not sure why).
The next (and final) part of the broadcast challenge is to make it more efficient. For this, we just need to batch out our messages; instead of sending one message at a time, we send n messages at a time. The message sending can work on a ticker, where we send deltas of messages every 100ms.
Conclusion
This challenge was a lot of fun; esp. the genetic algorithm part. I had a lot of fun implementing it. The final part of the challenge was to batch out messages, which I did not include along with code for the last part in this tutorial for brevity. I will leave that as an exercise for the reader. The next challenges are counter-based challenges, and I plan to do them in go, like the way the engineers at fly.io intended. The code for this part can be found here .