[ISSUE #375]💥Implement IndexService💥

Author: mxsm

rocketmq-common/src/utils/file_utils.rs

@@ -21,8 +21,11 @@ use std::{
     path::{Path, PathBuf},
 };
 
+use parking_lot::Mutex;
 use tracing::warn;
 
+static LOCK: Mutex<()> = Mutex::new(());
+
 pub fn file_to_string(file_name: &str) -> Result<String, io::Error> {
     if !PathBuf::from(file_name).exists() {
         warn!("file not exist:{}", file_name);
@@ -47,6 +50,8 @@ pub fn file_to_string_impl(file: &File) -> Result<String, io::Error> {
 }
 
 pub fn string_to_file(str_content: &str, file_name: &str) -> io::Result<()> {
+    let lock = LOCK.lock();
+
     let bak_file = format!("{}.bak", file_name);
 
     // Read previous content and create a backup
@@ -56,7 +61,7 @@ pub fn string_to_file(str_content: &str, file_name: &str) -> io::Result<()> {
 
     // Write new content to the file
     string_to_file_not_safe(str_content, file_name)?;
-
+    drop(lock);
     Ok(())
 }
 

rocketmq-store/src/config/message_store_config.rs

@@ -105,8 +105,8 @@ pub struct MessageStoreConfig {
     pub max_transfer_count_on_message_in_disk: usize,
     pub access_message_in_memory_max_ratio: usize,
     pub message_index_enable: bool,
-    pub max_hash_slot_num: usize,
-    pub max_index_num: usize,
+    pub max_hash_slot_num: u32,
+    pub max_index_num: u32,
     pub max_msgs_num_batch: usize,
     pub message_index_safe: bool,
     pub ha_listen_port: usize,
@@ -287,10 +287,10 @@ impl Default for MessageStoreConfig {
             max_transfer_bytes_on_message_in_disk: 0,
             max_transfer_count_on_message_in_disk: 0,
             access_message_in_memory_max_ratio: 0,
-            message_index_enable: false,
-            max_hash_slot_num: 0,
-            max_index_num: 0,
-            max_msgs_num_batch: 0,
+            message_index_enable: true,
+            max_hash_slot_num: 5000000,
+            max_index_num: 5000000 * 4,
+            max_msgs_num_batch: 64,
             message_index_safe: false,
             ha_listen_port: 0,
             ha_send_heartbeat_interval: 0,

rocketmq-store/src/index.rs

@@ -19,3 +19,4 @@ pub mod index_dispatch;
 pub mod index_file;
 pub mod index_header;
 pub mod index_service;
+pub mod query_offset_result;

rocketmq-store/src/index/index_file.rs

@@ -26,6 +26,20 @@ use crate::{
 };
 
 const HASH_SLOT_SIZE: usize = 4;
+
+/**
+ * Each index's store unit. Format:
+ * <pre>
+ * ┌───────────────┬───────────────────────────────┬───────────────┬───────────────┐
+ * │ Key HashCode  │        Physical Offset        │   Time Diff   │ Next Index Pos│
+ * │   (4 Bytes)   │          (8 Bytes)            │   (4 Bytes)   │   (4 Bytes)   │
+ * ├───────────────┴───────────────────────────────┴───────────────┴───────────────┤
+ * │                                 Index Store Unit                              │
+ * │                                                                               │
+ * </pre>
+ * Each index's store unit. Size:
+ * Key HashCode(4) + Physical Offset(8) + Time Diff(4) + Next Index Pos(4) = 20 Bytes
+ */
 const INDEX_SIZE: usize = 20;
 const INVALID_INDEX: i32 = 0;
 
@@ -37,6 +51,12 @@ pub struct IndexFile {
     index_header: IndexHeader,
 }
 
+impl PartialEq for IndexFile {
+    fn eq(&self, other: &Self) -> bool {
+        std::ptr::eq(self as *const IndexFile, other as *const IndexFile)
+    }
+}
+
 impl IndexFile {
     pub fn new(
         file_name: &str,

rocketmq-store/src/index/index_header.rs

@@ -58,7 +58,7 @@ impl IndexHeader {
     }
 
     pub fn load(&self) {
-        let mut buffer = self.mapped_file.get_data(0, INDEX_HEADER_SIZE).unwrap();
+        let mut buffer = self.mapped_file.get_bytes(0, INDEX_HEADER_SIZE).unwrap();
         self.begin_timestamp
             .store(buffer.get_i64(), Ordering::SeqCst);
         self.end_timestamp.store(buffer.get_i64(), Ordering::SeqCst);