refactor: rename io to queue (#60)

Author: toidiu

src/lib.rs

@@ -1,11 +1,11 @@
 #![allow(dead_code)]
 
 mod error;
-mod io;
 mod log;
 mod macros;
 mod mode;
 mod packet;
+mod queue;
 mod raft_state;
 mod server;
 mod state_machine;

src/mode.rs

@@ -31,10 +31,10 @@
 //! https://textik.com/#8dbf6540e0dd1676
 
 use crate::{
-    io::ServerEgress,
     macros::cast_unsafe,
     mode::{candidate::Candidate, follower::Follower, leader::Leader},
     packet::Rpc,
+    queue::ServerEgress,
     raft_state::RaftState,
     server::{PeerId, ServerId},
 };
@@ -213,8 +213,8 @@ pub enum ElectionResult {
 mod tests {
     use super::*;
     use crate::{
-        io::testing::{helper_inspect_one_sent_packet, MockIo},
         log::{Idx, Term, TermIdx},
+        queue::testing::{helper_inspect_one_sent_packet, MockIo},
         server::PeerId,
         timeout::Timeout,
     };

src/mode/candidate.rs

@@ -1,7 +1,7 @@
 use crate::{
-    io::ServerEgress,
     mode::{cast_unsafe, ElectionResult, Mode, ModeTransition},
     packet::{AppendEntries, RequestVoteResp, Rpc},
+    queue::ServerEgress,
     raft_state::RaftState,
     server::{Id, PeerId, ServerId},
 };
@@ -215,8 +215,8 @@ impl Candidate {
 mod tests {
     use super::*;
     use crate::{
-        io::testing::{helper_inspect_next_sent_packet, MockIo},
         log::{Term, TermIdx},
+        queue::testing::{helper_inspect_next_sent_packet, MockIo},
         raft_state::RaftState,
         server::PeerId,
         timeout::Timeout,

src/mode/follower.rs

@@ -1,8 +1,8 @@
 use crate::{
-    io::ServerEgress,
     log::MatchOutcome,
     mode::ModeTransition,
     packet::{AppendEntries, Rpc},
+    queue::ServerEgress,
     raft_state::RaftState,
     server::PeerId,
     state_machine::CurrentMode,
@@ -125,8 +125,8 @@ impl Follower {
 mod tests {
     use super::*;
     use crate::{
-        io::testing::{helper_inspect_one_sent_packet, MockIo},
         log::{Entry, Idx, Term, TermIdx},
+        queue::testing::{helper_inspect_one_sent_packet, MockIo},
         raft_state::RaftState,
         server::ServerId,
         timeout::Timeout,

src/mode/leader.rs

@@ -1,8 +1,8 @@
 use crate::{
-    io::ServerEgress,
     log::{Idx, TermIdx},
     mode::Mode,
     packet::{AppendEntriesResp, Rpc},
+    queue::ServerEgress,
     raft_state::RaftState,
     server::{PeerId, ServerId},
     state_machine::CurrentMode,
@@ -296,8 +296,8 @@ impl Leader {
 mod tests {
     use super::*;
     use crate::{
-        io::testing::{helper_inspect_next_sent_packet, MockIo},
         log::{MatchOutcome, Term},
+        queue::testing::{helper_inspect_next_sent_packet, MockIo},
         raft_state::RaftState,
         server::{PeerId, ServerId},
         timeout::Timeout,

src/packet/request_vote.rs

@@ -1,7 +1,7 @@
 use crate::{
-    io::ServerEgress,
     log::{Term, TermIdx},
     packet::Rpc,
+    queue::ServerEgress,
     raft_state::RaftState,
     server::{Id, PeerId},
 };
@@ -31,7 +31,7 @@ impl RequestVote {
         &self,
         peer_id: PeerId,
         raft_state: &mut RaftState,
-        io_egress: &mut E,
+        server: &mut E,
     ) {
         let current_term = raft_state.current_term;
 
@@ -73,7 +73,7 @@ impl RequestVote {
         }
 
         let rpc = Rpc::new_request_vote_resp(current_term, grant_vote);
-        io_egress.send_packet(peer_id, rpc);
+        server.send_packet(peer_id, rpc);
     }
 }
 
@@ -138,9 +138,9 @@ impl EncoderValue for RequestVoteResp {
 mod tests {
     use super::*;
     use crate::{
-        io::testing::{helper_inspect_one_sent_packet, MockIo},
         log::{Entry, Idx, Term, TermIdx},
         macros::cast_unsafe,
+        queue::testing::{helper_inspect_one_sent_packet, MockIo},
         raft_state::RaftState,
         server::{PeerId, ServerId},
         timeout::Timeout,

src/io.rs src/queue.rssrc/io.rs renamed to src/queue.rs

@@ -16,7 +16,7 @@ pub mod testing;
 #[allow(unused_imports)]
 pub use network::NetIngress;
 
-pub use network::{NetEgress, NetworkIoImpl};
+pub use network::{NetEgress, NetworkQueueImpl};
 pub use server_egress::{ServerEgress, ServerEgressImpl};
 pub use server_ingress::{ServerIngress, ServerIngressImpl};
 
@@ -72,14 +72,14 @@ const IO_BUF_LEN: usize = 1024;
 pub struct BufferIo;
 
 impl BufferIo {
-    pub fn split(server_id: ServerId) -> (ServerIngressImpl, ServerEgressImpl, NetworkIoImpl) {
+    pub fn split(server_id: ServerId) -> (ServerIngressImpl, ServerEgressImpl, NetworkQueueImpl) {
         let ingress_queue = Arc::new(Mutex::new(VecDeque::with_capacity(IO_BUF_LEN)));
         let ingress_waker = Arc::new(Mutex::new(None));
 
         let egress_queue = Arc::new(Mutex::new(VecDeque::with_capacity(IO_BUF_LEN)));
         let egress_waker = Arc::new(Mutex::new(None));
 
-        let network_io_handle = NetworkIoImpl {
+        let network_queue_handle = NetworkQueueImpl {
             buf: [0; IO_BUF_LEN],
             ingress_queue: ingress_queue.clone(),
             egress_queue: egress_queue.clone(),
@@ -102,7 +102,7 @@ impl BufferIo {
         (
             server_ingress_handle,
             server_egress_handle,
-            network_io_handle,
+            network_queue_handle,
         )
     }
 }
@@ -132,7 +132,7 @@ macro_rules! impl_io_ready(($io:ident, $fut:ident, $poll_fn:ident) => {
 
 // Implement Futures to poll the queue for readiness.
 impl_io_ready!(ServerIngressImpl, RxReady, poll_ingress_queue_ready);
-impl_io_ready!(NetworkIoImpl, TxReady, poll_egress_queue_ready);
+impl_io_ready!(NetworkQueueImpl, TxReady, poll_egress_queue_ready);
 
 #[cfg(test)]
 mod tests {
@@ -144,12 +144,12 @@ mod tests {
     fn test_helper_io_setup() -> (
         ServerIngressImpl,
         ServerEgressImpl,
-        NetworkIoImpl,
+        NetworkQueueImpl,
         AwokenCount,
         AwokenCount,
     ) {
         let server_id = ServerId::new([1; 16]);
-        let (mut server_ingress, server_egress, mut network_io) = BufferIo::split(server_id);
+        let (mut server_ingress, server_egress, mut network_queue) = BufferIo::split(server_id);
 
         let (ingress_waker, ingress_cnt) = new_count_waker();
         let mut ctx = Context::from_waker(&ingress_waker);
@@ -160,27 +160,27 @@ mod tests {
         let (egress_waker, egress_cnt) = new_count_waker();
         let mut ctx = Context::from_waker(&egress_waker);
         // A egress waker must be registed by calling poll_egress_ready on NetworkIO
-        let _ = network_io.poll_egress_queue_ready(&mut ctx);
+        let _ = network_queue.poll_egress_queue_ready(&mut ctx);
         assert_eq!(egress_cnt, 0);
 
         (
             server_ingress,
             server_egress,
-            network_io,
+            network_queue,
             ingress_cnt,
             egress_cnt,
         )
     }
 
     #[test]
     fn io_recv() {
-        let (mut server_ingress, _server_egress, mut network_io, ingress_cnt, _egress_cnt) =
+        let (mut server_ingress, _server_egress, mut network_queue, ingress_cnt, _egress_cnt) =
             test_helper_io_setup();
 
         // Recv
-        network_io.recv(vec![1]);
+        network_queue.push_recv_bytes(vec![1]);
         assert_eq!(ingress_cnt, 1);
-        network_io.recv(vec![2]);
+        network_queue.push_recv_bytes(vec![2]);
         assert_eq!(ingress_cnt, 2);
 
         assert_eq!(server_ingress.recv_raw(), Some(vec![1, 2]));
@@ -190,7 +190,7 @@ mod tests {
 
     #[test]
     fn io_send() {
-        let (_server_ingress, mut server_egress, mut network_io, _ingress_cnt, egress_cnt) =
+        let (_server_ingress, mut server_egress, mut network_queue, _ingress_cnt, egress_cnt) =
             test_helper_io_setup();
 
         // Send
@@ -200,18 +200,18 @@ mod tests {
         server_egress.send_raw(&[4]);
         assert_eq!(egress_cnt, 2);
 
-        assert_eq!(network_io.send(), Some(vec![3, 4]));
-        assert_eq!(network_io.send(), None);
+        assert_eq!(network_queue.get_send(), Some(vec![3, 4]));
+        assert_eq!(network_queue.get_send(), None);
         assert_eq!(egress_cnt, 2);
     }
 
     #[test]
     fn io_send_recv() {
-        let (mut server_ingress, mut server_egress, mut network_io, ingress_cnt, egress_cnt) =
+        let (mut server_ingress, mut server_egress, mut network_queue, ingress_cnt, egress_cnt) =
             test_helper_io_setup();
 
         // Interleaved send and recv
-        network_io.recv(vec![5]);
+        network_queue.push_recv_bytes(vec![5]);
         assert_eq!(ingress_cnt, 1);
         assert_eq!(egress_cnt, 0);
 
@@ -221,8 +221,8 @@ mod tests {
 
         assert_eq!(server_ingress.recv_raw(), Some(vec![5]));
         assert_eq!(server_ingress.recv_raw(), None);
-        assert_eq!(network_io.send(), Some(vec![6]));
-        assert_eq!(network_io.send(), None);
+        assert_eq!(network_queue.get_send(), Some(vec![6]));
+        assert_eq!(network_queue.get_send(), None);
 
         assert_eq!(ingress_cnt, 1);
         assert_eq!(egress_cnt, 1);

src/io/network.rs src/queue/network.rssrc/io/network.rs renamed to src/queue/network.rs

@@ -1,4 +1,4 @@
-use crate::io::{TxReady, IO_BUF_LEN};
+use crate::queue::{TxReady, IO_BUF_LEN};
 use core::task::{Context, Poll, Waker};
 use std::{
     collections::VecDeque,
@@ -7,9 +7,9 @@ use std::{
     sync::{Arc, Mutex},
 };
 
-/// A handle held by the network task.
+/// A handle held by the Network task for sending and receiving bytes.
 #[derive(Debug, Clone)]
-pub struct NetworkIoImpl {
+pub struct NetworkQueueImpl {
     pub buf: [u8; IO_BUF_LEN],
     pub ingress_queue: Arc<Mutex<VecDeque<u8>>>,
     pub egress_queue: Arc<Mutex<VecDeque<u8>>>,
@@ -20,15 +20,17 @@ pub struct NetworkIoImpl {
 /// Functionality to queue bytes that were received over a socket onto the `ingress_queue`.
 pub trait NetIngress {
     /// Push data to the `ingress_queue`.
-    fn recv(&mut self, data: Vec<u8>);
+    fn push_recv_bytes(&mut self, data: Vec<u8>);
 }
 
 /// Functionality to de-queue bytes from the `egress_queue` so that they can be sent over the
 /// network.
 pub trait NetEgress {
-    /// Send data over the network
-    fn send(&mut self) -> Option<Vec<u8>>;
+    /// Returns data from the `egress_queue` that should to be sent over the network.
+    fn get_send(&mut self) -> Option<Vec<u8>>;
 
+    /// MARKME this needs to be called.
+    ///
     /// Check if there are bytes available in the egress queue for that can be sent on the network.
     fn poll_egress_queue_ready(&mut self, cx: &mut Context) -> Poll<()>;
 
@@ -38,8 +40,8 @@ pub trait NetEgress {
     }
 }
 
-impl NetIngress for NetworkIoImpl {
-    fn recv(&mut self, data: Vec<u8>) {
+impl NetIngress for NetworkQueueImpl {
+    fn push_recv_bytes(&mut self, data: Vec<u8>) {
         // dbg!("  network <--- {}", &data);
 
         self.ingress_queue.lock().unwrap().extend(data);
@@ -49,8 +51,8 @@ impl NetIngress for NetworkIoImpl {
     }
 }
 
-impl NetEgress for NetworkIoImpl {
-    fn send(&mut self) -> Option<Vec<u8>> {
+impl NetEgress for NetworkQueueImpl {
+    fn get_send(&mut self) -> Option<Vec<u8>> {
         let bytes_to_send = self
             .egress_queue
             .lock()

src/io/server_egress.rs src/queue/server_egress.rssrc/io/server_egress.rs renamed to src/queue/server_egress.rs

@@ -1,6 +1,6 @@
 use crate::{
-    io::IO_BUF_LEN,
     packet::{Packet, Rpc},
+    queue::IO_BUF_LEN,
     server::{PeerId, ServerId},
 };
 use core::task::Waker;
@@ -11,7 +11,7 @@ use std::{
     sync::{Arc, Mutex},
 };
 
-/// A handle held by the Raft server task.
+/// A handle held by the Raft server task for sending bytes.
 #[derive(Debug)]
 pub struct ServerEgressImpl {
     pub server_id: ServerId,
@@ -22,9 +22,10 @@ pub struct ServerEgressImpl {
 
 pub trait ServerEgress {
     #[cfg(test)]
-    // Push data to the egress_queue
+    /// Push data to the `egress_queue`
     fn send_raw(&mut self, data: &[u8]);
 
+    /// Push packet to the `egress_queue`
     fn send_packet(&mut self, to: PeerId, rpc: Rpc);
 }
 

src/io/server_ingress.rs src/queue/server_ingress.rssrc/io/server_ingress.rs renamed to src/queue/server_ingress.rs

@@ -1,6 +1,6 @@
 use crate::{
-    io::{RxReady, IO_BUF_LEN},
     packet::Packet,
+    queue::{RxReady, IO_BUF_LEN},
 };
 use core::task::{Context, Poll, Waker};
 use s2n_codec::{DecoderBuffer, DecoderValue};
@@ -10,7 +10,7 @@ use std::{
     sync::{Arc, Mutex},
 };
 
-/// A handle held by the Raft server task.
+/// A handle held by the Raft server task for receiving bytes.
 #[derive(Debug)]
 pub struct ServerIngressImpl {
     pub buf: [u8; IO_BUF_LEN],
@@ -19,18 +19,21 @@ pub struct ServerIngressImpl {
 }
 
 pub trait ServerIngress {
+    /// Read bytes from the `ingress_queue` and return raw bytes.
     #[cfg(test)]
     fn recv_raw(&mut self) -> Option<Vec<u8>>;
 
-    /// Read bytes and return the wrapper [RecvPacket] which can be used to yield individual
-    /// [Packet]'s.
+    /// Read bytes from the `ingress_queue` and return the wrapper [RecvPacket] which can be used
+    /// to yield individual [Packet]'s.
     fn recv_packet(&mut self) -> Option<RecvPacket<'_>>;
 
+    /// MARKME this needs to be called.
+    ///
     /// Check if there are bytes available in the Ingress queue for the server to process.
     fn poll_ingress_queue_ready(&mut self, cx: &mut Context) -> Poll<()>;
 
     /// A Future which can be polled to check for new messages in the queue
-    fn ingress_queue_ready(&mut self) -> RxReady<'_, Self> {
+    fn rx_ready(&mut self) -> RxReady<'_, Self> {
         RxReady(self)
     }
 }