raft: replace Node with BoxedNode

Author: erikgrinaker

src/raft/node/candidate.rs

@@ -1,5 +1,7 @@
 use super::super::{Address, Event, Message};
-use super::{rand_election_timeout, Follower, Leader, Node, NodeID, RoleNode, Term, Ticks};
+use super::{
+    rand_election_timeout, BoxedNode, DynNode, Follower, Leader, NodeID, RoleNode, Term, Ticks,
+};
 use crate::error::{Error, Result};
 
 use ::log::{debug, info, warn};
@@ -81,22 +83,42 @@ impl RoleNode<Candidate> {
         Ok(node)
     }
 
-    /// Processes a message.
-    pub fn step(mut self, msg: Message) -> Result<Node> {
+    /// Campaign for leadership by increasing the term, voting for ourself, and
+    /// soliciting votes from all peers.
+    pub(super) fn campaign(&mut self) -> Result<()> {
+        let term = self.term + 1;
+        info!("Starting new election for term {}", term);
+        self.role = Candidate::new();
+        self.role.votes.insert(self.id); // vote for ourself
+        self.term = term;
+        self.log.set_term(term, Some(self.id))?;
+
+        let (last_index, last_term) = self.log.get_last_index();
+        self.send(Address::Broadcast, Event::SolicitVote { last_index, last_term })?;
+        Ok(())
+    }
+}
+
+impl DynNode for RoleNode<Candidate> {
+    fn id(&self) -> NodeID {
+        self.id
+    }
+
+    fn step(mut self: Box<Self>, msg: Message) -> Result<BoxedNode> {
         self.assert()?;
         self.assert_step(&msg);
 
         // Drop messages from past terms.
         if msg.term < self.term && msg.term > 0 {
             debug!("Dropping message from past term ({:?})", msg);
-            return Ok(self.into());
+            return Ok(self);
         }
 
         // If we receive a message for a future term, become a leaderless
         // follower in it and step the message. If the message is a Heartbeat or
         // AppendEntries from the leader, stepping it will follow the leader.
         if msg.term > self.term {
-            return self.become_follower(msg.term, None)?.step(msg);
+            return Box::new(self.become_follower(msg.term, None)?).step(msg);
         }
 
         match msg.event {
@@ -108,14 +130,15 @@ impl RoleNode<Candidate> {
             Event::GrantVote => {
                 self.role.votes.insert(msg.from.unwrap());
                 if self.role.votes.len() as u64 >= self.quorum() {
-                    return Ok(self.become_leader()?.into());
+                    return Ok(Box::new(self.become_leader()?));
                 }
             }
 
             // If we receive a heartbeat or entries in this term, we lost the
             // election and have a new leader. Follow it and step the message.
             Event::Heartbeat { .. } | Event::AppendEntries { .. } => {
-                return self.become_follower(msg.term, Some(msg.from.unwrap()))?.step(msg);
+                return Box::new(self.become_follower(msg.term, Some(msg.from.unwrap()))?)
+                    .step(msg);
             }
 
             // Abort any inbound client requests while candidate.
@@ -130,32 +153,16 @@ impl RoleNode<Candidate> {
             | Event::RejectEntries { .. }
             | Event::ClientResponse { .. } => warn!("Received unexpected message {:?}", msg),
         }
-        Ok(self.into())
+        Ok(self)
     }
 
-    /// Processes a logical clock tick.
-    pub fn tick(mut self) -> Result<Node> {
+    fn tick(mut self: Box<Self>) -> Result<BoxedNode> {
         self.assert()?;
 
         self.role.election_duration += 1;
         if self.role.election_duration >= self.role.election_timeout {
             self.campaign()?;
         }
-        Ok(self.into())
-    }
-
-    /// Campaign for leadership by increasing the term, voting for ourself, and
-    /// soliciting votes from all peers.
-    pub(super) fn campaign(&mut self) -> Result<()> {
-        let term = self.term + 1;
-        info!("Starting new election for term {}", term);
-        self.role = Candidate::new();
-        self.role.votes.insert(self.id); // vote for ourself
-        self.term = term;
-        self.log.set_term(term, Some(self.id))?;
-
-        let (last_index, last_term) = self.log.get_last_index();
-        self.send(Address::Broadcast, Event::SolicitVote { last_index, last_term })?;
-        Ok(())
+        Ok(self)
     }
 }

src/raft/node/follower.rs

@@ -1,5 +1,5 @@
 use super::super::{Address, Event, Instruction, Message, RequestID, Response};
-use super::{rand_election_timeout, Candidate, Node, NodeID, RoleNode, Term, Ticks};
+use super::{rand_election_timeout, BoxedNode, Candidate, DynNode, NodeID, RoleNode, Term, Ticks};
 use crate::error::{Error, Result};
 
 use ::log::{debug, error, info, warn};
@@ -94,22 +94,46 @@ impl RoleNode<Follower> {
         Ok(self)
     }
 
-    /// Processes a message.
-    pub fn step(mut self, msg: Message) -> Result<Node> {
+    /// Aborts all forwarded requests.
+    fn abort_forwarded(&mut self) -> Result<()> {
+        for id in std::mem::take(&mut self.role.forwarded) {
+            debug!("Aborting forwarded request {:x?}", id);
+            self.send(Address::Client, Event::ClientResponse { id, response: Err(Error::Abort) })?;
+        }
+        Ok(())
+    }
+
+    /// Checks if an address is the current leader.
+    fn is_leader(&self, from: &Address) -> bool {
+        if let Some(leader) = &self.role.leader {
+            if let Address::Node(from) = from {
+                return leader == from;
+            }
+        }
+        false
+    }
+}
+
+impl DynNode for RoleNode<Follower> {
+    fn id(&self) -> NodeID {
+        self.id
+    }
+
+    fn step(mut self: Box<Self>, msg: Message) -> Result<BoxedNode> {
         self.assert()?;
         self.assert_step(&msg);
 
         // Drop messages from past terms.
         if msg.term < self.term && msg.term > 0 {
             debug!("Dropping message from past term ({:?})", msg);
-            return Ok(self.into());
+            return Ok(self);
         }
 
         // If we receive a message for a future term, become a leaderless
         // follower in it and step the message. If the message is a Heartbeat or
         // AppendEntries from the leader, stepping it will follow the leader.
         if msg.term > self.term {
-            return self.become_follower(None, msg.term)?.step(msg);
+            return Box::new(self.become_follower(None, msg.term)?).step(msg);
         }
 
         // Record when we last saw a message from the leader (if any).
@@ -126,7 +150,7 @@ impl RoleNode<Follower> {
                 let from = msg.from.unwrap();
                 match self.role.leader {
                     Some(leader) => assert_eq!(from, leader, "Multiple leaders in term"),
-                    None => self = self.become_follower(Some(from), msg.term)?,
+                    None => self = Box::new(self.become_follower(Some(from), msg.term)?),
                 }
 
                 // Advance commit index and apply entries if possible.
@@ -149,7 +173,7 @@ impl RoleNode<Follower> {
                 let from = msg.from.unwrap();
                 match self.role.leader {
                     Some(leader) => assert_eq!(from, leader, "Multiple leaders in term"),
-                    None => self = self.become_follower(Some(from), msg.term)?,
+                    None => self = Box::new(self.become_follower(Some(from), msg.term)?),
                 }
 
                 // Append the entries, if possible.
@@ -169,7 +193,7 @@ impl RoleNode<Follower> {
                 // If we already voted for someone else in this term, ignore it.
                 if let Some(voted_for) = self.role.voted_for {
                     if from != voted_for {
-                        return Ok(self.into());
+                        return Ok(self);
                     }
                 }
 
@@ -189,7 +213,7 @@ impl RoleNode<Follower> {
             Event::ClientRequest { ref id, .. } => {
                 if msg.from != Address::Client {
                     error!("Received client request from non-client {:?}", msg.from);
-                    return Ok(self.into());
+                    return Ok(self);
                 }
 
                 let id = id.clone();
@@ -206,7 +230,7 @@ impl RoleNode<Follower> {
             Event::ClientResponse { id, mut response } => {
                 if !self.is_leader(&msg.from) {
                     error!("Received client response from non-leader {:?}", msg.from);
-                    return Ok(self.into());
+                    return Ok(self);
                 }
 
                 // TODO: Get rid of this field, it should be returned at the RPC
@@ -225,36 +249,16 @@ impl RoleNode<Follower> {
             | Event::RejectEntries { .. }
             | Event::GrantVote { .. } => warn!("Received unexpected message {:?}", msg),
         };
-        Ok(self.into())
+        Ok(self)
     }
 
-    /// Processes a logical clock tick.
-    pub fn tick(mut self) -> Result<Node> {
+    fn tick(mut self: Box<Self>) -> Result<BoxedNode> {
         self.assert()?;
 
         self.role.leader_seen += 1;
         if self.role.leader_seen >= self.role.election_timeout {
-            return Ok(self.become_candidate()?.into());
+            return Ok(Box::new(self.become_candidate()?));
         }
-        Ok(self.into())
-    }
-
-    /// Aborts all forwarded requests.
-    fn abort_forwarded(&mut self) -> Result<()> {
-        for id in std::mem::take(&mut self.role.forwarded) {
-            debug!("Aborting forwarded request {:x?}", id);
-            self.send(Address::Client, Event::ClientResponse { id, response: Err(Error::Abort) })?;
-        }
-        Ok(())
-    }
-
-    /// Checks if an address is the current leader.
-    fn is_leader(&self, from: &Address) -> bool {
-        if let Some(leader) = &self.role.leader {
-            if let Address::Node(from) = from {
-                return leader == from;
-            }
-        }
-        false
+        Ok(self)
     }
 }