Merge branch streaming (#552)

Merge streaming branch

.github/workflows/cmake.yml

@@ -9,6 +9,7 @@ on:
   pull_request:
     branches:
       - master
+      - streaming
 
 env:
   # Customize the CMake build type here (Release, Debug, RelWithDebInfo, etc.)

CMakeLists.txt

@@ -332,6 +332,8 @@ if (CODE_COVERAGE GREATER 0)
         stat_mgr_test
         logger_test
         new_joiner_test
+        asio_service_stream_test
+        stream_functional_test
     )
 
     # lcov

include/libnuraft/asio_service_options.hxx

@@ -125,6 +125,7 @@ struct asio_service_options {
         , crc_on_entire_message_(false)
         , crc_on_payload_(false)
         , corrupted_msg_handler_(nullptr)
+        , streaming_mode_(false)
         {}
 
     /**
@@ -276,6 +277,13 @@ struct asio_service_options {
      */
     std::function< void( std::shared_ptr<buffer>,
                          std::shared_ptr<buffer> ) > corrupted_msg_handler_;
+
+    /**
+     * If `true`,  NuRaft will use streaming mode, which allows it to send
+     * subsequent requests without waiting for the response to previous requests.
+     * The order of responses will be identical to the order of requests.
+     */
+    bool streaming_mode_;
 };
 
 }

include/libnuraft/peer.hxx

@@ -30,6 +30,7 @@ limitations under the License.
 #include "srv_config.hxx"
 
 #include <atomic>
+#include <cassert>
 
 namespace nuraft {
 
@@ -77,6 +78,8 @@ public:
         , lost_by_leader_(false)
         , rsv_msg_(nullptr)
         , rsv_msg_handler_(nullptr)
+        , last_streamed_log_idx_(0)
+        , bytes_in_flight_(0)
         , l_(logger)
     {
         reset_ls_timer();
@@ -221,7 +224,8 @@ public:
 
     void send_req(ptr<peer> myself,
                   ptr<req_msg>& req,
-                  rpc_handler& handler);
+                  rpc_handler& handler,
+                  bool streaming = false);
 
     void shutdown();
 
@@ -307,6 +311,37 @@ public:
     ptr<req_msg> get_rsv_msg() const { return rsv_msg_; }
     rpc_handler get_rsv_msg_handler() const { return rsv_msg_handler_; }
 
+    ulong get_last_streamed_log_idx() {
+        return last_streamed_log_idx_.load();
+    }
+
+    void set_last_streamed_log_idx(ulong expected, ulong idx) {
+        last_streamed_log_idx_.compare_exchange_strong(expected, idx);
+    }
+
+    void reset_stream() {
+        last_streamed_log_idx_.store(0);
+    }
+
+    int64_t get_bytes_in_flight() {
+        return bytes_in_flight_.load();
+    }
+
+    void bytes_in_flight_add(size_t req_size_bytes) {
+        bytes_in_flight_.fetch_add(req_size_bytes);
+    }
+
+    void bytes_in_flight_sub(size_t req_size_bytes) {
+        bytes_in_flight_.fetch_sub(req_size_bytes);
+        assert(bytes_in_flight_ >= 0);
+    }
+
+    void reset_bytes_in_flight() {
+        bytes_in_flight_.store(0);
+    }
+
+    void try_set_free(msg_type type, bool streaming);
+
     bool is_lost() const { return lost_by_leader_; }
     void set_lost() { lost_by_leader_ = true; }
     void set_recovered() { lost_by_leader_ = false; }
@@ -316,6 +351,8 @@ private:
                            ptr<rpc_client> my_rpc_client,
                            ptr<req_msg>& req,
                            ptr<rpc_result>& pending_result,
+                           bool streaming,
+                           size_t req_size_bytes,
                            ptr<resp_msg>& resp,
                            ptr<rpc_exception>& err);
 
@@ -523,6 +560,16 @@ private:
      */
     rpc_handler rsv_msg_handler_;
 
+    /**
+     * Last log index sent in stream mode.
+     */
+    std::atomic<ulong> last_streamed_log_idx_;
+
+    /**
+     * Current bytes of in-flight append entry requests.
+     */
+    std::atomic<int64_t> bytes_in_flight_;
+
     /**
      * Logger instance.
      */

include/libnuraft/raft_params.hxx

@@ -98,6 +98,8 @@ struct raft_params {
         , use_bg_thread_for_snapshot_io_(false)
         , use_full_consensus_among_healthy_members_(false)
         , parallel_log_appending_(false)
+        , max_log_gap_in_stream_(0)
+        , max_bytes_in_flight_in_stream_(0)
         {}
 
     /**
@@ -619,6 +621,22 @@ public:
      * before returning the response.
      */
     bool parallel_log_appending_;
+
+    /**
+     * If non-zero, streaming mode is enabled and `append_entries` requests are
+     * dispatched instantly without awaiting the response from the prior request.
+     *,
+     * The count of logs in-flight will be capped by this value, allowing it
+     * to function as a throttling mechanism, in conjunction with
+     * `max_bytes_in_flight_in_stream_`.
+     */
+    int32 max_log_gap_in_stream_;
+
+    /**
+     * If non-zero, the volume of data in-flight will be restricted to this
+     * specified byte limit. This limitation is effective only in streaming mode.
+     */
+    int64_t max_bytes_in_flight_in_stream_;
 };
 
 }

include/libnuraft/raft_server.hxx

@@ -929,6 +929,7 @@ protected:
     void request_vote(bool force_vote);
     void request_append_entries();
     bool request_append_entries(ptr<peer> p);
+    bool send_request(ptr<peer>& p, ptr<req_msg>& msg, rpc_handler& m_handler, bool streaming = false);
     void handle_peer_resp(ptr<resp_msg>& resp, ptr<rpc_exception>& err);
     void handle_append_entries_resp(resp_msg& resp);
     void handle_install_snapshot_resp(resp_msg& resp);
@@ -946,7 +947,7 @@ protected:
     void handle_join_leave_rpc_err(msg_type t_msg, ptr<peer> p);
     void reset_srv_to_join();
     void reset_srv_to_leave();
-    ptr<req_msg> create_append_entries_req(ptr<peer>& pp);
+    ptr<req_msg> create_append_entries_req(ptr<peer>& pp, ulong custom_last_log_idx = 0);
     ptr<req_msg> create_sync_snapshot_req(ptr<peer>& pp,
                                           ulong last_log_idx,
                                           ulong term,

scripts/test/runtests.sh

@@ -10,3 +10,5 @@ set -e
 ./tests/new_joiner_test --abort-on-failure
 ./tests/failure_test --abort-on-failure
 ./tests/asio_service_test --abort-on-failure
+./tests/asio_service_stream_test --abort-on-failure
+./tests/stream_functional_test --abort-on-failure