Remove async for Raft server processing

Cargo.toml

@@ -16,8 +16,6 @@ config = "~0.14.0"
 crossbeam = { version = "0.8.4", features = ["crossbeam-channel"] }
 derivative = "~2.2.0"
 fs4 = "~0.8.1"
-futures = "~0.3.15"
-futures-util = "~0.3.15"
 hdrhistogram = "~7.5.4"
 hex = "~0.4.3"
 itertools = "0.12.1"
@@ -31,18 +29,6 @@ serde = "~1.0.126"
 serde_bytes = "~0.11.12"
 serde_derive = "~1.0.126"
 simplelog = "~0.12.1"
-tokio = { version = "~1.37.0", features = [
- "macros",
- "rt",
- "rt-multi-thread",
- "net",
- "io-util",
- "time",
- "sync",
-] }
-tokio-serde = { version = "~0.9.0", features = ["bincode"] }
-tokio-stream = { version = "~0.1.6", features = ["net"] }
-tokio-util = { version = "~0.7.8", features = ["codec"] }
 uuid = { version = "~1.8.0", features = ["v4"] }
 
 [dev-dependencies]

src/bin/toydb.rs

@@ -8,7 +8,6 @@
 
 use serde_derive::Deserialize;
 use std::collections::HashMap;
-use tokio::net::TcpListener;
 use toydb::error::{Error, Result};
 use toydb::raft;
 use toydb::sql;
@@ -17,8 +16,7 @@ use toydb::Server;
 
 const COMPACT_MIN_BYTES: u64 = 1024 * 1024;
 
-#[tokio::main]
-async fn main() -> Result<()> {
+fn main() -> Result<()> {
  let args = clap::command!()
  .arg(
  clap::Arg::new("config")
@@ -68,10 +66,10 @@ async fn main() -> Result<()> {
 
  let srv = Server::new(cfg.id, cfg.peers, raft_log, raft_state)?;
 
- let raft_listener = TcpListener::bind(&cfg.listen_raft).await?;
+ let raft_listener = std::net::TcpListener::bind(&cfg.listen_raft)?;
  let sql_listener = std::net::TcpListener::bind(&cfg.listen_sql)?;
 
- srv.serve(raft_listener, sql_listener).await
+ srv.serve(raft_listener, sql_listener)
 }
 
 #[derive(Debug, Deserialize)]

src/error.rs

@@ -168,33 +168,3 @@ impl<T> From<std::sync::PoisonError<T>> for Error {
  Error::Internal(err.to_string())
  }
 }
-
-impl From<tokio::task::JoinError> for Error {
- fn from(err: tokio::task::JoinError) -> Self {
- Error::Internal(err.to_string())
- }
-}
-
-impl From<tokio::sync::mpsc::error::TryRecvError> for Error {
- fn from(err: tokio::sync::mpsc::error::TryRecvError) -> Self {
- Error::Internal(err.to_string())
- }
-}
-
-impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {
- fn from(err: tokio::sync::mpsc::error::SendError<T>) -> Self {
- Error::Internal(err.to_string())
- }
-}
-
-impl<T> From<tokio::sync::mpsc::error::TrySendError<T>> for Error {
- fn from(err: tokio::sync::mpsc::error::TrySendError<T>) -> Self {
- Error::Internal(err.to_string())
- }
-}
-
-impl From<tokio::sync::oneshot::error::RecvError> for Error {
- fn from(err: tokio::sync::oneshot::error::RecvError) -> Self {
- Error::Internal(err.to_string())
- }
-}

src/raft/node/candidate.rs

@@ -173,11 +173,10 @@ mod tests {
  use super::super::tests::{assert_messages, assert_node};
  use super::*;
  use crate::storage;
- use tokio::sync::mpsc;
 
  #[allow(clippy::type_complexity)]
- fn setup() -> Result<(RawNode<Candidate>, mpsc::UnboundedReceiver<Message>)> {
- let (node_tx, node_rx) = mpsc::unbounded_channel();
+ fn setup() -> Result<(RawNode<Candidate>, crossbeam::channel::Receiver<Message>)> {
+ let (node_tx, node_rx) = crossbeam::channel::unbounded();
  let state = Box::new(TestState::new(0));
  let mut log = Log::new(storage::Memory::new(), false)?;
 

src/raft/node/follower.rs

@@ -4,7 +4,6 @@ use crate::error::{Error, Result};
 
 use ::log::{debug, info};
 use std::collections::HashSet;
-use tokio::sync::mpsc;
 
 // A follower replicates state from a leader.
 #[derive(Clone, Debug, PartialEq)]
@@ -44,7 +43,7 @@ impl RawNode<Follower> {
  peers: HashSet<NodeID>,
  mut log: Log,
  state: Box<dyn State>,
- node_tx: mpsc::UnboundedSender<Message>,
+ node_tx: crossbeam::channel::Sender<Message>,
  ) -> Result<Self> {
  let (term, voted_for) = log.get_term()?;
  let role = Follower::new(None, voted_for);
@@ -288,11 +287,10 @@ pub mod tests {
  use super::*;
  use crate::error::Error;
  use crate::storage;
- use tokio::sync::mpsc;
 
  #[allow(clippy::type_complexity)]
- fn setup() -> Result<(RawNode<Follower>, mpsc::UnboundedReceiver<Message>)> {
- let (node_tx, node_rx) = mpsc::unbounded_channel();
+ fn setup() -> Result<(RawNode<Follower>, crossbeam::channel::Receiver<Message>)> {
+ let (node_tx, node_rx) = crossbeam::channel::unbounded();
  let state = Box::new(TestState::new(0));
  let mut log = Log::new(storage::Memory::new(), false)?;
  log.append(1, Some(vec![0x01]))?;
@@ -590,7 +588,7 @@ pub mod tests {
  // AppendEntries accepts some entries at base 0 without changes
  fn step_appendentries_base0() -> Result<()> {
  // TODO: Move this into a setup function.
- let (node_tx, mut node_rx) = mpsc::unbounded_channel();
+ let (node_tx, mut node_rx) = crossbeam::channel::unbounded();
  let mut log = Log::new(storage::Memory::new(), false)?;
  log.append(1, Some(vec![0x01]))?;
  log.append(1, Some(vec![0x02]))?;

src/raft/node/leader.rs

@@ -407,11 +407,10 @@ mod tests {
  use super::*;
  use crate::storage;
  use pretty_assertions::assert_eq;
- use tokio::sync::mpsc;
 
  #[allow(clippy::type_complexity)]
- fn setup() -> Result<(RawNode<Leader>, mpsc::UnboundedReceiver<Message>)> {
- let (node_tx, node_rx) = mpsc::unbounded_channel();
+ fn setup() -> Result<(RawNode<Leader>, crossbeam::channel::Receiver<Message>)> {
+ let (node_tx, node_rx) = crossbeam::channel::unbounded();
  let peers = HashSet::from([2, 3, 4, 5]);
  let state = Box::new(TestState::new(0));
  let mut log = Log::new(storage::Memory::new(), false)?;
@@ -688,7 +687,7 @@ mod tests {
  #[test]
  // Sending a mutate request should append it to log, replicate it to peers, and register notification.
  fn step_clientrequest_mutate() -> Result<()> {
- let (leader, mut node_rx) = setup()?;
+ let (leader, node_rx) = setup()?;
  let peers = leader.peers.clone();
  let mut node: Node = leader.into();
 

src/raft/node/mod.rs

@@ -11,7 +11,6 @@ use leader::Leader;
 use ::log::debug;
 use rand::Rng as _;
 use std::collections::HashSet;
-use tokio::sync::mpsc;
 
 /// A node ID.
 pub type NodeID = u8;
@@ -56,7 +55,7 @@ impl Node {
  peers: HashSet<NodeID>,
  log: Log,
  state: Box<dyn State>,
- node_tx: mpsc::UnboundedSender<Message>,
+ node_tx: crossbeam::channel::Sender<Message>,
  ) -> Result<Self> {
  let node = RawNode::new(id, peers, log, state, node_tx)?;
  if node.peers.is_empty() {
@@ -126,7 +125,7 @@ pub struct RawNode<R: Role = Follower> {
  term: Term,
  log: Log,
  state: Box<dyn State>,
- node_tx: mpsc::UnboundedSender<Message>,
+ node_tx: crossbeam::channel::Sender<Message>,
  role: R,
 }
 
@@ -258,11 +257,10 @@ mod tests {
  use crate::storage;
  use pretty_assertions::assert_eq;
  use std::collections::HashSet;
- use tokio::sync::mpsc;
 
  #[track_caller]
  pub fn assert_messages<T: std::fmt::Debug + PartialEq>(
- rx: &mut mpsc::UnboundedReceiver<T>,
+ rx: &mut crossbeam::channel::Receiver<T>,
  msgs: Vec<T>,
  ) {
  let mut actual = Vec::new();
@@ -432,14 +430,14 @@ mod tests {
  NodeAsserter::new(node)
  }
 
- fn setup_rolenode() -> Result<(RawNode<Follower>, mpsc::UnboundedReceiver<Message>)> {
+ fn setup_rolenode() -> Result<(RawNode<Follower>, crossbeam::channel::Receiver<Message>)> {
  setup_rolenode_peers(vec![2, 3])
  }
 
  fn setup_rolenode_peers(
  peers: Vec<NodeID>,
- ) -> Result<(RawNode<Follower>, mpsc::UnboundedReceiver<Message>)> {
- let (node_tx, node_rx) = mpsc::unbounded_channel();
+ ) -> Result<(RawNode<Follower>, crossbeam::channel::Receiver<Message>)> {
+ let (node_tx, node_rx) = crossbeam::channel::unbounded();
  let node = RawNode {
  role: Follower::new(None, None),
  id: 1,
@@ -454,7 +452,7 @@ mod tests {
 
  #[test]
  fn new() -> Result<()> {
- let (node_tx, _) = mpsc::unbounded_channel();
+ let (node_tx, _) = crossbeam::channel::unbounded();
  let node = Node::new(
  1,
  HashSet::from([2, 3]),
@@ -475,7 +473,7 @@ mod tests {
 
  #[test]
  fn new_single() -> Result<()> {
- let (node_tx, _node_rx) = mpsc::unbounded_channel();
+ let (node_tx, _node_rx) = crossbeam::channel::unbounded();
  let node = Node::new(
  1,
  HashSet::new(),