Refactor: remove ClientRequestEntry;

It does not need to cache the entry, only the response tx.

Log entry can be fetched from storage by index.
Only tx channels for sending back result to client are stored and are
indexed by log-index.

Author: drmingdrmer

openraft/src/core/admin.rs

@@ -1,11 +1,9 @@
 use std::collections::BTreeMap;
 use std::collections::BTreeSet;
 use std::option::Option::None;
-use std::sync::Arc;
 
 use tracing::warn;
 
-use crate::core::client::ClientRequestEntry;
 use crate::core::LeaderState;
 use crate::core::LearnerState;
 use crate::core::State;
@@ -287,12 +285,7 @@ impl<'a, C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> LeaderS
 
         self.core.metrics_flags.set_data_changed();
 
-        let cr_entry = ClientRequestEntry {
-            entry: Arc::new(entry),
-            tx: resp_tx,
-        };
-
-        self.replicate_client_request(cr_entry).await?;
+        self.replicate_client_request(entry, resp_tx).await?;
 
         Ok(())
     }

openraft/src/core/client.rs

@@ -1,5 +1,3 @@
-use std::sync::Arc;
-
 use futures::future::TryFutureExt;
 use futures::stream::FuturesUnordered;
 use futures::stream::StreamExt;
@@ -32,24 +30,6 @@ use crate::RaftStorage;
 use crate::RaftTypeConfig;
 use crate::StorageError;
 
-/// A wrapper around a ClientRequest which has been transformed into an Entry, along with its response channel.
-pub(super) struct ClientRequestEntry<C: RaftTypeConfig> {
-    /// The Arc'd entry of the ClientRequest.
-    ///
-    /// This value is Arc'd so that it may be sent across thread boundaries for replication
-    /// without having to clone the data payload itself.
-    pub entry: Arc<Entry<C>>,
-
-    /// The response channel for the request.
-    pub tx: Option<RaftRespTx<ClientWriteResponse<C>, ClientWriteError<C::NodeId>>>,
-}
-
-impl<C: RaftTypeConfig> MessageSummary for ClientRequestEntry<C> {
-    fn summary(&self) -> String {
-        format!("entry:{}", self.entry.summary())
-    }
-}
-
 impl<'a, C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> LeaderState<'a, C, N, S> {
     /// Commit the initial entry which new leaders are obligated to create when first coming to power, per §8.
     #[tracing::instrument(level = "trace", skip(self))]
@@ -58,12 +38,7 @@ impl<'a, C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> LeaderS
 
         self.core.metrics_flags.set_data_changed();
 
-        let cr_entry = ClientRequestEntry {
-            entry: Arc::new(entry),
-            tx: None,
-        };
-
-        self.replicate_client_request(cr_entry).await?;
+        self.replicate_client_request(entry, None).await?;
 
         Ok(())
     }
@@ -194,14 +169,10 @@ impl<'a, C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> LeaderS
         tx: RaftRespTx<ClientWriteResponse<C>, ClientWriteError<C::NodeId>>,
     ) -> Result<(), StorageError<C::NodeId>> {
         let entry = self.core.append_payload_to_log(rpc.payload).await?;
-        let entry = ClientRequestEntry {
-            entry: Arc::new(entry),
-            tx: Some(tx),
-        };
 
         self.core.metrics_flags.set_data_changed();
 
-        self.replicate_client_request(entry).await?;
+        self.replicate_client_request(entry, Some(tx)).await?;
         Ok(())
     }
 
@@ -210,15 +181,20 @@ impl<'a, C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> LeaderS
     /// NOTE WELL: this routine does not wait for the request to actually finish replication, it
     /// merely beings the process. Once the request is committed to the cluster, its response will
     /// be generated asynchronously.
-    #[tracing::instrument(level = "debug", skip(self, req), fields(req=%req.summary()))]
+    #[tracing::instrument(level = "debug", skip(self, entry, resp_tx), fields(req=%entry.summary()))]
     pub(super) async fn replicate_client_request(
         &mut self,
-        req: ClientRequestEntry<C>,
+        entry: Entry<C>,
+        resp_tx: Option<RaftRespTx<ClientWriteResponse<C>, ClientWriteError<C::NodeId>>>,
     ) -> Result<(), StorageError<C::NodeId>> {
         // Replicate the request if there are other cluster members. The client response will be
         // returned elsewhere after the entry has been committed to the cluster.
 
-        let log_id = req.entry.log_id;
+        let log_id = entry.log_id;
+        if let Some(tx) = resp_tx {
+            self.client_resp_channels.insert(log_id.index, tx);
+        }
+
         let quorum_granted = self.core.effective_membership.membership.is_majority(&btreeset! {self.core.id});
 
         if quorum_granted {
@@ -228,9 +204,7 @@ impl<'a, C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> LeaderS
             tracing::debug!(?self.core.committed, "update committed, no need to replicate");
 
             self.core.metrics_flags.set_data_changed();
-            self.client_request_post_commit(req).await?;
-        } else {
-            self.awaiting_committed.push(req);
+            self.client_request_post_commit(log_id.index).await?;
         }
 
         for node in self.nodes.values() {
@@ -247,16 +221,16 @@ impl<'a, C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> LeaderS
     }
 
     /// Handle the post-commit logic for a client request.
-    #[tracing::instrument(level = "debug", skip(self, req))]
-    pub(super) async fn client_request_post_commit(
-        &mut self,
-        req: ClientRequestEntry<C>,
-    ) -> Result<(), StorageError<C::NodeId>> {
-        let entry = &req.entry;
+    #[tracing::instrument(level = "debug", skip(self))]
+    pub(super) async fn client_request_post_commit(&mut self, log_index: u64) -> Result<(), StorageError<C::NodeId>> {
+        let entries = self.core.storage.get_log_entries(log_index..=log_index).await?;
+        let entry = &entries[0];
+
+        let tx = self.client_resp_channels.remove(&log_index);
 
         let apply_res = self.apply_entry_to_state_machine(entry).await?;
 
-        self.send_response(entry, apply_res, req.tx).await;
+        self.send_response(entry, apply_res, tx).await;
 
         // Trigger log compaction if needed.
         self.core.trigger_log_compaction_if_needed(false).await;

openraft/src/core/mod.rs

@@ -35,8 +35,8 @@ use tracing::Span;
 
 use crate::config::Config;
 use crate::config::SnapshotPolicy;
-use crate::core::client::ClientRequestEntry;
 use crate::error::AddLearnerError;
+use crate::error::ClientWriteError;
 use crate::error::ExtractFatal;
 use crate::error::Fatal;
 use crate::error::ForwardToLeader;
@@ -45,6 +45,7 @@ use crate::error::NotAllowed;
 use crate::metrics::RaftMetrics;
 use crate::metrics::ReplicationMetrics;
 use crate::raft::AddLearnerResponse;
+use crate::raft::ClientWriteResponse;
 use crate::raft::RaftMsg;
 use crate::raft::RaftRespTx;
 use crate::raft::VoteResponse;
@@ -797,8 +798,8 @@ struct LeaderState<'a, C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStora
     /// The cloneable sender channel for replication stream events.
     pub(super) replication_tx: mpsc::UnboundedSender<(ReplicaEvent<C, S::SnapshotData>, Span)>,
 
-    /// A buffer of client requests which have been appended locally and are awaiting to be committed to the cluster.
-    pub(super) awaiting_committed: Vec<ClientRequestEntry<C>>,
+    /// Channels to send result back to client when logs are committed.
+    pub(super) client_resp_channels: BTreeMap<u64, RaftRespTx<ClientWriteResponse<C>, ClientWriteError<C::NodeId>>>,
 }
 
 impl<'a, C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> MetricsProvider<C::NodeId>
@@ -819,7 +820,7 @@ impl<'a, C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> LeaderS
             replication_metrics: Versioned::new(ReplicationMetrics::default()),
             replication_tx,
             replication_rx,
-            awaiting_committed: Vec::new(),
+            client_resp_channels: Default::default(),
         }
     }
 

openraft/src/core/replication.rs

@@ -20,6 +20,7 @@ use crate::summary::MessageSummary;
 use crate::versioned::Updatable;
 use crate::vote::Vote;
 use crate::LogId;
+use crate::LogIdOptionExt;
 use crate::RaftNetworkFactory;
 use crate::RaftStorage;
 use crate::RaftTypeConfig;
@@ -156,21 +157,9 @@ impl<'a, C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> LeaderS
                 ));
             }
 
-            // Check if there are any pending requests which need to be processed.
-            let filter = self
-                .awaiting_committed
-                .iter()
-                .enumerate()
-                .take_while(|(_idx, elem)| Some(elem.entry.log_id) <= self.core.committed)
-                .last()
-                .map(|(idx, _)| idx);
-
-            if let Some(offset) = filter {
-                // Build a new ApplyLogsTask from each of the given client requests.
-
-                for request in self.awaiting_committed.drain(..=offset).collect::<Vec<_>>() {
-                    self.client_request_post_commit(request).await?;
-                }
+            // Apply committed entries, and send applying result to client if there is a channel awaiting it
+            for i in self.core.last_applied.next_index()..self.core.committed.next_index() {
+                self.client_request_post_commit(i).await?;
             }
 
             self.core.metrics_flags.set_data_changed();