diff --git a/src/db/loki_kv/control.rs b/src/db/loki_kv/control.rs index f5d5c5f..90ed5c7 100644 --- a/src/db/loki_kv/control.rs +++ b/src/db/loki_kv/control.rs @@ -14,6 +14,8 @@ pub struct ControlFile { last_checkpoint_id: u64, checkpoint_directory_path: String, wal_directory_path: String, + current_leader_value: Option, + self_identifier: Option, } impl ControlFile { @@ -23,12 +25,39 @@ impl ControlFile { pub fn get_next_timeline_id(&self) -> u64 { self.last_wal_timeline + 1 } + pub fn get_wal_directory_path(&self) -> &str { &self.wal_directory_path } + pub fn get_checkpoint_directory_path(&self) -> &str { &self.checkpoint_directory_path } + + pub fn get_current_leader_identifier(&self) -> Option { + self.current_leader_value.clone() + } + + pub fn get_self_identifier(&self) -> Option { + self.self_identifier.clone() + } + + pub fn set_current_leader_identifier(&mut self, current_leader_value: u64) { + self.current_leader_value = Some(current_leader_value.clone()); + } + + pub fn set_self_identifier(&mut self, id: u64) { + self.self_identifier = Some(id.clone()); + } + + pub fn is_leader(&self) -> bool { + if self.self_identifier.is_some() && self.current_leader_value.is_some() { + if self.self_identifier.unwrap() == self.current_leader_value.unwrap() { + return true; + } + } + return false; + } pub fn write( path: String, last_wal_timeline: u64, @@ -53,6 +82,8 @@ impl ControlFile { last_checkpoint_id, checkpoint_directory_path, wal_directory_path, + current_leader_value: None, + self_identifier: Some(1 as u64), }; // Take lock on control file diff --git a/src/db/server_multithread/mod.rs b/src/db/server_multithread/mod.rs index 74f47ad..9556874 100644 --- a/src/db/server_multithread/mod.rs +++ b/src/db/server_multithread/mod.rs @@ -1 +1,2 @@ +pub mod paxos; pub mod server; diff --git a/src/db/server_multithread/paxos.rs b/src/db/server_multithread/paxos.rs new file mode 100644 index 0000000..62bae88 --- /dev/null +++ b/src/db/server_multithread/paxos.rs @@ -0,0 +1,52 @@ +use std::net::SocketAddr; + +use tokio::net::UdpSocket; + +use crate::loki_kv::control::ControlFile; + +pub struct ServiceManager { + udp_socket: UdpSocket, + BROADCAST_ADDRESS: SocketAddr, +} + +impl ServiceManager { + pub fn new() -> Self { + let listen_addr: SocketAddr = "0.0.0.0:8080".parse().unwrap(); + let std_socket = std::net::UdpSocket::bind(listen_addr).unwrap(); + + std_socket.set_broadcast(true); + + ServiceManager { + udp_socket: UdpSocket::from_std(std_socket).unwrap(), + BROADCAST_ADDRESS: "255.255.255.255:8080".parse().unwrap(), + } + } + + pub async fn broadcast_message(&self, msg: &str) -> Result<(), String> { + self.udp_socket.send_to(msg.as_bytes(), self.BROADCAST_ADDRESS).await.unwrap(); + Ok(()) + } +} + +pub struct PaxosNode { + ctrl_file: ControlFile, + service_manager: ServiceManager, +} + +impl PaxosNode { + pub 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!") + } + } + + pub fn accept() {} + pub fn learn() {} +}