diff --git a/README.md b/README.md index 924f603..9253eba 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,6 @@ ![Rust](https://github.com/TommyCpp/monolith/workflows/Rust/badge.svg) [![Project Status: WIP – Initial development is in progress, but there has not yet been a stable, usable release suitable for the public.](https://www.repostatus.org/badges/latest/wip.svg)](https://www.repostatus.org/#wip) - Timeseries database. It could be used as an storage backend for prometheus to store metric data. It's still a **WIP** project(Including this README). diff --git a/doc/wal.md b/doc/wal.md index e8cd664..80294e4 100644 --- a/doc/wal.md +++ b/doc/wal.md @@ -1,11 +1,50 @@ ## Write Ahead Log ### Objective -In order to reduce the IO, which is expensive. We should implement a in-memory cache for current opening chunk. And when that chunk get swapped. We can compact it and store it in backend. +In order to reduce the IO, which is expensive. We should implement an in-memory cache for current opening chunk. And when that chunk get swapped. We can compact it and store it in backend. But if the system crushes, we will loss all data in memory. Thus, it's necessary we have a WAL to record every operation we applied to the data store. -### Possible Situation -Note that currently, the chunk will store index information first, and based on the series id returned by `Indexer`. It will processed and store the time point. We will make sure that storing index is always happens before storing time point. And when and only when the index and time point stored in WAL, we will return result to client and change in memory. +### Approaches +1. Cache before Indexer/Storage +``` + +---------------+ + | | + | Indexer | + | | + +-------+-------+ ++------------+ | +| | | +| WAL +-------------------+ +| | | ++------------+ | + +-------+-------+ + | | + | Storage | + | | + +---------------+ +``` +This is the straight-forward method. + +However, one problem with this problem is that this way WAL will be force to stay in local machine, which could take quite some memory. + +2. Cache within Indexer/Storage +``` ++------------+ +---------------+ +| | | | +| WAL +-+ Indexer | +| | | | ++------------+ +---------------+ + ++------------+ +---------------+ +| | | | +| WAL +-+ Storage | +| | | | ++------------+ +---------------+ + +``` +Note that we need to take cautious about how to cache WAL in this case. + +Currently, the chunk will store index information first, and based on the series id returned by `Indexer`. It will processed and store the time point. We will make sure that storing index is always happens before storing time point. And when and only when the index and time point stored in WAL, we will return result to client and change in memory. 1. Crush before storing index. Nothing saved in memory or WAL. 2. Crush after storing index in WAL. But not yet applied to memory. No effect. But when restored, we may find a empty time series. @@ -13,7 +52,6 @@ Note that currently, the chunk will store index information first, and based on 4. Crush after storing index, and the time point into WAL, but not yet applied to memory. If crushes, we can recover from WAL 5. Crush happens after all operation. We can recover from WAL. -### Cache In most of implementation, WAL will have some kind of cache. For example, MongoDB flush wal's content every 100ms. So it's possible to lose the data we haven't flushed to disk yet. Here, we must make sure that the index is up to date, that's to say **we must flush for every new time series**. But we can allow caching for time point data. diff --git a/src/common/label.rs b/src/common/label.rs index 47151f0..f75daf4 100644 --- a/src/common/label.rs +++ b/src/common/label.rs @@ -9,7 +9,6 @@ pub struct Label { value: String, } -/// /// Label will be sort with alphabet order but note that two label will only be equal if and only if /// they have the same key and same value impl Label { diff --git a/src/common/mod.rs b/src/common/mod.rs index c41537f..a5400e9 100644 --- a/src/common/mod.rs +++ b/src/common/mod.rs @@ -54,7 +54,7 @@ impl IdGenerator { } } -/// Build a Indexer or Storage object. +/// Build an Indexer or Storage object. /// /// Implementation may add more function to let user pass more configs or options pub trait Builder { @@ -74,7 +74,7 @@ pub trait Builder { /// Write additional config or metadata information in db dir. fn write_config(&self, dir: &Path) -> Result<()>; - /// Read additional config or metadata information from db dir. + /// Read additional configs or metadata information from db dir. fn read_config(&self, dir: &Path) -> Result<()>; } diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs index b3b3fcd..af968f0 100644 --- a/src/compaction/mod.rs +++ b/src/compaction/mod.rs @@ -185,7 +185,7 @@ impl Decompactor { /// Bstream is not thread safe. /// /// We must filling one byte before appending another one to data. -/// The `remaining` indicates how may bit is available in current byte. +/// The `remaining` indicates how may bit is available in the current byte. /// #[derive(Clone)] pub(crate) struct Bstream { @@ -201,7 +201,7 @@ impl Bstream { } } - /// Create Bstream from vector of bytes and remaining + /// Create Bstream from the vector of bytes and remaining pub fn from(data: Vec, remaining: u8) -> Bstream { Bstream { data, remaining } } @@ -232,7 +232,7 @@ impl Bstream { } } - /// How many bit in this bit stream + /// How many bits in this bit stream pub fn bitlen(&self) -> usize { self.data.len() * 8 - self.remaining as usize } diff --git a/src/error.rs b/src/error.rs index 58c96d4..920d73e 100644 --- a/src/error.rs +++ b/src/error.rs @@ -4,6 +4,7 @@ use failure::_core::num::{ParseFloatError, ParseIntError}; use crate::chunk::Chunk; use crate::indexer::Indexer; use crate::storage::Storage; +use crate::wal; use serde_json::Error; use std::string::FromUtf8Error; use std::sync::Arc; @@ -32,8 +33,8 @@ pub enum MonolithErr { TiKvErr(tikv_client::Error), #[fail(display = "Error when compaction or de-compaction, {}", _0)] CompactionErr(crate::compaction::CompactionErr), - #[fail(display= "Write ahead log error, {}", _0)] - WalErr(crate::wal::WalErr) + #[fail(display = "[wal] {}", _0)] + WalErr(crate::wal::WalErr), } pub type Result = std::result::Result; @@ -101,3 +102,9 @@ impl std::convert::From for MonolithErr { MonolithErr::SerdeYamlErr(err) } } + +impl std::convert::From for MonolithErr { + fn from(err: wal::WalErr) -> Self { + MonolithErr::WalErr(err) + } +} diff --git a/src/indexer/common.rs b/src/indexer/common.rs index 86958b9..a16005d 100644 --- a/src/indexer/common.rs +++ b/src/indexer/common.rs @@ -7,7 +7,7 @@ use crate::{HasTypeName, Result}; /// Indexer is in charge of query appropriate time series based on the labels. /// pub trait Indexer: Sized + HasTypeName { - /// Get all time series and their meta data which contains __all__ labels + /// Get all time series, and their metadata which contains __all__ labels /// /// Note that the result time series may contains other labels fn get_series_metadata_contains_labels( @@ -26,7 +26,7 @@ pub trait Indexer: Sized + HasTypeName { /// /// time_series_id must be single increasing. /// __Will not re-sort__ the time_series_id in values - /// Will create three kind of mapping: + /// Will create three kinds of mapping: /// 1. mapping from each label to time series id, used to search by label /// 2. mapping form label set to time series id, used to find target series id by complete label set /// 3. mapping from time series id to label set, used to get all meta data from time series id. diff --git a/src/indexer/sled_indexer.rs b/src/indexer/sled_indexer.rs index 0beeccc..eb80086 100644 --- a/src/indexer/sled_indexer.rs +++ b/src/indexer/sled_indexer.rs @@ -17,11 +17,11 @@ const LABEL_PREFIX: &str = "L"; const ID_PREFIX: &str = "I"; /// -/// Sled based indexer, use to search timeseries id based on metadata. +/// Sled based indexer, used to search timeseries id based on metadata. /// /// SledIndexer will establish three kinds of mapping /// 1. Reverse index mapping, from single label to list of ids, e.g LR= -> 1,2,3,4,5... -/// 2. index mapping, meta data for a single time series, from id to a list of labels, e.g I1 -> L=,=... +/// 2. index mapping, metadata for a single time series, from ids to a list of labels, e.g I1 -> L=,=... /// 3. labels set mapping, similar to second one but in reverse, e.g L=,=... -> 1 #[derive(Clone)] pub struct SledIndexer { diff --git a/src/lib.rs b/src/lib.rs index da5fcba..50955ca 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,27 +2,26 @@ use std::time::Duration; +mod backend; mod common; mod error; -mod backend; mod wal; -pub mod compaction; pub mod chunk; +pub mod compaction; pub mod db; -pub mod server; pub mod indexer; +pub mod server; pub mod storage; - /// Generated proto definition pub(crate) mod proto; pub use common::*; pub use error::*; -pub use db::MonolithDb; -pub use backend::TiKvRawBackendSingleton; pub use backend::TiKvBackendConfigFile; +pub use backend::TiKvRawBackendSingleton; +pub use db::MonolithDb; pub type Timestamp = u64; pub type Value = f64; diff --git a/src/storage/common.rs b/src/storage/common.rs index 1fbdead..b7a58bb 100644 --- a/src/storage/common.rs +++ b/src/storage/common.rs @@ -2,7 +2,6 @@ use crate::common::time_point::TimePoint; use crate::common::time_series::TimeSeriesId; use crate::{HasTypeName, Result, Timestamp, Value}; -/// /// Storage is in charge of storing time series data /// Note that the label should be store in Indexer instead of Storage pub trait Storage: Sized + HasTypeName { diff --git a/src/wal/mod.rs b/src/wal/mod.rs index 5f3a12f..95e6dd8 100644 --- a/src/wal/mod.rs +++ b/src/wal/mod.rs @@ -1,23 +1,17 @@ -use crc::Hasher32; -use std::path::PathBuf; use std::sync::{Arc, Mutex}; +use crc::Hasher32; +use std::thread::JoinHandle; +use std::fs::File; + mod segment; -mod writer; +mod wal; /// Note that we only use the first three byte in this magic number /// User can still use the remaining 5 bytes to add more metadata /// Like WAL for different component pub(crate) const WAL_MAGIC_NUMBER: u64 = 0x57414C0000000000u64; -// Wal File structure -// | component | length | -// |-------------|---------| -// | Magic Number| u64 | -// | Entries | n * u64 | -// | ....... | ....... | -// | CRC64 | u64 | - /// Entry encoding /// ----------------------------------------------------------------- /// | seq_num(u64) | type(u8) | len(u16) | data(bytes) | CRC32(u32) | @@ -52,17 +46,19 @@ impl Default for Entry { } impl Entry { - pub fn new(seq_id: u64, entry_type: EntryType) -> Entry { + pub fn new(seq_id: u64, entry_type: EntryType, content: Vec) -> Entry { Entry { seq_id, entry_type: entry_type as u8, - content: vec![], + content, crc: crc::crc32::Digest::new(crc::crc32::IEEE), } } + /// Return number of bytes in entry + /// Including seq_id, entry_type, content length, content, crc32 pub fn len(&self) -> usize { - 8 + 1 + self.content.len() + 4 // seq_id + entry_type + content + crc32 + 8 + 1 + 2 + self.content.len() + 4 } pub fn get_bytes(&self) -> Vec { @@ -82,11 +78,8 @@ impl Entry { } } -pub struct WalConfig { - pub filepath: PathBuf, -} - -pub enum FlushPolicy { +#[derive(Clone, Copy)] +pub enum SyncPolicy { TimeBased(std::time::Duration), // flush based on the num of entries. NumBased(usize), @@ -97,11 +90,11 @@ pub enum FlushPolicy { Immediate, } -/// FlushCache tells segment how to cache bytes -pub enum FlushCache { +/// FlushCache tells segments how to cache bytes +pub enum SyncCache { TimeBased { - handler: std::thread::JoinHandle<()>, - cache: Arc>>, + message_handle: std::sync::mpsc::Sender, + ticker: JoinHandle<()>, }, NumBased { limit: usize, @@ -110,30 +103,42 @@ pub enum FlushCache { }, SizeBased { limit: usize, - size: usize, // num of bytes + size: usize, + // num of bytes cache: Vec, }, None, } +// Messages used in time based sync +pub enum TimeSyncMessage { + Insert(Entry), + Shutdown, + Flush, +} + pub enum EntryType { - Default = 0, // + Default = 0, } #[derive(Debug, Fail)] pub enum WalErr { - #[fail(display = "Internal error {}", _0)] + #[fail(display = "{}", _0)] InternalError(String), - #[fail(display = "{}", _0)] + #[fail(display = "io error, {}", _0)] FileIoErr(std::io::Error), + + #[fail(display = "cannot found entry")] + NotFoundErr, } #[cfg(test)] mod tests { - use crate::wal::Entry; use crc::Hasher32; + use crate::wal::Entry; + #[test] pub fn test_entry_get_bytes() { let mut entry = Entry::default(); diff --git a/src/wal/segment.rs b/src/wal/segment.rs index dd31035..df5edeb 100644 --- a/src/wal/segment.rs +++ b/src/wal/segment.rs @@ -1,132 +1,231 @@ -use crate::wal::{Entry, FlushCache, FlushPolicy, WalErr, WAL_MAGIC_NUMBER}; -use crate::Result; -use std::fs::File; +use crate::wal::WalErr::FileIoErr; +use crate::wal::{Entry, SyncCache, SyncPolicy, WalErr, WAL_MAGIC_NUMBER, TimeSyncMessage}; +use crate::{MonolithErr, Result}; +use futures::io::Error; +use std::convert::TryInto; +use std::fs::{File, OpenOptions}; use std::hash::Hasher; -use std::io::Write; +use std::io::{ErrorKind, Read, Seek, SeekFrom, Write}; +use std::path::Path; use std::sync::atomic::AtomicU64; use std::sync::{Arc, Mutex}; +use std::thread::{spawn, sleep}; type FileMutex = Arc>; -const AVG_NUM_BYTES_IN_ENTRY: usize = 8; +const AVG_NUM_BYTES_IN_ENTRY: usize = 32; -/// Segment is one write ahead file +///! Wal File structure +///! | component | length | +///! |-------------|---------| +///! | Magic Number| u64 | +///! | Entries | n * u8 | +///! | ....... | ....... | +///! | First seqid | u64 | +///! | Last seqid | u64 | +///! | CRC64 | u64 | + +/// SegmentWriter is the writer for segment file /// -/// Segment is **NOT** concurrent-safe. -pub struct Segment { - writer: W, - cache: FlushCache, +/// SegmentWriter is **NOT** concurrent-safe. +pub struct SegmentWriter { + first_idx: u64, + last_idx: u64, + + file: File, crc: crc::crc64::Digest, + + cache: SyncCache, } -impl Write for Segment { - fn write(&mut self, bytes: &[u8]) -> std::io::Result { - self.crc.write(bytes); // update crc - self.writer.write(bytes) - } +impl SegmentWriter { + /// Create a segment file, open with append-only + /// + /// If there is no file yet. Create one and write the overhead. + /// If there is one, update `first_idx` and `last_idx`, also verify the crc and reset the cursor + /// so that we could override crc at the end of the file. + pub fn new>(sync_policy: SyncPolicy, path: P) -> Result { + let (first_idx, last_idx, crc_initial, mut file) = if path.as_ref().exists() { + // if there is file + let mut file = OpenOptions::new().read(true).append(true).open(&path)?; - fn flush(&mut self) -> std::io::Result<()> { - let remaining = match &mut self.cache { - &mut FlushCache::None => vec![], - &mut FlushCache::NumBased { - limit: ref size, - ref mut idx, - ref mut cache, - } => { - *idx = 0; - let _cache = cache.clone(); - *cache = vec![Entry::default(); *size]; - _cache - } - &mut FlushCache::TimeBased { .. } => { - // fixme and above - vec![] + // check for crc + // get crc value in file + let pos = file.seek(SeekFrom::End(-8))?; + let crc_val = file.read_be_u64()?; + // read content + file.seek(SeekFrom::Start(0)); + let mut buf = vec![0; pos as usize]; + file.read_exact(buf.as_mut_slice()); + if crc::crc64::checksum_ecma(buf.as_slice()) != crc_val { + // corrupt file. return error + return Err(MonolithErr::from(WalErr::FileIoErr(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "crc fail to match", + )))); } - &mut FlushCache::SizeBased { - ref limit, - ref mut cache, - ref mut size, - } => { - *size = 0; - let _cache = cache.clone(); - *cache = Vec::with_capacity(limit * AVG_NUM_BYTES_IN_ENTRY); - _cache - } - }; - if remaining.len() > 0 { - for entry in remaining { - self.write(entry.content.as_slice())?; - } + // get first id and last id + file.seek(SeekFrom::Current(-16)); + let first_id = file.read_be_u64()?; + let last_id = file.read_be_u64()?; + + // reset cursor so that we could over write crc + // we need to minus 16 because we also need to overwrite `first_index` and `last_index` + // but they are part of the value to calculate CRC. So we cannot just do it before we verify + // the crc. + file.set_len(pos - 16); + + // re-calculate CRC. The old crc including `first_index` and `last_index` + // So we need to remove that part and re-calculate + let new_crc_val = crc::crc64::checksum_ecma(&buf.as_slice()[..buf.len() - 16]); + + Ok((first_id, last_id, new_crc_val, file)) + } else { + let mut file = OpenOptions::new().append(true).create(true).open(&path)?; + + // write magic number. Note that we don't include this when compute CRC + file.write(&WAL_MAGIC_NUMBER.to_be_bytes()[..]); + let crc_val = crc::crc64::checksum_ecma(&WAL_MAGIC_NUMBER.to_be_bytes()[..]); + file.sync_all(); + Ok((0, 0, crc_val, file)) } + .map_err(|err| WalErr::FileIoErr(err))?; - self.writer.flush() - } -} + let mut crc = crc::crc64::Digest::new_with_initial(crc::crc64::ECMA, crc_initial); + Ok(SegmentWriter { + first_idx, + last_idx, + cache: match sync_policy { + SyncPolicy::Immediate => SyncCache::None, + SyncPolicy::TimeBased(dur) => { + let (main_sender, receiver) = std::sync::mpsc::channel::(); + let ticker_sender = main_sender.clone(); + let (mut file, mut crc) = (file.try_clone().unwrap(), crc::crc64::Digest::new_with_initial(crc::crc64::ECMA, crc_initial)); + let (first_idx, last_idx) = (first_idx.clone(), last_idx.clone()); + let _handler = spawn(move || { + let mut cache = vec![]; + loop { + match receiver.recv() { + Ok(msg) => { + match msg { + TimeSyncMessage::Insert(entry) => { + cache.push(entry); + } + TimeSyncMessage::Shutdown => { + let bytes: Vec = cache.iter().flat_map(|entry| entry.get_bytes()).collect(); + crc.write(bytes.as_slice()); + file.write(bytes.as_slice()); -impl Segment - where - W: Write, -{ - pub fn new(flush_policy: FlushPolicy, mut writer: W) -> Result> { - // write magic number. Note that we don't include this when compute CRC - writer.write(&WAL_MAGIC_NUMBER.to_be_bytes()[..]); - writer.flush(); - - Ok(Segment { - writer, - cache: match flush_policy { - FlushPolicy::Immediate => FlushCache::None, - FlushPolicy::TimeBased(dur) => { - let cache = Arc::new(Mutex::new(vec![])); - FlushCache::TimeBased { - handler: std::thread::spawn(|| {}), //todo: figure out how to do time based flush - cache, + // append first and last sequence id + let _ = file.write(&first_idx.to_be_bytes()[..]); + let _ = file.write(&last_idx.to_be_bytes()[..]); + + // append crc + let crc = crc.finish(); + let _ = file.write(&crc.to_be_bytes()); // append crc result + return; + } + TimeSyncMessage::Flush => { + let bytes: Vec = cache.iter().flat_map(|entry| entry.get_bytes()).collect(); + crc.write(bytes.as_slice()); + file.write(bytes.as_slice()); + cache = vec![]; + } + } + } + Err(_err) => { + //todo: error handling + } + } + } + }); + SyncCache::TimeBased { + ticker: spawn(move || { + loop { + sleep(dur); + let _ = ticker_sender.send(TimeSyncMessage::Flush); + } + }), + message_handle: main_sender, } } - FlushPolicy::NumBased(limit) => FlushCache::NumBased { + SyncPolicy::NumBased(limit) => SyncCache::NumBased { limit, idx: 0, - cache: Vec::with_capacity(limit), + cache: vec![Entry::default(); limit], }, - FlushPolicy::SizeBased(limit) => FlushCache::SizeBased { + SyncPolicy::SizeBased(limit) => SyncCache::SizeBased { limit, size: 0, cache: Vec::with_capacity(limit * AVG_NUM_BYTES_IN_ENTRY), }, }, - crc: crc::crc64::Digest::new(crc::crc64::ECMA), + file, + crc, }) } - /// flush all data and close the file + /// Sync all data and close the file pub fn close(&mut self) -> Result<()> { + self.flush()?; // flush all cached entries + + // append first and last sequence id + self.write(&self.first_idx.to_be_bytes())?; + self.write(&self.last_idx.to_be_bytes())?; + + // append crc let _crc = self.crc.finish(); - self.write(&_crc.to_be_bytes())?; // append crc result + self.file.write(&_crc.to_be_bytes())?; // append crc result + + self.sync(); // make sure all data synced. Ok(()) } + /// Ensure all data has been written to disk. + pub fn sync(&mut self) -> Result<()> { + self.flush()?; + self.file.sync_all()?; + Ok(()) + } + + /// Write an entry into segment file. + /// + /// Based on different sync policy. It will trigger sync to make sure data reaches disk. + /// User can also manually call `sync` method to ask segment to sync immediately flush cache and + /// sync disk. pub fn write_entry(&mut self, entry: Entry) -> Result<()> { + if self.first_idx == 0 { + self.first_idx = entry.seq_id; + } + if self.last_idx >= entry.seq_id { + return Err(MonolithErr::WalErr(WalErr::InternalError( + "sequence id is smaller than the last sequence id in writer".to_string(), + ))); + } + self.last_idx = entry.seq_id; + + // Update cache match &mut self.cache { - &mut FlushCache::None => { + &mut SyncCache::None => { self.write(entry.get_bytes().as_slice())?; - self.flush(); + self.sync(); } - &mut FlushCache::NumBased { - limit: ref size, + &mut SyncCache::NumBased { + limit: size, ref mut idx, ref mut cache, } => { cache[*idx] = entry; *idx += 1; - if idx == size { - self.flush(); + if *idx == size { + self.sync(); } } - &mut FlushCache::TimeBased { ref mut cache, .. } => { - cache.lock().unwrap().push(entry); + &mut SyncCache::TimeBased { ref message_handle, .. } => { + let _ = message_handle.send(TimeSyncMessage::Insert(entry)); } - &mut FlushCache::SizeBased { + &mut SyncCache::SizeBased { ref mut cache, ref limit, ref mut size, @@ -134,7 +233,7 @@ impl Segment let entry_len = entry.len(); cache.push(entry); if *size + entry_len >= *limit { - self.flush(); + self.sync(); } else { *size += entry_len; } @@ -144,25 +243,118 @@ impl Segment } } +impl Write for SegmentWriter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.crc.write(bytes); // update crc + self.file.write(bytes) + } + + /// Flush all data into OS's file system cache. + /// Note that there is no guarantee that it has been synced to disk. + /// To make sure all data has been synced to disk. Use `Segment::sync()` + fn flush(&mut self) -> std::io::Result<()> { + let remaining: Vec = match &mut self.cache { + &mut SyncCache::None => vec![], + &mut SyncCache::NumBased { + limit: ref size, + ref mut idx, + ref mut cache, + } => { + // Note that here the cache have a fix size and we initial every element to be default + // entry. So we need to get the actual meaningful entries. Thus, we only take the + // first `idx` entries. + // Also note that `idx` will always insert index of next entry. + let mut _cache = Vec::::with_capacity(*idx); + _cache.extend_from_slice(&cache.as_slice()[0..*idx]); + + *idx = 0; + *cache = vec![Entry::default(); *size]; + _cache + } + &mut SyncCache::TimeBased { ref message_handle, .. } => { + let _ = message_handle.send(TimeSyncMessage::Flush); + vec![] + } + &mut SyncCache::SizeBased { + ref limit, + ref mut cache, + ref mut size, + } => { + *size = 0; + let _cache = cache.clone(); + *cache = Vec::with_capacity(limit / AVG_NUM_BYTES_IN_ENTRY); + _cache + } + }; + + if remaining.len() > 0 { + for entry in remaining { + self.write(entry.get_bytes().as_slice())?; + } + } + + Ok(()) + } +} + +impl Drop for SegmentWriter { + fn drop(&mut self) { + if let SyncCache::TimeBased { ref message_handle, .. } = self.cache { + message_handle.send(TimeSyncMessage::Shutdown); + } else { + self.close(); + } + } +} + +/// SegmentReader if reader for segment file. +pub struct SegmentReader { + first_idx: u64, + last_idx: u64, + file: File, +} + +impl SegmentReader { + pub fn new>(path: P) -> Result { + unimplemented!(); + } +} + +trait ReadExt { + fn read_be_u64(&mut self) -> std::result::Result; +} + +impl ReadExt for T + where + T: Read, +{ + fn read_be_u64(&mut self) -> std::result::Result { + let mut buffer = vec![0; 8]; + self.read_exact(buffer.as_mut_slice())?; + + Ok(u64::from_be_bytes(buffer.as_slice().try_into().unwrap())) + } +} + #[cfg(test)] mod tests { - use crate::wal::segment::Segment; - use crate::wal::{Entry, EntryType, FlushPolicy, WAL_MAGIC_NUMBER}; + use crate::wal::segment::SegmentWriter; + use crate::wal::{Entry, EntryType, SyncPolicy, WAL_MAGIC_NUMBER}; use crate::Result; use futures::io::SeekFrom; + use std::fs; use std::fs::{File, OpenOptions}; - use std::io::{Read, Seek}; + use std::hash::Hasher; + use std::io::SeekFrom::Start; + use std::io::{Read, Seek, Write}; use tempfile::TempPath; #[test] pub fn test_write_entry() -> Result<()> { let temp_path = tempfile::tempdir()?; - let mut temp_file = OpenOptions::new() - .append(true) - .create(true) - .open(temp_path.path().join("test1"))?; - let mut segment = Segment::::new(FlushPolicy::Immediate, temp_file)?; - let mut entry = Entry::new(1u64, EntryType::Default); + let mut segment = + SegmentWriter::new(SyncPolicy::Immediate, &temp_path.path().join("test1"))?; + let mut entry = Entry::new(1u64, EntryType::Default, vec![]); entry.push(vec![1, 2, 3, 4]); segment.write_entry(entry)?; @@ -204,4 +396,80 @@ mod tests { Ok(()) } + + #[test] + pub fn test_write_entry_with_num_cache() -> Result<()> { + let temp_path = tempfile::tempdir()?; + let file_path = temp_path.path().join("test1"); + let mut segment = SegmentWriter::new(SyncPolicy::NumBased(5), &file_path)?; + + for i in 1..5 { + segment.write_entry(Entry::new(i as u64, EntryType::Default, vec![])); + } + + // no entry yet. + assert_eq!(fs::read_to_string(&file_path)?.len(), 8); + + segment.write_entry(Entry::new(5u64, EntryType::Default, vec![])); + + // Assert we have flush 5 entries. + assert_eq!(fs::read_to_string(&file_path)?.len(), 8 + 15 * 5); + + segment.write_entry(Entry::new(6u64, EntryType::Default, vec![])); + segment.flush(); // force flush + assert_eq!(fs::read_to_string(&file_path)?.len(), 8 + 15 * 6); + + Ok(()) + } + + #[test] + pub fn test_close_entry() -> Result<()> { + let temp_path = tempfile::tempdir()?; + let file_path = temp_path.path().join("test1"); + let mut segment = SegmentWriter::new(SyncPolicy::NumBased(20), &file_path)?; + segment.write_entry(Entry::new(1u64, EntryType::Default, vec![])); + + assert_eq!(fs::read_to_string(&file_path)?.len(), 8); // no entry yet. + + segment.close(); + + // flushed one entry(15 bytes) and + // first index + // last index + // crc64 + let mut buf = Vec::new(); + let mut file = File::open(&file_path)?; + file.read_to_end(buf.as_mut()); + assert_eq!(buf.len(), 8 + 15 + 8 + 8 + 8); + + Ok(()) + } + + #[test] + pub fn test_open_existing_log_file() -> Result<()> { + let temp_path = tempfile::tempdir()?; + let file_path = temp_path.path().join("test1"); + let mut segment = SegmentWriter::new(SyncPolicy::Immediate, &file_path)?; + assert_eq!( + segment.crc.finish(), + crc::crc64::checksum_ecma(&WAL_MAGIC_NUMBER.to_be_bytes()[..]) + ); + segment.write_entry(Entry::new(1, EntryType::Default, vec![])); + segment.write_entry(Entry::new(2, EntryType::Default, vec![])); + drop(segment); // close segment; + + let mut segment = SegmentWriter::new(SyncPolicy::Immediate, &file_path)?; + assert_eq!(segment.first_idx, 1); + assert_eq!(segment.last_idx, 2); + + segment.write_entry(Entry::new(3, EntryType::Default, vec![])); + segment.write_entry(Entry::new(4, EntryType::Default, vec![])); + drop(segment); + + let mut segment = SegmentWriter::new(SyncPolicy::Immediate, &file_path)?; + assert_eq!(segment.first_idx, 1); + assert_eq!(segment.last_idx, 4); + + Ok(()) + } } diff --git a/src/wal/wal.rs b/src/wal/wal.rs new file mode 100644 index 0000000..377973b --- /dev/null +++ b/src/wal/wal.rs @@ -0,0 +1,71 @@ +use crate::wal::SyncPolicy; +use crate::{ + wal::{segment::SegmentWriter, Entry, EntryType}, + Result, +}; +use std::fs::File; +use std::io::Write; +use std::path::Path; +use std::sync::atomic::AtomicU64; +use std::sync::{Arc, Mutex, RwLock}; + +type LockedSegment = Arc>; + +pub struct WalConfig { + sync_policy: SyncPolicy, + // must be larger than 1. + initial_seq: u64, +} + +impl Default for WalConfig { + fn default() -> Self { + WalConfig { + sync_policy: SyncPolicy::Immediate, + initial_seq: 1, + } + } +} + +/// Wal manages a dictionary of segment wal files. +/// +/// Wal is concurrent-safe +/// +/// Wal will +/// 1. Manage and assign sequence id. +/// 2. Rotate segment when its size is too large. +/// 3. Clean up segment file once user know the change has been written. +pub struct Wal<'a> { + dir: &'a Path, + sync_policy: SyncPolicy, // sync policy of segment. + + // generate the file name to maintain file order. + next_seq: AtomicU64, + // active segment + active_seg: LockedSegment, + // read only ordered in creating time. From earliest to latest + segments: Vec>, +} + +impl<'a> Wal<'a> { + /// Open an dictionary to read and write write ahead log. + /// If there are already some segment file in dictionary. Read them. + fn open>(path: P, config: Option) -> Result> { + unimplemented!(); + } + + /// Write an entry into wal. + /// Not necessary to sync immediately. + fn write(&self, entry_type: EntryType, data: &[u8]) -> Result<()> { + unimplemented!(); + } + + /// Force flush to make sure all contents in cache will be written to storage + fn flush(&self) -> Result<()> { + unimplemented!() + } + + /// Ready one entry with entry id `idx` + fn read(&self, idx: u64) -> Result { + unimplemented!(); + } +} diff --git a/src/wal/writer.rs b/src/wal/writer.rs deleted file mode 100644 index d0c8f5c..0000000 --- a/src/wal/writer.rs +++ /dev/null @@ -1,23 +0,0 @@ -use std::sync::Mutex; -use std::fs::File; -use std::io::Write; -use std::sync::atomic::AtomicU64; -use crate::wal::segment::Segment; - -pub struct WalWriter { - next_seq: AtomicU64, // generate the file name to maintain file order. - active_seg: Segment, // active segment - segments: Vec>, // cannot write again - crc: crc::crc64::Digest -} - -impl Write for WalWriter - where W: Write{ - fn write(&mut self, buf: &[u8]) -> std::io::Result { - unimplemented!() - } - - fn flush(&mut self) -> std::io::Result<()> { - unimplemented!() - } -} \ No newline at end of file