starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-06 00:09:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: wal rorate in concurrency (#5881)

Browse Source
This commit is contained in:
Hengfei Yang
2025-02-05 18:28:44 +08:00
committed by GitHub
parent 0b682773c5
commit 99a7485f6a
1 changed files with 14 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ingester/src/writer.rs
+14 -13
View File
@@ -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);