Wal

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).

doc/wal.md

@@ -1,19 +1,57 @@
 ## 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.
 3. Crush after storing index in WAL and applying to memory. No effect since user will not get any response before `Storage` finishes. Again, we may find an empty time series when restored. 
 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. 

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 {

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<T> {
@@ -74,7 +74,7 @@ pub trait Builder<T> {
     /// 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<()>;
 }
 

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<u8>, 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
     }

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<T> = std::result::Result<T, MonolithErr>;
@@ -101,3 +102,9 @@ impl std::convert::From<serde_yaml::Error> for MonolithErr {
         MonolithErr::SerdeYamlErr(err)
     }
 }
+
+impl std::convert::From<wal::WalErr> for MonolithErr {
+    fn from(err: wal::WalErr) -> Self {
+        MonolithErr::WalErr(err)
+    }
+}

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.

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<label_key>=<label_value> -> 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<label_key>=<label_value>,<label_key>=<label_value>...
+/// 2. index mapping, metadata for a single time series, from ids to a list of labels, e.g I1 -> L<label_key>=<label_value>,<label_key>=<label_value>...
 /// 3. labels set mapping, similar to second one but in reverse, e.g L<label_key>=<label_value>,<label_key>=<label_value>... -> 1
 #[derive(Clone)]
 pub struct SledIndexer {

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;

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 {

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<u8>) -> 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<u8> {
@@ -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<Mutex<Vec<Entry>>>,
+        message_handle: std::sync::mpsc::Sender<TimeSyncMessage>,
+        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<Entry>,
     },
     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();