pub fn recover<M: ReplayMachine, FA: Factory<M>>(
        &mut self,
        machine_factory: &FA,
    ) -> Result<(M, M)> {
        let threads = self.cfg.recovery_threads;
        let pool = rayon::ThreadPoolBuilder::new()
            .num_threads(threads)
            .build()
            .unwrap();
        let (append_concurrency, rewrite_concurrency) =
            match (self.append_files.len(), self.rewrite_files.len()) {
                (a, b) if a > 0 && b > 0 => {
                    let a_threads = std::cmp::max(1, threads * a / (a + b));
                    let b_threads = std::cmp::max(1, threads.saturating_sub(a_threads));
                    (a_threads, b_threads)
                }
                _ => (threads, threads),
            };

        let append_recovery_cfg = RecoveryConfig {
            queue: LogQueue::Append,
            mode: self.cfg.recovery_mode,
            concurrency: append_concurrency,
            read_block_size: self.cfg.recovery_read_block_size.0,
        };
        let rewrite_recovery_cfg = RecoveryConfig {
            queue: LogQueue::Rewrite,
            concurrency: rewrite_concurrency,
            ..append_recovery_cfg
        };
        let append_files = &mut self.append_files;
        let rewrite_files = &mut self.rewrite_files;
        let file_system = self.file_system.clone();
        // As the `recover_queue` would update the `Version` of each log file in
        // `apend_files` and `rewrite_files`, we re-design the implementation on
        // `recover_queue` to make it compatiable to concurrent processing
        // with ThreadPool.
        let (append, rewrite) = pool.join(
            || {
                DualPipesBuilder::recover_queue_imp(
                    file_system.clone(),
                    append_recovery_cfg,
                    append_files,
                    machine_factory,
                )
            },
            || {
                DualPipesBuilder::recover_queue_imp(
                    file_system.clone(),
                    rewrite_recovery_cfg,
                    rewrite_files,
                    machine_factory,
                )
            },
        );
        Ok((append?, rewrite?))
    }
/// Reads through log items in all available log files, and replays them to
/// specific [`ReplayMachine`]s that can be constructed via
/// `machine_factory`.
Reads through log items in all available log files, and replays them to
specific [`ReplayMachine`]s that can be constructed via
`machine_factory`.
Reads through log items in all available log files and replays them to specific [ ` ReplayMachine ` ] s that can be constructed via ` machine_factory ` .
pub fn recover<M: ReplayMachine, FA: Factory<M>>(
        &mut self,
        machine_factory: &FA,
    ) -> Result<(M, M)> {
        let threads = self.cfg.recovery_threads;
        let pool = rayon::ThreadPoolBuilder::new()
            .num_threads(threads)
            .build()
            .unwrap();
        let (append_concurrency, rewrite_concurrency) =
            match (self.append_files.len(), self.rewrite_files.len()) {
                (a, b) if a > 0 && b > 0 => {
                    let a_threads = std::cmp::max(1, threads * a / (a + b));
                    let b_threads = std::cmp::max(1, threads.saturating_sub(a_threads));
                    (a_threads, b_threads)
                }
                _ => (threads, threads),
            };
        let append_recovery_cfg = RecoveryConfig {
            queue: LogQueue::Append,
            mode: self.cfg.recovery_mode,
            concurrency: append_concurrency,
            read_block_size: self.cfg.recovery_read_block_size.0,
        };
        let rewrite_recovery_cfg = RecoveryConfig {
            queue: LogQueue::Rewrite,
            concurrency: rewrite_concurrency,
            ..append_recovery_cfg
        };
        let append_files = &mut self.append_files;
        let rewrite_files = &mut self.rewrite_files;
        let file_system = self.file_system.clone();
        // As the `recover_queue` would update the `Version` of each log file in
        // `apend_files` and `rewrite_files`, we re-design the implementation on
        // `recover_queue` to make it compatiable to concurrent processing
        // with ThreadPool.
        let (append, rewrite) = pool.join(
            || {
                DualPipesBuilder::recover_queue_imp(
                    file_system.clone(),
                    append_recovery_cfg,
                    append_files,
                    machine_factory,
                )
            },
            || {
                DualPipesBuilder::recover_queue_imp(
                    file_system.clone(),
                    rewrite_recovery_cfg,
                    rewrite_files,
                    machine_factory,
                )
            },
        );
        Ok((append?, rewrite?))
    }
pub fn recover < M : ReplayMachine , FA : Factory < M > > ( & mut self , machine_factory : & FA , ) -> Result < ( M , M ) > { let threads = self . cfg . recovery_threads ; let pool = rayon :: ThreadPoolBuilder :: new ( ) . num_threads ( threads ) . build ( ) . unwrap ( ) ; let ( append_concurrency , rewrite_concurrency ) = match ( self . append_files . len ( ) , self . rewrite_files . len ( ) ) { ( a , b ) if a > 0 && b > 0 => { let a_threads = std :: cmp :: max ( 1 , threads * a / ( a + b ) ) ; let b_threads = std :: cmp :: max ( 1 , threads . saturating_sub ( a_threads ) ) ; ( a_threads , b_threads ) } _ => ( threads , threads ) , } ; let append_recovery_cfg = RecoveryConfig { queue : LogQueue :: Append , mode : self . cfg . recovery_mode , concurrency : append_concurrency , read_block_size : self . cfg . recovery_read_block_size . 0 , } ; let rewrite_recovery_cfg = RecoveryConfig { queue : LogQueue :: Rewrite , concurrency : rewrite_concurrency , .. append_recovery_cfg } ; let append_files = & mut self . append_files ; let rewrite_files = & mut self . rewrite_files ; let file_system = self . file_system . clone ( ) ; // As the `recover_queue` would update the `Version` of each log file in // `apend_files` and `rewrite_files`, we re-design the implementation on // `recover_queue` to make it compatiable to concurrent processing // with ThreadPool. let ( append , rewrite ) = pool . join ( | | { DualPipesBuilder :: recover_queue_imp ( file_system . clone ( ) , append_recovery_cfg , append_files , machine_factory , ) } , | | { DualPipesBuilder :: recover_queue_imp ( file_system . clone ( ) , rewrite_recovery_cfg , rewrite_files , machine_factory , ) } , ) ; Ok ( ( append ? , rewrite ? ) ) }
Reads through log items in all available log files, and replays them to
specific [`ReplayMachine`]s that can be constructed via
`machine_factory`.
Reads through log items in all available log files and replays them to specific [ ` ReplayMachine ` ] s that can be constructed via ` machine_factory ` .
&mut self
machine_factory
&FA