You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: raftlog_gc.rs
    path of file: ./repos/tikv/components/raftstore/src/store/worker
    
    the code of the file until where you have to start completion: // Copyright 2016 TiKV Project Authors. Licensed under Apache-2.0.

use std::{
    error::Error as StdError,
    fmt::{self, Display, Formatter},
};

use engine_traits::{Engines, KvEngine, RaftEngine};
use file_system::{IoType, WithIoType};
use thiserror::Error;
use tikv_util::{
    box_try, debug, error,
    time::{Duration, Instant},
    warn,
    worker::{Runnable, RunnableWithTimer},
};

use crate::store::worker::metrics::*;

const MAX_GC_REGION_BATCH: usize = 512;
const MAX_REGION_NORMAL_GC_LOG_NUMBER: u64 = 10240;

pub struct Task {
    region_id: u64,
    start_idx: u64,
    end_idx: u64,
    flush: bool,
    cb: Option<Box<dyn FnOnce() + Send>>,
}

impl Task {
    pub fn gc(region_id: u64, start: u64, end: u64) -> Self {
        Task {
            region_id,
            start_idx: start,
            end_idx: end,
            flush: false,
            cb: None,
        }
    }

    pub fn flush(mut self) -> Self {
        self.flush = true;
        self
    }

    pub fn when_done(mut self, callback: impl FnOnce() + Send + 'static) -> Self {
        self.cb = Some(Box::new(callback));
        self
    }
}

impl Display for Task {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "GC Raft Logs [region: {}, from: {}, to: {}, has_cb: {}]",
            self.region_id,
            self.start_idx,
            self.end_idx,
            self.cb.is_some()
        )
    }
}

#[derive(Debug, Error)]
enum Error {
    #[error("raftlog gc failed {0:?}")]
    Other(#[from] Box<dyn StdError + Sync + Send>),
}

pub struct Runner<EK: KvEngine, ER: RaftEngine> {
    tasks: Vec<Task>,
    engines: Engines<EK, ER>,
    compact_sync_interval: Duration,
}

impl<EK: KvEngine, ER: RaftEngine> Runner<EK, ER> {
    pub fn new(engines: Engines<EK, ER>, compact_log_interval: Duration) -> Runner<EK, ER> {
 
    
    fn flush(&mut self) {
        if self.tasks.is_empty() {
            return;
        }
        fail::fail_point!("worker_gc_raft_log_flush");
        // Sync wal of kv_db to make sure the data before apply_index has been persisted
        // to disk.
        let start = Instant::now();
        self.engines.kv.sync().unwrap_or_else(|e| {
            panic!("failed to sync kv_engine in raft_log_gc: {:?}", e);
        });
        RAFT_LOG_GC_KV_SYNC_DURATION_HISTOGRAM.observe(start.saturating_elapsed_secs());

        let tasks = std::mem::take(&mut self.tasks);
        let mut cbs = Vec::new();
        let mut batch = self.engines.raft.log_batch(tasks.len());
        let start = Instant::now();
        for t in tasks {
            debug!("gc raft log"; "region_id" => t.region_id, "start_index" => t.start_idx, "end_index" => t.end_idx);
            if let Some(cb) = t.cb {
                cbs.push(cb);
            }
            if t.start_idx == t.end_idx {
                // It's only for flush.
                continue;
            }
            if t.start_idx == 0 {
                RAFT_LOG_GC_SEEK_OPERATIONS.inc();
            } else if t.end_idx > t.start_idx + MAX_REGION_NORMAL_GC_LOG_NUMBER {
                warn!(
                    "gc raft log with a large range";
                    "region_id" => t.region_id,
                    "start_index" => t.start_idx,
                    "end_index" => t.end_idx,
                );
            }
            if let Err(e) = self
                .engines
                .raft
                .gc(t.region_id, t.start_idx, t.end_idx, &mut batch)
            {
                error!("failed to gc"; "err" => %e);
                RAFT_LOG_GC_FAILED.inc();
            }
        }
        if let Err(e) = self.raft_log_gc(batch) {
            error!("failed to write gc task"; "err" => %e);
            RAFT_LOG_GC_FAILED.inc();
        }
        RAFT_LOG_GC_WRITE_DURATION_HISTOGRAM.observe(start.saturating_elapsed_secs());
        for cb in cbs {
            cb();
        }
    }