fix: use multicast instead of broadcast

Dockerfile

@@ -0,0 +1,37 @@
+# Stage 1: Build the application
+FROM rust:1-slim as builder
+
+# Install cargo-chef for dependency caching
+RUN apt-get update && apt-get install -y protobuf-compiler libssl-dev pkg-config
+RUN cargo install cargo-chef 
+
+# Set the working directory
+WORKDIR /app
+
+# Copy the dependency manifests
+COPY Cargo.toml Cargo.lock ./
+COPY control.toml .
+COPY . .
+
+# Build the release binary
+RUN cargo build --release --bin server-db
+
+# Stage 2: Create the runtime image
+FROM gcr.io/distroless/cc-debian12
+
+WORKDIR /app
+
+# Copy the binary from the builder stage
+COPY --from=builder /app/target/release/server-db /usr/local/bin/server-db
+
+# Copy the control file from the builder stage
+COPY --from=builder /app/control.toml /app/control.toml
+
+# Set the environment variable for the control file path
+ENV CONTROL_FILE_PATH="/app/control.toml"
+
+# Create and set up a volume for data persistence
+VOLUME /app/data
+
+# Set the command to run the server
+CMD ["/usr/local/bin/server-db"]

control.toml

@@ -0,0 +1,13 @@
+host = "localhost"
+port = 8766
+last_wal_timeline = 0
+last_checkpoint_id = 0
+checkpoint_directory_path = "/app/data/checkpoints"
+wal_directory_path = "/app/data/wal"
+current_leader_value = 12
+self_identifier = 12
+send_addr = "0.0.0.0:8080"
+consume_addr = "0.0.0.0:8081"
+checkpoint_timer_interval = 1
+paxos_timer_interval = 1
+gossip_timeout = 200

docker-compose.yml

@@ -0,0 +1,35 @@
+version: '3.8'
+services:
+  lokikv-1:
+    build: .
+    container_name: lokikv-1
+    ports:
+      - "7878:7878"
+    networks:
+      - lokikv-net
+    volumes:
+      - ./control.toml:/app/control.toml
+      - lokikv-data-1:/app/data
+    environment:
+      - CONTROL_FILE_PATH=/app/control.toml
+
+  lokikv-2:
+    build: .
+    container_name: lokikv-2
+    ports:
+      - "7879:7878"
+    networks:
+      - lokikv-net
+    volumes:
+      - ./control.toml:/app/control.toml
+      - lokikv-data-2:/app/data
+    environment:
+      - CONTROL_FILE_PATH=/app/control.toml
+
+networks:
+  lokikv-net:
+    driver: bridge
+
+volumes:
+  lokikv-data-1:
+  lokikv-data-2:

src/db/loki_kv/control.rs

@@ -22,14 +22,14 @@ pub struct ControlFile {
     consume_addr: String,
     checkpoint_timer_interval: Option<u64>,
     paxos_timer_interval: Option<u64>,
-    gossip_timeout: Option<u64>
+    gossip_timeout: Option<u64>,
 }
 
 impl ControlFile {
-    pub fn get_hostname(&self) -> String{
+    pub fn get_hostname(&self) -> String {
         return self.host.clone();
     }
-    pub fn get_port(&self) -> u16{
+    pub fn get_port(&self) -> u16 {
         return self.port;
     }
     pub fn get_next_checkpoint_id(&self) -> u64 {
@@ -39,11 +39,11 @@ impl ControlFile {
         self.last_wal_timeline + 1
     }
 
-    pub fn get_send_addr(&self) -> &str{
+    pub fn get_send_addr(&self) -> &str {
         &self.send_addr
     }
 
-    pub fn get_consume_addr(&self) -> &str{
+    pub fn get_consume_addr(&self) -> &str {
         &self.consume_addr
     }
 
@@ -64,23 +64,23 @@ impl ControlFile {
     }
 
     pub fn get_checkpoint_timer_interval(&self) -> u64 {
-        match self.checkpoint_timer_interval{
+        match self.checkpoint_timer_interval {
             Some(val) => return val,
-            None => return 1
+            None => return 1,
         }
     }
 
     pub fn get_paxos_timer_interval(&self) -> u64 {
-        match self.paxos_timer_interval{
+        match self.paxos_timer_interval {
             Some(val) => return val,
-            None => return 5
+            None => return 5,
         }
     }
 
-    pub fn get_gossip_timeout(&self) -> u64{
-        match self.gossip_timeout{
+    pub fn get_gossip_timeout(&self) -> u64 {
+        match self.gossip_timeout {
             Some(val) => return val,
-            None => return 300
+            None => return 300,
         }
     }
 
@@ -112,7 +112,7 @@ impl ControlFile {
         consume_addr: Option<String>,
         checkpoint_timer_interval: Option<u64>,
         paxos_timer_interval: Option<u64>,
-        gossip_timeout: Option<u64>
+        gossip_timeout: Option<u64>,
     ) -> Result<ControlFile, String> {
         // Create the WAL and checkpoint directories
         let wal_dir = Path::new(&wal_directory_path);
@@ -129,15 +129,19 @@ impl ControlFile {
         let final_send_addr: String = match send_addr {
             Some(addr) => addr,
             None => {
-                info_string("no listening address provided.. defaulting to 0.0.0.0:8080".to_string());
+                info_string(
+                    "no listening address provided.. defaulting to 0.0.0.0:8080".to_string(),
+                );
                 "0.0.0.0:8080".to_string()
             }
         };
 
-        let final_consume_addr: String = match consume_addr{
+        let final_consume_addr: String = match consume_addr {
             Some(addr) => addr,
             None => {
-                info_string("no listening address provided.. defaulting to 0.0.0.0:8081".to_string());
+                info_string(
+                    "no listening address provided.. defaulting to 0.0.0.0:8081".to_string(),
+                );
                 "0.0.0.0:8081".to_string()
             }
         };
@@ -155,7 +159,7 @@ impl ControlFile {
             consume_addr: final_consume_addr,
             checkpoint_timer_interval,
             paxos_timer_interval,
-            gossip_timeout
+            gossip_timeout,
         };
 
         // Take lock on control file

src/db/loki_kv/wal.rs

@@ -65,7 +65,7 @@ impl WALManager {
             Some("0.0.0.0:8081".to_string()),
             None,
             None,
-            None
+            None,
         )
         .unwrap();
         let timeline = control_file.get_next_timeline_id();

src/db/server_multithread/paxos.rs

@@ -1,18 +1,25 @@
-use std::{collections::{HashMap, HashSet}, io::Split, net::SocketAddr};
 use local_ip_address::local_ip;
+use std::{
+    collections::{HashMap, HashSet},
+    io::Split,
+    net::{Ipv4Addr, SocketAddr},
+};
 
-use tokio::{net::UdpSocket, time::timeout};
 use tokio::time::Duration;
+use tokio::{net::UdpSocket, time::timeout};
 
-use crate::{loki_kv::{control::ControlFile, loki_kv::get_control_file_path}, utils::{info_string, warning, warning_string}};
+use crate::{
+    loki_kv::{control::ControlFile, loki_kv::get_control_file_path},
+    utils::{info_string, warning, warning_string},
+};
 
 // ---------------------------- SERVICE MANAGER -------------------------------------------
 
 pub struct ServiceManager {
     udp_socket_send: UdpSocket,
     udp_socket_recv: UdpSocket,
     node_directory: HashSet<(String, String)>, // Hashset of node_id, address(ip + port)
-    BROADCAST_ADDRESS: SocketAddr,
+    multicast_addr: SocketAddr,
 }
 
 impl ServiceManager {
@@ -26,34 +33,38 @@ impl ServiceManager {
 
         let mut node_directory: HashSet<(String, String)> = HashSet::new();
 
-        std_socket.set_broadcast(true);
+        let recv_socket = UdpSocket::from_std(std_consumer_socket).unwrap();
+        _ = recv_socket.join_multicast_v4(Ipv4Addr::new(239, 1, 1, 1), Ipv4Addr::UNSPECIFIED);
 
         ServiceManager {
             udp_socket_send: UdpSocket::from_std(std_socket).unwrap(),
-            udp_socket_recv: UdpSocket::from_std(std_consumer_socket).unwrap(),
+            udp_socket_recv: recv_socket,
             node_directory: node_directory,
-            BROADCAST_ADDRESS: "255.255.255.255:8080".parse().unwrap(),
+            multicast_addr: format!("239.1.1.1:{}", consume_addr.port())
+                .parse()
+                .unwrap(),
         }
     }
 
     pub async fn broadcast_message(&self, msg: &str) -> Result<(), String> {
-        self.udp_socket_send.send_to(msg.as_bytes(), self.BROADCAST_ADDRESS).await.unwrap();
+        self.udp_socket_send
+            .send_to(msg.as_bytes(), self.multicast_addr)
+            .await
+            .unwrap();
         Ok(())
     }
 
     pub async fn start_consumption(&self) -> Result<(), ()> {
-        loop{
+        loop {
             // TODO: Add consumption logic
             // Somehitng like a go-routine treatment here?
             let mut msg_bytes: Vec<u8> = vec![];
-            self.udp_socket_recv.recv_from(&mut msg_bytes);
+            _ = self.udp_socket_recv.recv_from(&mut msg_bytes);
 
-            tokio::spawn(
-                async move {
-                    // Log message
-                    info_string(format!("Recieved the following message: {:?}", msg_bytes));
-                }
-            );
+            tokio::spawn(async move {
+                // Log message
+                info_string(format!("Recieved the following message: {:?}", msg_bytes));
+            });
 
             break;
         }
@@ -67,10 +78,14 @@ impl ServiceManager {
 
 // ---------------- PAXOS NODE ----------------------------
 
-fn get_ip_addr(addr: String) -> String{
+fn get_ip_addr(addr: String) -> String {
     let my_local_ip = local_ip().unwrap();
     let mut tks = addr.split(":");
-    let ip = format!("{}:{}", my_local_ip.to_string(), tks.nth(1).unwrap().to_string());
+    let ip = format!(
+        "{}:{}",
+        my_local_ip.to_string(),
+        tks.nth(1).unwrap().to_string()
+    );
     return ip;
 }
 
@@ -80,20 +95,23 @@ pub struct PaxosNode {
 }
 
 impl PaxosNode {
-    pub fn new_node() -> Self{
+    pub fn new_node() -> Self {
         let control_file = ControlFile::read_from_file_path(get_control_file_path()).unwrap();
-        PaxosNode { ctrl_file: control_file, service_manager:  ServiceManager::new()}
+        PaxosNode {
+            ctrl_file: control_file,
+            service_manager: ServiceManager::new(),
+        }
     }
     pub async fn propose(&self) {
         let value = self.ctrl_file.get_self_identifier();
 
         // Broadcast to network
-        match value{
-           Some(val) => {
-               let msg = format!("PROPOSE {}", val);
-               self.service_manager.broadcast_message(msg.as_str());
-           },
-           None => panic!("No value to broadcast!")
+        match value {
+            Some(val) => {
+                let msg = format!("PROPOSE {}", val);
+                self.service_manager.broadcast_message(msg.as_str());
+            }
+            None => panic!("No value to broadcast!"),
         }
     }
 
@@ -103,38 +121,51 @@ impl PaxosNode {
     // Gossip for node discovery
     pub async fn gossip(&self) -> Result<(), String> {
         let node_id = self.ctrl_file.get_self_identifier().unwrap();
-        let data = format!("{}~{}", node_id.to_string(), get_ip_addr(self.ctrl_file.get_consume_addr().to_string()));
+        let data = format!(
+            "{}~{}",
+            node_id.to_string(),
+            get_ip_addr(self.ctrl_file.get_consume_addr().to_string())
+        );
         info_string(format!("Sending -> {}", data));
         let result = self.service_manager.broadcast_message(data.as_str()).await;
         return result;
     }
 
     pub async fn gossip_consume(&mut self) {
         let MAX_GOSSIP_CONSUMPTION = 10;
-        for i in 0..MAX_GOSSIP_CONSUMPTION{
+        for i in 0..MAX_GOSSIP_CONSUMPTION {
             info_string(format!("{} gossip trial", i));
             // Somehitng like a go-routine treatment here?
-            let mut msg_bytes: Vec<u8> = vec![];
-            match timeout(Duration::from_secs(self.ctrl_file.get_gossip_timeout()), self.service_manager.udp_socket_recv.recv_from(&mut msg_bytes)).await{
+            let mut msg_bytes: Vec<u8> = vec![0u8; 2048];
+            match timeout(
+                Duration::from_secs(self.ctrl_file.get_gossip_timeout()),
+                self.service_manager
+                    .udp_socket_recv
+                    .recv_from(&mut msg_bytes),
+            )
+            .await
+            {
                 Ok(Ok((_, _))) => {
                     let data = String::from_utf8(msg_bytes).unwrap();
                     let mut tokens;
-                    if data.contains("~"){
+                    if data.contains("~") {
                         tokens = data.as_str().split("~");
-                    }else{
+                    } else {
                         let msg = format!("Token does not contain any ~ {:?}, skipping..", data);
                         warning(msg.as_str());
                         continue;
                     }
 
                     // Log message
                     info_string(format!("Recieved the following message: {:?}", tokens));
-                    self.service_manager.update_node_directory(tokens.nth(0).unwrap().to_string(), tokens.nth(1).unwrap().to_string());
-                },
+                    self.service_manager.update_node_directory(
+                        tokens.nth(0).unwrap().to_string(),
+                        tokens.nth(1).unwrap().to_string(),
+                    );
+                }
                 Ok(Err(e)) => panic!("{}", e),
-                Err(e) => panic!("{}", e)
+                Err(e) => panic!("{}", e),
             };
-
         }
     }
 }