[ISSUE #3512]🚀Async High Availability (HA) Service Architecture Enhancement

✨

Author: mxsm

rocketmq-broker/src/broker_runtime.rs

@@ -779,10 +779,11 @@ impl BrokerRuntime {
 
     fn initial_request_pipeline(&mut self) {}
 
-    fn start_basic_service(&mut self) {
+    async fn start_basic_service(&mut self) {
         if let Some(ref mut message_store) = self.inner.message_store {
             message_store
                 .start()
+                .await
                 .unwrap_or_else(|e| panic!("Failed to start message store: {e}"));
         } else {
             panic!("Message store is not initialized");
@@ -890,7 +891,7 @@ impl BrokerRuntime {
         }
 
         self.inner.broker_outer_api.start().await;
-        self.start_basic_service();
+        self.start_basic_service().await;
 
         if !self.inner.is_isolated.load(Ordering::Acquire)
             && !self.inner.message_store_config.enable_dledger_commit_log

rocketmq-store/src/base/message_store.rs

@@ -65,7 +65,7 @@ pub trait MessageStoreInner: Sync + 'static {
     async fn load(&mut self) -> bool;
 
     /// Launch this message store.
-    fn start(&mut self) -> Result<(), StoreError>;
+    async fn start(&mut self) -> Result<(), StoreError>;
 
     /// Shutdown this message store.
     fn shutdown(&mut self);

rocketmq-store/src/config/message_store_config.rs

@@ -277,6 +277,9 @@ mod defaults {
     pub fn ha_send_heartbeat_interval() -> u64 {
         1000 * 5
     }
+    pub fn ha_listen_port() -> usize {
+        10912
+    }
 }
 
 #[derive(Clone, Debug, Deserialize, PartialEq)]
@@ -510,7 +513,7 @@ pub struct MessageStoreConfig {
     #[serde(default)]
     pub message_index_safe: bool,
 
-    #[serde(default)]
+    #[serde(default = "defaults::ha_listen_port")]
     pub ha_listen_port: usize,
 
     #[serde(default = "defaults::ha_send_heartbeat_interval")]
@@ -899,7 +902,7 @@ impl Default for MessageStoreConfig {
             max_index_num: 5000000 * 4,
             max_msgs_num_batch: 64,
             message_index_safe: false,
-            ha_listen_port: 0,
+            ha_listen_port: 10912,
             ha_send_heartbeat_interval: 1000 * 5,
             ha_housekeeping_interval: 1000 * 20,
             ha_transfer_batch_size: 0,

rocketmq-store/src/ha/auto_switch/auto_switch_ha_service.rs

@@ -33,7 +33,7 @@ use crate::store_error::HAResult;
 pub struct AutoSwitchHAService;
 
 impl HAService for AutoSwitchHAService {
-    fn start(&mut self) -> HAResult<()> {
+    async fn start(&mut self) -> HAResult<()> {
         error!("DefaultHAService start not implemented");
         Ok(())
     }

rocketmq-store/src/ha/default_ha_client.rs

@@ -39,6 +39,7 @@ use tracing::warn;
 
 use crate::base::message_store::MessageStore;
 use crate::ha::flow_monitor::FlowMonitor;
+use crate::ha::ha_client::HAClient;
 use crate::ha::ha_connection_state::HAConnectionState;
 use crate::message_store::local_file_message_store::LocalFileMessageStore;
 
@@ -609,9 +610,9 @@ impl DefaultHAClient {
         sleep(Duration::from_secs(5)).await;
     }
 
-    /// Start the HA client service
-    pub async fn start(self: Arc<Self>) -> Result<(), HAClientError> {
-        let self_clone = Arc::clone(&self);
+    // Start the HA client service
+    /*    pub async fn start(self: ArcMut<Self>) -> Result<(), HAClientError> {
+        let self_clone = ArcMut::clone(&self);
         let handle = tokio::spawn(async move {
             self_clone.run_service().await;
         });
@@ -620,7 +621,7 @@ impl DefaultHAClient {
         *service_handle = Some(handle);
 
         Ok(())
-    }
+    }*/
 
     /// Shutdown the HA client
     pub async fn shutdown(self: Arc<Self>) {
@@ -663,6 +664,59 @@ impl DefaultHAClient {
     }
 }
 
+impl HAClient for DefaultHAClient {
+    async fn start(&self) {
+        error!("GeneralHAService does not implement start directly, use specific service");
+    }
+
+    async fn shutdown(&self) {
+        todo!()
+    }
+
+    async fn wakeup(&self) {
+        todo!()
+    }
+
+    async fn update_master_address(&self, new_address: &str) {
+        todo!()
+    }
+
+    async fn update_ha_master_address(&self, new_address: &str) {
+        todo!()
+    }
+
+    fn get_master_address(&self) -> String {
+        todo!()
+    }
+
+    fn get_ha_master_address(&self) -> String {
+        todo!()
+    }
+
+    fn get_last_read_timestamp(&self) -> i64 {
+        todo!()
+    }
+
+    fn get_last_write_timestamp(&self) -> i64 {
+        todo!()
+    }
+
+    fn get_current_state(&self) -> HAConnectionState {
+        todo!()
+    }
+
+    fn change_current_state(&self, ha_connection_state: HAConnectionState) {
+        todo!()
+    }
+
+    async fn close_master(&self) {
+        todo!()
+    }
+
+    fn get_transferred_byte_in_second(&self) -> i64 {
+        todo!()
+    }
+}
 /// Error types
 #[derive(Debug, thiserror::Error)]
 pub enum HAClientError {

rocketmq-store/src/ha/default_ha_connection.rs

@@ -24,6 +24,7 @@ use std::time::Instant;
 use bytes::BufMut;
 use bytes::Bytes;
 use bytes::BytesMut;
+use rocketmq_rust::ArcMut;
 use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
 use tokio::net::tcp::OwnedReadHalf;
@@ -48,7 +49,7 @@ use crate::ha::ha_connection_state::HAConnectionState;
 pub const TRANSFER_HEADER_SIZE: usize = 8 + 4;
 
 pub struct DefaultHAConnection {
-    ha_service: Arc<DefaultHAService>,
+    ha_service: ArcMut<DefaultHAService>,
     socket_stream: Option<TcpStream>,
     client_address: String,
     write_socket_service: Option<WriteSocketService>,
@@ -64,7 +65,7 @@ pub struct DefaultHAConnection {
 impl DefaultHAConnection {
     /// Create a new DefaultHAConnection
     pub async fn new(
-        ha_service: Arc<DefaultHAService>,
+        ha_service: ArcMut<DefaultHAService>,
         socket_stream: TcpStream,
         message_store_config: Arc<MessageStoreConfig>,
     ) -> Result<Self, HAConnectionError> {
@@ -83,7 +84,9 @@ impl DefaultHAConnection {
         let flow_monitor = Arc::new(FlowMonitor::new(message_store_config.clone()));
 
         // Increment connection count
-        // ha_service.increment_connection_count();
+        ha_service
+            .get_connection_count()
+            .fetch_add(1, Ordering::SeqCst);
 
         let (shutdown_sender, shutdown_receiver) = mpsc::channel(1);
 
@@ -116,7 +119,7 @@ impl DefaultHAConnection {
         let read_service = ReadSocketService::new(
             reader,
             self.client_address.clone(),
-            Arc::clone(&self.ha_service),
+            ArcMut::clone(&self.ha_service),
             Arc::clone(&self.current_state),
             self.slave_request_offset.clone(),
             self.slave_ack_offset.clone(),
@@ -128,7 +131,7 @@ impl DefaultHAConnection {
         let write_service = WriteSocketService::new(
             writer,
             self.client_address.clone(),
-            Arc::clone(&self.ha_service),
+            ArcMut::clone(&self.ha_service),
             Arc::clone(&self.current_state),
             self.slave_request_offset.clone(),
             Arc::clone(&self.flow_monitor),
@@ -228,7 +231,7 @@ const REPORT_HEADER_SIZE: usize = 8;
 pub struct ReadSocketService {
     reader: Option<OwnedReadHalf>,
     client_address: String,
-    ha_service: Arc<DefaultHAService>,
+    ha_service: ArcMut<DefaultHAService>,
     current_state: Arc<RwLock<HAConnectionState>>,
     slave_request_offset: Arc<AtomicI64>,
     slave_ack_offset: Arc<AtomicI64>,
@@ -241,7 +244,7 @@ impl ReadSocketService {
     pub async fn new(
         reader: OwnedReadHalf,
         client_address: String,
-        ha_service: Arc<DefaultHAService>,
+        ha_service: ArcMut<DefaultHAService>,
         current_state: Arc<RwLock<HAConnectionState>>,
         slave_request_offset: Arc<AtomicI64>,
         slave_ack_offset: Arc<AtomicI64>,
@@ -265,7 +268,7 @@ impl ReadSocketService {
     pub async fn start(&mut self) -> Result<(), HAConnectionError> {
         let socket_stream = self.reader.take();
         let client_address = self.client_address.clone();
-        let ha_service = Arc::clone(&self.ha_service);
+        let ha_service = ArcMut::clone(&self.ha_service);
         let current_state = Arc::clone(&self.current_state);
         let slave_request_offset = self.slave_request_offset.clone();
         let slave_ack_offset = self.slave_ack_offset.clone();
@@ -291,7 +294,7 @@ impl ReadSocketService {
     async fn run_service(
         mut socket_stream: Option<OwnedReadHalf>,
         client_address: String,
-        ha_service: Arc<DefaultHAService>,
+        ha_service: ArcMut<DefaultHAService>,
         current_state: Arc<RwLock<HAConnectionState>>,
         slave_request_offset: Arc<AtomicI64>,
         slave_ack_offset: Arc<AtomicI64>,
@@ -404,7 +407,7 @@ impl ReadSocketService {
 pub struct WriteSocketService {
     writer: Option<OwnedWriteHalf>,
     client_address: String,
-    ha_service: Arc<DefaultHAService>,
+    ha_service: ArcMut<DefaultHAService>,
     current_state: Arc<RwLock<HAConnectionState>>,
     slave_request_offset: Arc<AtomicI64>,
     flow_monitor: Arc<FlowMonitor>,
@@ -417,7 +420,7 @@ impl WriteSocketService {
     pub async fn new(
         writer: OwnedWriteHalf,
         client_address: String,
-        ha_service: Arc<DefaultHAService>,
+        ha_service: ArcMut<DefaultHAService>,
         current_state: Arc<RwLock<HAConnectionState>>,
         slave_request_offset: Arc<AtomicI64>,
         flow_monitor: Arc<FlowMonitor>,
@@ -439,7 +442,7 @@ impl WriteSocketService {
     pub async fn start(&mut self) -> Result<(), HAConnectionError> {
         let socket_stream = self.writer.take();
         let client_address = self.client_address.clone();
-        let ha_service = Arc::clone(&self.ha_service);
+        let ha_service = ArcMut::clone(&self.ha_service);
         let current_state = Arc::clone(&self.current_state);
         let slave_request_offset = self.slave_request_offset.clone();
         let flow_monitor = Arc::clone(&self.flow_monitor);
@@ -467,7 +470,7 @@ impl WriteSocketService {
     async fn run_service(
         mut socket_stream: Option<OwnedWriteHalf>,
         client_address: String,
-        ha_service: Arc<DefaultHAService>,
+        ha_service: ArcMut<DefaultHAService>,
         current_state: Arc<RwLock<HAConnectionState>>,
         slave_request_offset: Arc<AtomicI64>,
         flow_monitor: Arc<FlowMonitor>,