why RaftCore::append_to_log is temporarily set to blocking

[ukernel]

This serializes all client writes. I hope to know why it's there and if it will be fixed in the future. Thanks

[ukernel]

can following code fix this issue? Thanks

diff --git a/openraft/src/core/raft_core.rs b/openraft/src/core/raft_core.rs
index 77870bfc..de034882 100644
--- a/openraft/src/core/raft_core.rs
+++ b/openraft/src/core/raft_core.rs
@@ -9,9 +9,10 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyerror::AnyError;
-use futures::stream::FuturesUnordered;
-use futures::StreamExt;
-use futures::TryFutureExt;
+use futures::{
+    stream::{FuturesOrdered, FuturesUnordered},
+    StreamExt, TryFutureExt
+};
 use maplit::btreeset;
 use tokio::select;
 use tokio::sync::mpsc;
@@ -158,6 +159,10 @@ impl LeaderData {
 
 // TODO: remove SM
 /// The core type implementing the Raft protocol.
+
+type LogFLushResult = Result::NodeId>, std::io::Error>;
+type OneshotReceiver = <::AsyncRuntime as AsyncRuntime>::OneshotReceiver;
+
 pub struct RaftCore
 where
     C: RaftTypeConfig,
@@ -208,6 +213,8 @@ where
 
     pub(crate) span: Span,
 
+    pub(crate) stream_log_flushed: FuturesOrdered>>,
+
     pub(crate) _p: PhantomData,
 }
 
@@ -735,9 +742,8 @@ where
         let callback = LogFlushed::new(log_io_id, tx);
 
         self.log_store.append(entries, callback).await?;
-        rx.await
-            .map_err(|e| StorageIOError::write_logs(AnyError::error(e)))?
-            .map_err(|e| StorageIOError::write_logs(AnyError::error(e)))?;
+
+        self.stream_log_flushed.push_back(rx);
         Ok(())
     }
 
@@ -931,6 +937,28 @@ where
             // In each loop, the first step is blocking waiting for any message from any channel.
             // Then if there is any message, process as many as possible to maximize throughput.
 
+            let poll_log_flush = async {
+                if self.stream_log_flushed.is_empty() {
+                    return None;
+                }
+                while let Some(flush_res) = self.stream_log_flushed.next().await {
+                    let log_io_id = flush_res
+                        .map_err(|e| AnyError::error(e))
+                        .and_then(|r| r.map_err(|e| AnyError::error(e)));
+                    match log_io_id {
+                        Ok(log_io_id) => {
+                            // The leader may have changed.
+                            // But reporting to a different leader is not a problem.
+                            if let Ok(mut lh) = self.engine.leader_handler() {
+                                lh.replication_handler().update_local_progress(log_io_id.log_id);
+                            }
+                        }
+                        Err(err) => { return Some(Err(err)); }
+                    }
+                }
+                Some(Ok(()))
+            };
+
             select! {
                 // Check shutdown in each loop first so that a message flood in `tx_api` won't block shutting down.
                 // `select!` without `biased` provides a random fairness.
@@ -943,6 +971,10 @@ where
                     return Err(Fatal::Stopped);
                 }
 
+                Some(flush_res) = poll_log_flush => {
+                    flush_res.map_err(|e| Into::>::into(StorageIOError::write_logs(e)))?;
+                }
+
                 notify_res = self.rx_notify.recv() => {
                     match notify_res {
                         Some(notify) => self.handle_notify(notify)?,
@@ -1611,15 +1643,9 @@ where
             }
             Command::AppendInputEntries { vote, entries } => {
                 let last_log_id = *entries.last().unwrap().get_log_id();
-                tracing::debug!("AppendInputEntries: {}", DisplaySlice::<_>(&entries),);
-
+                tracing::debug!("AppendInputEntries: {}", DisplaySlice::<_>(&entries));
                 self.append_to_log(entries, vote, last_log_id).await?;
 
-                // The leader may have changed.
-                // But reporting to a different leader is not a problem.
-                if let Ok(mut lh) = self.engine.leader_handler() {
-                    lh.replication_handler().update_local_progress(Some(last_log_id));
-                }
             }
             Command::SaveVote { vote } => {
                 self.log_store.save_vote(&vote).await?;
diff --git a/openraft/src/raft/mod.rs b/openraft/src/raft/mod.rs
index a540ef37..3aba1f22 100644
--- a/openraft/src/raft/mod.rs
+++ b/openraft/src/raft/mod.rs
@@ -300,6 +300,8 @@ where C: RaftTypeConfig
             command_state: CommandState::default(),
             span: core_span,
 
+            stream_log_flushed: futures::stream::FuturesOrdered::new(),
+
             _p: Default::default(),
         };

[schreter]

Hi @ukernel ,

thanks for looking into it.

Yes, the issue should be fixed ultimately to allow more dataflow-like processing. Your proposal seems on the first look like a workable solution (after some clean-ups), but I don't know all the dependencies inside of RaftCore, i.e., whether this might cause other problems.

In any case, feel free to post it as a PR, it's easier to review and comment than a diff in an issue. Invite @drmingdrmer for the review, he's the main maintainer of the crate and he definitely knows the dependencies the best.

The issue I see is that FuturesOrdered seems to be optimized for a use case with large number of heavyweight futures with a large number of memory allocations happening in process (vectors, deques, binary heap, etc.). We have a small number of lightweight operations. I.e., the solution will cause unnecessary overhead and later headaches when cleaning it up. Even if you look at join_all implementation in futures crate, it will not switch to FuturesOrdered until quite late - at >30 (!) futures. There's a reason for that. My expectation is that we'll have regularly 0..1 and sometimes a few outstanding futures.

But, the primary issue is that those callbacks are currently represented as a oneshot channel with a respective Future to await. They don't have to be. A better solution would be to represent a sequence of callbacks as a mpsc queue, where each callback pushes its completion into the queue. E.g., the completion message would be a pair of index (taken at the time of callback creation) and the result of the I/O. The main task then just select!s this queue and automatically has results w/o any other helpers/complex structures. These results need to be potentially reordered, since callbacks can complete out-of-order, but otherwise it's pretty simple to handle. I.e., in the select above you wouldn't poll future completions, but rather completion messages from the mpsc queue and reorder them. Either the result is at the expected next index, then you process it right away, else you push the result into a BinaryHeap sorted by the index. After you processed the result, you peek into the BinaryHeap whether the next reordered result is the next expected result and as long as it is so, you keep processing results from BinaryHeap (until it is empty or you find a hole in index sequence).

Also please note that the serialization at this point is not as bad as it seems since the logs are processed in batches. I.e., as long as we are waiting for the first batch to complete, the next batch is put together. Having said that, it's still not an excuse not to optimize it :-).

[drmingdrmer]

Thank you for bringing attention to this issue and suggesting a solution.

To make RaftLogStorage I/O asynchronous, the solution might indeed be similar to the one you've proposed. However, there are several issues that need to be addressed before proceeding with this refactor.

To ensure that log I/O is asynchronous, all I/O operations in RaftLogStorage must be asynchronous. For example, save_vote(v1) should not be submitted to RaftLogStorage until the last submitted append(log2) is confirmed as flushed. If the application does not manage it correctly, the storage might persist v1 before persisting log2.

A more effective storage API would handle all write I/O in a callback manner, allowing the application to serialize all I/O operations in a single internal queue or similar structure.

Currently, log I/O operations are blocking, which naturally serializes them based on the execution order. Therefore, to transition to non-blocking and asynchronous operations, the RaftLogStorage API would need to be refactored first.

[ukernel]

Thank you for bringing attention to this issue and suggesting a solution.

To make RaftLogStorage I/O asynchronous, the solution might indeed be similar to the one you've proposed. However, there are several issues that need to be addressed before proceeding with this refactor.

To ensure that log I/O is asynchronous, all I/O operations in RaftLogStorage must be asynchronous. For example, save_vote(v1) should not be submitted to RaftLogStorage until the last submitted append(log2) is confirmed as flushed. If the application does not manage it correctly, the storage might persist v1 before persisting log2.

Can we assume RaftLogStorage behaves like a log. If save_vote(v1) is submitted after append(log2) and save_vote(v1) is persisted, append(log2) must be persisted as well.

A more effective storage API would handle all write I/O in a callback manner, allowing the application to serialize all I/O operations in a single internal queue or similar structure.

Currently, log I/O operations are blocking, which naturally serializes them based on the execution order. Therefore, to transition to non-blocking and asynchronous operations, the RaftLogStorage API would need to be refactored first.

I checked 0.9.x code, all functions in RaftLogStorage trait are async functions. Do you mean change them to callback manner?

[drmingdrmer]

Can we assume RaftLogStorage behaves like a log. If save_vote(v1) is submitted after append(log2) and save_vote(v1) is persisted, append(log2) must be persisted as well.

It should behave like that. But if save_vote() is a blocking API and append() is a non-blocking API, the application may not implement it correctly.

I checked 0.9.x code, all functions in RaftLogStorage trait are async functions. Do you mean change them to callback manner?

Yes.