chore: reflush memory tables after flush failed (#891)

Author: MichaelLeeHZ

analytic_engine/src/instance/flush_compaction.rs

@@ -96,11 +96,16 @@ pub enum Error {
     WriteSst { path: String, source: GenericError },
 
     #[snafu(display(
-        "Background flush failed, cannot write more data, err:{}.\nBacktrace:\n{}",
+        "Background flush failed, cannot write more data, retry_count:{}, err:{}.\nBacktrace:\n{}",
+        retry_count,
         msg,
         backtrace
     ))]
-    BackgroundFlushFailed { msg: String, backtrace: Backtrace },
+    BackgroundFlushFailed {
+        msg: String,
+        retry_count: usize,
+        backtrace: Backtrace,
+    },
 
     #[snafu(display("Failed to build merge iterator, table:{}, err:{}", table, source))]
     BuildMergeIterator {
@@ -144,6 +149,10 @@ pub struct TableFlushOptions {
     ///
     /// If it is [None], no compaction will be scheduled.
     pub compact_after_flush: Option<CompactionSchedulerRef>,
+    /// Max retry limit After flush failed
+    ///
+    /// Default is 0
+    pub max_retry_flush_limit: usize,
 }
 
 impl fmt::Debug for TableFlushOptions {
@@ -286,7 +295,7 @@ impl Flusher {
 
         // TODO: The immediate compaction after flush is not a good idea because it may
         // block on the write procedure.
-        if let Some(compaction_scheduler) = opts.compact_after_flush {
+        if let Some(compaction_scheduler) = opts.compact_after_flush.clone() {
             // Schedule compaction if flush completed successfully.
             let compact_req = TableCompactionRequest::no_waiter(table_data.clone());
             let on_flush_success = async move {
@@ -301,7 +310,7 @@ impl Flusher {
                     flush_job,
                     on_flush_success,
                     block_on,
-                    opts.res_sender,
+                    opts,
                     &self.runtime,
                     &table_data.metrics,
                 )
@@ -312,7 +321,7 @@ impl Flusher {
                     flush_job,
                     async {},
                     block_on,
-                    opts.res_sender,
+                    opts,
                     &self.runtime,
                     &table_data.metrics,
                 )
@@ -430,6 +439,7 @@ impl FlushTask {
                 meta_edit: MetaEdit::Update(meta_update),
             }
         };
+        // Update manifest and remove immutable memtables
         self.space_store
             .manifest
             .apply_edit(edit_req)

analytic_engine/src/instance/mod.rs

@@ -152,6 +152,8 @@ pub struct Instance {
     pub(crate) replay_batch_size: usize,
     /// Write sst max buffer size
     pub(crate) write_sst_max_buffer_size: usize,
+    /// Max retry limit to flush memtables
+    pub(crate) max_retry_flush_limit: usize,
     /// Max bytes per write batch
     pub(crate) max_bytes_per_write_batch: Option<usize>,
     /// Options for scanning sst
@@ -192,6 +194,7 @@ impl Instance {
             } else {
                 None
             },
+            max_retry_flush_limit: 0,
         };
 
         let flusher = self.make_flusher();
@@ -275,6 +278,11 @@ impl Instance {
             write_sst_max_buffer_size: self.write_sst_max_buffer_size,
         }
     }
+
+    #[inline]
+    fn max_retry_flush_limit(&self) -> usize {
+        self.max_retry_flush_limit
+    }
 }
 
 /// Instance reference

analytic_engine/src/instance/open.rs

@@ -125,6 +125,7 @@ impl Instance {
             space_write_buffer_size: ctx.config.space_write_buffer_size,
             replay_batch_size: ctx.config.replay_batch_size,
             write_sst_max_buffer_size: ctx.config.write_sst_max_buffer_size.as_byte() as usize,
+            max_retry_flush_limit: ctx.config.max_retry_flush_limit,
             max_bytes_per_write_batch: ctx
                 .config
                 .max_bytes_per_write_batch
@@ -326,6 +327,7 @@ impl Instance {
                         let opts = TableFlushOptions {
                             res_sender: None,
                             compact_after_flush: None,
+                            max_retry_flush_limit: self.max_retry_flush_limit,
                         };
                         let flusher = self.make_flusher();
                         let flush_scheduler = serial_exec.flush_scheduler();

analytic_engine/src/instance/serial_executor.rs

@@ -1,21 +1,25 @@
 // Copyright 2023 CeresDB Project Authors. Licensed under Apache-2.0.
 
 use std::{
-    sync::{Arc, Mutex},
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc, Mutex,
+    },
     time::Instant,
 };
 
 use common_util::{runtime::Runtime, time::InstantExt};
 use futures::Future;
-use log::error;
+use log::{error, warn};
 use table_engine::table::TableId;
 use tokio::sync::{
     oneshot,
     watch::{self, Receiver, Sender},
 };
 
+use super::flush_compaction::{BackgroundFlushFailed, TableFlushOptions};
 use crate::{
-    instance::flush_compaction::{BackgroundFlushFailed, Other, Result},
+    instance::flush_compaction::{Other, Result},
     table::metrics::Metrics,
 };
 
@@ -34,6 +38,26 @@ type ScheduleSyncRef = Arc<ScheduleSync>;
 struct ScheduleSync {
     state: Mutex<FlushState>,
     notifier: Sender<()>,
+    continuous_flush_failure_count: AtomicUsize,
+}
+
+impl ScheduleSync {
+    #[inline]
+    pub fn should_retry_flush(&self, max_retry_limit: usize) -> bool {
+        self.continuous_flush_failure_count.load(Ordering::Relaxed) < max_retry_limit
+    }
+
+    #[inline]
+    pub fn reset_flush_failure_count(&self) {
+        self.continuous_flush_failure_count
+            .store(0, Ordering::Relaxed);
+    }
+
+    #[inline]
+    pub fn inc_flush_failure_count(&self) {
+        self.continuous_flush_failure_count
+            .fetch_add(1, Ordering::Relaxed);
+    }
 }
 
 pub struct TableFlushScheduler {
@@ -47,6 +71,7 @@ impl Default for TableFlushScheduler {
         let schedule_sync = ScheduleSync {
             state: Mutex::new(FlushState::Ready),
             notifier: tx,
+            continuous_flush_failure_count: AtomicUsize::new(0),
         };
         Self {
             schedule_sync: Arc::new(schedule_sync),
@@ -105,7 +130,7 @@ impl TableFlushScheduler {
         flush_job: F,
         on_flush_success: T,
         block_on_write_thread: bool,
-        res_sender: Option<oneshot::Sender<Result<()>>>,
+        opts: TableFlushOptions,
         runtime: &Runtime,
         metrics: &Metrics,
     ) -> Result<()>
@@ -131,7 +156,21 @@ impl TableFlushScheduler {
                     }
                     FlushState::Flushing => (),
                     FlushState::Failed { err_msg } => {
-                        return BackgroundFlushFailed { msg: err_msg }.fail();
+                        if self
+                            .schedule_sync
+                            .should_retry_flush(opts.max_retry_flush_limit)
+                        {
+                            warn!("Re-flush memory tables after background flush failed:{err_msg}");
+                            // Mark the worker is flushing.
+                            *flush_state = FlushState::Flushing;
+                            break;
+                        } else {
+                            return BackgroundFlushFailed {
+                                msg: err_msg,
+                                retry_count: opts.max_retry_flush_limit,
+                            }
+                            .fail();
+                        }
                     }
                 }
 
@@ -164,7 +203,7 @@ impl TableFlushScheduler {
             if flush_res.is_ok() {
                 on_flush_success.await;
             }
-            send_flush_result(res_sender, flush_res);
+            send_flush_result(opts.res_sender, flush_res);
         };
 
         if block_on_write_thread {
@@ -182,9 +221,11 @@ fn on_flush_finished(schedule_sync: ScheduleSyncRef, res: &Result<()>) {
         let mut flush_state = schedule_sync.state.lock().unwrap();
         match res {
             Ok(()) => {
+                schedule_sync.reset_flush_failure_count();
                 *flush_state = FlushState::Ready;
             }
             Err(e) => {
+                schedule_sync.inc_flush_failure_count();
                 let err_msg = e.to_string();
                 *flush_state = FlushState::Failed { err_msg };
             }

analytic_engine/src/instance/write.rs

@@ -590,7 +590,11 @@ impl<'a> Writer<'a> {
     /// acquired in advance. And in order to avoid deadlock, we should not wait
     /// for the lock.
     async fn handle_memtable_flush(&mut self, table_data: &TableDataRef) -> Result<()> {
-        let opts = TableFlushOptions::default();
+        let opts = TableFlushOptions {
+            res_sender: None,
+            compact_after_flush: None,
+            max_retry_flush_limit: self.instance.max_retry_flush_limit(),
+        };
         let flusher = self.instance.make_flusher();
         if table_data.id == self.table_data.id {
             let flush_scheduler = self.serial_exec.flush_scheduler();

analytic_engine/src/lib.rs

@@ -83,6 +83,8 @@ pub struct Config {
     pub sst_background_read_parallelism: usize,
     /// Max buffer size for writing sst
     pub write_sst_max_buffer_size: ReadableSize,
+    /// Max retry limit After flush failed
+    pub max_retry_flush_limit: usize,
     /// Max bytes per write batch.
     ///
     /// If this is set, the atomicity of write request will be broken.
@@ -122,6 +124,7 @@ impl Default for Config {
             sst_background_read_parallelism: 8,
             scan_max_record_batches_in_flight: 1024,
             write_sst_max_buffer_size: ReadableSize::mb(10),
+            max_retry_flush_limit: 0,
             max_bytes_per_write_batch: None,
             wal: WalStorageConfig::RocksDB(Box::default()),
             remote_engine_client: remote_engine_client::config::Config::default(),