pub fn write(&self, log_batch: &mut LogBatch, mut sync: bool) -> Result<usize> {
        let start = Instant::now();
        let len = log_batch.finish_populate(self.cfg.batch_compression_threshold.0 as usize)?;
        let block_handle = {
            let mut writer = Writer::new(log_batch, sync, start);
            if let Some(mut group) = self.write_barrier.enter(&mut writer) {
                let now = Instant::now();
                let _t = StopWatch::new_with(&ENGINE_WRITE_LEADER_DURATION_HISTOGRAM, now);
                for writer in group.iter_mut() {
                    ENGINE_WRITE_PREPROCESS_DURATION_HISTOGRAM.observe(
                        now.saturating_duration_since(writer.start_time)
                            .as_secs_f64(),
                    );
                    sync |= writer.sync;
                    let log_batch = writer.get_payload();
                    let res = if !log_batch.is_empty() {
                        // @TODO(lucasliang): bind `Version` to each `LogBatch`
                        self.pipe_log
                            .append(LogQueue::Append, log_batch.encoded_bytes())
                    } else {
                        // TODO(tabokie): use Option<FileBlockHandle> instead.
                        Ok(FileBlockHandle {
                            id: FileId::new(LogQueue::Append, 0),
                            offset: 0,
                            len: 0,
                        })
                    };
                    writer.set_output(res);
                }
                if let Err(e) = self.pipe_log.maybe_sync(LogQueue::Append, sync) {
                    panic!(
                        "Cannot sync {:?} queue due to IO error: {}",
                        LogQueue::Append,
                        e
                    );
                }
            }
            writer.finish()?
        };

        let mut now = Instant::now();
        if len > 0 {
            log_batch.finish_write(block_handle);
            self.memtables.apply_append_writes(log_batch.drain());
            for listener in &self.listeners {
                listener.post_apply_memtables(block_handle.id);
            }
            let end = Instant::now();
            ENGINE_WRITE_APPLY_DURATION_HISTOGRAM
                .observe(end.saturating_duration_since(now).as_secs_f64());
            now = end;
        }
        ENGINE_WRITE_DURATION_HISTOGRAM.observe(now.saturating_duration_since(start).as_secs_f64());
        ENGINE_WRITE_SIZE_HISTOGRAM.observe(len as f64);
        Ok(len)
    }
/// Writes the content of `log_batch` into the engine and returns written
/// bytes. If `sync` is true, the write will be followed by a call to
/// `fdatasync` on the log file.
Writes the content of `log_batch` into the engine and returns written
bytes. If `sync` is true, the write will be followed by a call to
`fdatasync` on the log file.
Writes the content of ` log_batch ` into the engine and returns written bytes . If ` sync ` is true the write will be followed by a call to ` fdatasync ` on the log file .
pub fn write(&self, log_batch: &mut LogBatch, mut sync: bool) -> Result<usize> {
        let start = Instant::now();
        let len = log_batch.finish_populate(self.cfg.batch_compression_threshold.0 as usize)?;
        let block_handle = {
            let mut writer = Writer::new(log_batch, sync, start);
            if let Some(mut group) = self.write_barrier.enter(&mut writer) {
                let now = Instant::now();
                let _t = StopWatch::new_with(&ENGINE_WRITE_LEADER_DURATION_HISTOGRAM, now);
                for writer in group.iter_mut() {
                    ENGINE_WRITE_PREPROCESS_DURATION_HISTOGRAM.observe(
                        now.saturating_duration_since(writer.start_time)
                            .as_secs_f64(),
                    );
                    sync |= writer.sync;
                    let log_batch = writer.get_payload();
                    let res = if !log_batch.is_empty() {
                        // @TODO(lucasliang): bind `Version` to each `LogBatch`
                        self.pipe_log
                            .append(LogQueue::Append, log_batch.encoded_bytes())
                    } else {
                        // TODO(tabokie): use Option<FileBlockHandle> instead.
                        Ok(FileBlockHandle {
                            id: FileId::new(LogQueue::Append, 0),
                            offset: 0,
                            len: 0,
                        })
                    };
                    writer.set_output(res);
                }
                if let Err(e) = self.pipe_log.maybe_sync(LogQueue::Append, sync) {
                    panic!(
                        "Cannot sync {:?} queue due to IO error: {}",
                        LogQueue::Append,
                        e
                    );
                }
            }
            writer.finish()?
        };
        let mut now = Instant::now();
        if len > 0 {
            log_batch.finish_write(block_handle);
            self.memtables.apply_append_writes(log_batch.drain());
            for listener in &self.listeners {
                listener.post_apply_memtables(block_handle.id);
            }
            let end = Instant::now();
            ENGINE_WRITE_APPLY_DURATION_HISTOGRAM
                .observe(end.saturating_duration_since(now).as_secs_f64());
            now = end;
        }
        ENGINE_WRITE_DURATION_HISTOGRAM.observe(now.saturating_duration_since(start).as_secs_f64());
        ENGINE_WRITE_SIZE_HISTOGRAM.observe(len as f64);
        Ok(len)
    }
pub fn write ( & self , log_batch : & mut LogBatch , mut sync : bool ) -> Result < usize > { let start = Instant :: now ( ) ; let len = log_batch . finish_populate ( self . cfg . batch_compression_threshold . 0 as usize ) ? ; let block_handle = { let mut writer = Writer :: new ( log_batch , sync , start ) ; if let Some ( mut group ) = self . write_barrier . enter ( & mut writer ) { let now = Instant :: now ( ) ; let _t = StopWatch :: new_with ( & ENGINE_WRITE_LEADER_DURATION_HISTOGRAM , now ) ; for writer in group . iter_mut ( ) { ENGINE_WRITE_PREPROCESS_DURATION_HISTOGRAM . observe ( now . saturating_duration_since ( writer . start_time ) . as_secs_f64 ( ) , ) ; sync |= writer . sync ; let log_batch = writer . get_payload ( ) ; let res = if ! log_batch . is_empty ( ) { // @TODO(lucasliang): bind `Version` to each `LogBatch` self . pipe_log . append ( LogQueue :: Append , log_batch . encoded_bytes ( ) ) } else { // TODO(tabokie): use Option<FileBlockHandle> instead. Ok ( FileBlockHandle { id : FileId :: new ( LogQueue :: Append , 0 ) , offset : 0 , len : 0 , } ) } ; writer . set_output ( res ) ; } if let Err ( e ) = self . pipe_log . maybe_sync ( LogQueue :: Append , sync ) { panic ! ( " " LogQueue Append e ) ; } } writer . finish ( ) ? } ; let mut now = Instant :: now ( ) ; if len > 0 { log_batch . finish_write ( block_handle ) ; self . memtables . apply_append_writes ( log_batch . drain ( ) ) ; for listener in & self . listeners { listener . post_apply_memtables ( block_handle . id ) ; } let end = Instant :: now ( ) ; ENGINE_WRITE_APPLY_DURATION_HISTOGRAM . observe ( end . saturating_duration_since ( now ) . as_secs_f64 ( ) ) ; now = end ; } ENGINE_WRITE_DURATION_HISTOGRAM . observe ( now . saturating_duration_since ( start ) . as_secs_f64 ( ) ) ; ENGINE_WRITE_SIZE_HISTOGRAM . observe ( len as f64 ) ; Ok ( len ) }
Writes the content of `log_batch` into the engine and returns written
bytes.
Writes the content of ` log_batch ` into the engine and returns written bytes .
&self
log_batch
&mut LogBatch
sync
bool