From 99a7485f6a265b2b530860a9b5ceaf3d99f8c2b2 Mon Sep 17 00:00:00 2001
From: Hengfei Yang <hengfei.yang@gmail.com>
Date: Wed, 5 Feb 2025 18:28:44 +0800
Subject: [PATCH] fix: wal rorate in concurrency (#5881)

---
 src/ingester/src/writer.rs | 27 ++++++++++++++-------------
 1 file changed, 14 insertions(+), 13 deletions(-)

diff --git a/src/ingester/src/writer.rs b/src/ingester/src/writer.rs
index 4fa9317873..55d6241299 100644
--- a/src/ingester/src/writer.rs
+++ b/src/ingester/src/writer.rs
@@ -31,7 +31,7 @@ use config::{
 use hashbrown::HashSet;
 use once_cell::sync::Lazy;
 use snafu::ResultExt;
-use tokio::sync::RwLock;
+use tokio::sync::{Mutex, RwLock};
 use wal::Writer as WalWriter;
 
 use crate::{
@@ -56,7 +56,7 @@ static WRITERS: Lazy<Vec<RwMap<WriterKey, Arc<Writer>>>> = Lazy::new(|| {
 pub struct Writer {
     idx: usize,
     key: WriterKey,
-    wal: Arc<RwLock<WalWriter>>,
+    wal: Arc<Mutex<WalWriter>>,
     memtable: Arc<RwLock<MemTable>>,
     next_seq: AtomicU64,
     created_at: AtomicI64,
@@ -182,7 +182,7 @@ impl Writer {
         Self {
             idx,
             key: key.clone(),
-            wal: Arc::new(RwLock::new(
+            wal: Arc::new(Mutex::new(
                 WalWriter::new(
                     wal_dir,
                     &key.org_id,
@@ -243,7 +243,7 @@ impl Writer {
 
         // write into wal
         let start = std::time::Instant::now();
-        let mut wal = self.wal.write().await;
+        let mut wal = self.wal.lock().await;
         let wal_lock_time = start.elapsed().as_millis() as f64;
         metrics::INGEST_WAL_LOCK_TIME
             .with_label_values(&[&self.key.org_id])
@@ -273,7 +273,7 @@ impl Writer {
 
         // check fsync
         if fsync {
-            let mut wal = self.wal.write().await;
+            let mut wal = self.wal.lock().await;
             wal.sync().context(WalSnafu)?;
             drop(wal);
         }
@@ -283,7 +283,14 @@ impl Writer {
 
     // rotate is used to rotate the wal and memtable if the size exceeds the threshold
     async fn rotate(&self, entry_bytes_size: usize, entry_batch_size: usize) -> Result<()> {
-        if !self.check_wal_threshold(self.wal.read().await.size(), entry_bytes_size)
+        let start = std::time::Instant::now();
+        let mut wal = self.wal.lock().await;
+        let wal_lock_time = start.elapsed().as_millis() as f64;
+        metrics::INGEST_WAL_LOCK_TIME
+            .with_label_values(&[&self.key.org_id])
+            .observe(wal_lock_time);
+
+        if !self.check_wal_threshold(wal.size(), entry_bytes_size)
             && !self.check_mem_threshold(self.memtable.read().await.size(), entry_batch_size)
         {
             return Ok(());
@@ -311,12 +318,6 @@ impl Writer {
             cfg.limit.wal_write_buffer_size,
         )
         .context(WalSnafu)?;
-        let start = std::time::Instant::now();
-        let mut wal = self.wal.write().await;
-        let wal_lock_time = start.elapsed().as_millis() as f64;
-        metrics::INGEST_WAL_LOCK_TIME
-            .with_label_values(&[&self.key.org_id])
-            .observe(wal_lock_time);
         wal.sync().context(WalSnafu)?; // sync wal before rotation
         let old_wal = std::mem::replace(&mut *wal, new_wal);
         drop(wal);
@@ -348,7 +349,7 @@ impl Writer {
 
     pub async fn close(&self) -> Result<()> {
         // rotation wal
-        let mut wal = self.wal.write().await;
+        let mut wal = self.wal.lock().await;
         wal.sync().context(WalSnafu)?;
         let path = wal.path().clone();
         drop(wal);