Improve: create a channel notify specifically for interal messages

`tx_notify` will be used for components such as state-machine worker or
replication stream to send back notification when an action is done.

`tx_api` is left for receiving only external messages, such as
append-entries request or client-write request.

A `Balancer` is added to prevent one channel from starving the others.

The benchmark shows a better performance with 64 clients:

| clients | put/s       | ns/op      | Changes         |
| --:     | --:         | --:        | :--             |
|  64     | **730,000** |    1,369   | This commit     |
|  64     | **652,000** |    1,532   | Previous commit |

Author: drmingdrmer

README.md

@@ -95,6 +95,7 @@ Benchmark history:
 
 |  Date      | clients | put/s       | ns/op      | Changes                              |
 | :--        | --:     | --:         | --:        | :--                                  |
+| 2023-04-25 |  64     | **730,000** |    1,369   | Split channels                       |
 | 2023-04-24 |  64     | **652,000** |    1,532   | Reduce metrics report rate           |
 | 2023-04-23 |  64     | **467,000** |    2,139   | State-machine moved to separate task |
 |            |   1     |    70,000   | **14,273** |                                      |

openraft/src/core/balancer.rs

@@ -0,0 +1,46 @@
+//! This mod defines and manipulates the ratio of multiple messages to handle.
+//!
+//! The ratio should be adjust so that every channel won't be starved.
+
+/// Balance the ratio of different kind of message to handle.
+pub(crate) struct Balancer {
+    total: u64,
+
+    /// The number of RaftMsg to handle in each round.
+    raft_msg: u64,
+}
+
+impl Balancer {
+    pub(crate) fn new(total: u64) -> Self {
+        Self {
+            total,
+            // RaftMsg is the input entry.
+            // We should consume as many as internal messages as possible.
+            raft_msg: total / 10,
+        }
+    }
+
+    pub(crate) fn raft_msg(&self) -> u64 {
+        self.raft_msg
+    }
+
+    pub(crate) fn notify(&self) -> u64 {
+        self.total - self.raft_msg
+    }
+
+    pub(crate) fn increase_notify(&mut self) {
+        self.raft_msg = self.raft_msg * 15 / 16;
+        if self.raft_msg == 0 {
+            self.raft_msg = 1;
+        }
+    }
+
+    pub(crate) fn increase_raft_msg(&mut self) {
+        self.raft_msg = self.raft_msg * 17 / 16;
+
+        // Always leave some budget for other channels
+        if self.raft_msg > self.total / 2 {
+            self.raft_msg = self.total / 2;
+        }
+    }
+}

openraft/src/core/mod.rs

@@ -4,6 +4,8 @@
 //! Also it receives and execute `Command` emitted by `Engine` to apply raft state to underlying
 //! storage or forward messages to other raft nodes.
 
+pub(crate) mod balancer;
+pub(crate) mod notify;
 mod raft_core;
 mod replication_state;
 mod server_state;

openraft/src/core/notify.rs

@@ -0,0 +1,113 @@
+use crate::core::sm;
+use crate::raft::VoteResponse;
+use crate::replication::ReplicationResult;
+use crate::replication::ReplicationSessionId;
+use crate::MessageSummary;
+use crate::RaftTypeConfig;
+use crate::StorageError;
+use crate::Vote;
+
+/// A message coming from the internal components.
+pub(crate) enum Notify<C>
+where C: RaftTypeConfig
+{
+    VoteResponse {
+        target: C::NodeId,
+        resp: VoteResponse<C::NodeId>,
+
+        /// Which ServerState sent this message. It is also the requested vote.
+        vote: Vote<C::NodeId>,
+    },
+
+    /// A tick event to wake up RaftCore to check timeout etc.
+    Tick {
+        /// ith tick
+        i: u64,
+    },
+
+    // /// Logs that are submitted to append has been persisted to disk.
+    // LogPersisted {},
+    /// Update the `matched` log id of a replication target.
+    /// Sent by a replication task `ReplicationCore`.
+    UpdateReplicationProgress {
+        /// The ID of the target node for which the match index is to be updated.
+        target: C::NodeId,
+
+        /// The id of the subject that submit this replication action.
+        ///
+        /// It is only used for debugging purpose.
+        id: u64,
+
+        /// Either the last log id that has been successfully replicated to the target,
+        /// or an error in string.
+        result: Result<ReplicationResult<C::NodeId>, String>,
+
+        /// In which session this message is sent.
+        /// A replication session(vote,membership_log_id) should ignore message from other session.
+        session_id: ReplicationSessionId<C::NodeId>,
+    },
+
+    /// [`StorageError`] error has taken place locally(not on remote node) when replicating, and
+    /// [`RaftCore`] needs to shutdown. Sent by a replication task
+    /// [`crate::replication::ReplicationCore`].
+    ReplicationStorageError { error: StorageError<C::NodeId> },
+
+    /// ReplicationCore has seen a higher `vote`.
+    /// Sent by a replication task `ReplicationCore`.
+    HigherVote {
+        /// The ID of the target node from which the new term was observed.
+        target: C::NodeId,
+
+        /// The higher vote observed.
+        higher: Vote<C::NodeId>,
+
+        /// Which ServerState sent this message
+        vote: Vote<C::NodeId>,
+        // TODO: need this?
+        // /// The cluster this replication works for.
+        // membership_log_id: Option<LogId<C::NodeId>>,
+    },
+
+    /// Result of executing a command sent from state machine worker.
+    StateMachine { command_result: sm::CommandResult<C> },
+}
+
+impl<C> MessageSummary<Notify<C>> for Notify<C>
+where C: RaftTypeConfig
+{
+    fn summary(&self) -> String {
+        match self {
+            Notify::VoteResponse { target, resp, vote } => {
+                format!("VoteResponse: from: {}: {}, res-vote: {}", target, resp.summary(), vote)
+            }
+            Notify::Tick { i } => {
+                format!("Tick {}", i)
+            }
+            Notify::UpdateReplicationProgress {
+                ref target,
+                ref id,
+                ref result,
+                ref session_id,
+            } => {
+                format!(
+                    "UpdateReplicationProgress: target: {}, id: {}, result: {:?}, session_id: {}",
+                    target, id, result, session_id,
+                )
+            }
+            Notify::HigherVote {
+                ref target,
+                higher: ref new_vote,
+                ref vote,
+            } => {
+                format!(
+                    "Seen a higher vote: target: {}, vote: {}, server_state_vote: {}",
+                    target, new_vote, vote
+                )
+            }
+            Notify::ReplicationStorageError { error } => format!("ReplicationFatal: {}", error),
+            Notify::StateMachine { command_result: done } => {
+                format!("StateMachine command done: {:?}", done)
+            }
+        }
+    }
+}