From 6ca9610276aacd0328c6c87df24f83fabc375329 Mon Sep 17 00:00:00 2001
From: "zhongyang.wu" <zhongyang.wu@outlook.com>
Date: Sun, 26 Jul 2020 21:51:00 -0400
Subject: [PATCH 1/5] [wal] Fix bugs, add tests.

---
 src/error.rs       |   4 +-
 src/lib.rs         |  11 ++-
 src/wal/mod.rs     |  25 ++++---
 src/wal/segment.rs | 164 ++++++++++++++++++++++++++++++++-------------
 src/wal/wal.rs     |  61 +++++++++++++++++
 src/wal/writer.rs  |  23 -------
 6 files changed, 198 insertions(+), 90 deletions(-)
 create mode 100644 src/wal/wal.rs
 delete mode 100644 src/wal/writer.rs

diff --git a/src/error.rs b/src/error.rs
index 58c96d4..08f635e 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -32,8 +32,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 = "Write ahead log error, {}", _0)]
+    WalErr(crate::wal::WalErr),
 }
 
 pub type Result<T> = std::result::Result<T, MonolithErr>;
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/wal/mod.rs b/src/wal/mod.rs
index 5f3a12f..aefa8f2 100644
--- a/src/wal/mod.rs
+++ b/src/wal/mod.rs
@@ -3,7 +3,7 @@ use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 
 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
@@ -52,17 +52,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 +84,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),
@@ -98,7 +97,7 @@ pub enum FlushPolicy {
 }
 
 /// FlushCache tells segment how to cache bytes
-pub enum FlushCache {
+pub enum SyncCache {
     TimeBased {
         handler: std::thread::JoinHandle<()>,
         cache: Arc<Mutex<Vec<Entry>>>,
@@ -117,15 +116,15 @@ pub enum FlushCache {
 }
 
 pub enum EntryType {
-    Default = 0, //
+    Default = 0,
 }
 
 #[derive(Debug, Fail)]
 pub enum WalErr {
-    #[fail(display = "Internal error {}", _0)]
+    #[fail(display = "wal: {}", _0)]
     InternalError(String),
 
-    #[fail(display = "{}", _0)]
+    #[fail(display = "wal: io error, {}", _0)]
     FileIoErr(std::io::Error),
 }
 
diff --git a/src/wal/segment.rs b/src/wal/segment.rs
index dd31035..6a6a32e 100644
--- a/src/wal/segment.rs
+++ b/src/wal/segment.rs
@@ -1,95 +1,103 @@
-use crate::wal::{Entry, FlushCache, FlushPolicy, WalErr, WAL_MAGIC_NUMBER};
+use crate::wal::{Entry, SyncCache, SyncPolicy, WalErr, WAL_MAGIC_NUMBER};
 use crate::Result;
 use std::fs::File;
 use std::hash::Hasher;
-use std::io::Write;
+use std::io::{Read, Write};
 use std::sync::atomic::AtomicU64;
 use std::sync::{Arc, Mutex};
 
 type FileMutex = Arc<Mutex<File>>;
 
-const AVG_NUM_BYTES_IN_ENTRY: usize = 8;
+const AVG_NUM_BYTES_IN_ENTRY: usize = 32;
 
 /// Segment is one write ahead file
 ///
 /// Segment is **NOT** concurrent-safe.
-pub struct Segment<W: Write> {
-    writer: W,
-    cache: FlushCache,
+pub struct Segment {
+    last_idx: u64,
+    file: File,
+    cache: SyncCache,
     crc: crc::crc64::Digest,
 }
 
-impl<W: Write> Write for Segment<W> {
+impl Write for Segment {
     fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
         self.crc.write(bytes); // update crc
-        self.writer.write(bytes)
+        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 = match &mut self.cache {
-            &mut FlushCache::None => vec![],
-            &mut FlushCache::NumBased {
+        let remaining: Vec<Entry> = 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::<Entry>::with_capacity(*idx);
+                _cache.extend_from_slice(&cache.as_slice()[0..*idx]);
+
                 *idx = 0;
-                let _cache = cache.clone();
                 *cache = vec![Entry::default(); *size];
                 _cache
             }
-            &mut FlushCache::TimeBased { .. } => {
+            &mut SyncCache::TimeBased { .. } => {
                 // fixme and above
                 vec![]
             }
-            &mut FlushCache::SizeBased {
+            &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 = Vec::with_capacity(limit / AVG_NUM_BYTES_IN_ENTRY);
                 _cache
             }
         };
 
         if remaining.len() > 0 {
             for entry in remaining {
-                self.write(entry.content.as_slice())?;
+                self.write(entry.get_bytes().as_slice())?;
             }
         }
 
-        self.writer.flush()
+        Ok(())
     }
 }
 
-impl<W> Segment<W>
-    where
-        W: Write,
-{
-    pub fn new(flush_policy: FlushPolicy, mut writer: W) -> Result<Segment<W>> {
+impl Segment {
+    pub fn new(flush_policy: SyncPolicy, mut file: File) -> Result<Segment> {
         // write magic number. Note that we don't include this when compute CRC
-        writer.write(&WAL_MAGIC_NUMBER.to_be_bytes()[..]);
-        writer.flush();
+        file.write(&WAL_MAGIC_NUMBER.to_be_bytes()[..]);
+        file.sync_all();
 
         Ok(Segment {
-            writer,
+            last_idx: 0,
+            file,
             cache: match flush_policy {
-                FlushPolicy::Immediate => FlushCache::None,
-                FlushPolicy::TimeBased(dur) => {
+                SyncPolicy::Immediate => SyncCache::None,
+                SyncPolicy::TimeBased(dur) => {
                     let cache = Arc::new(Mutex::new(vec![]));
-                    FlushCache::TimeBased {
+                    SyncCache::TimeBased {
                         handler: std::thread::spawn(|| {}), //todo: figure out how to do time based flush
                         cache,
                     }
                 }
-                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),
@@ -99,34 +107,47 @@ impl<W> Segment<W>
         })
     }
 
-    /// flush all data and close the file
+    /// Sync all data and close the file
     pub fn close(&mut self) -> Result<()> {
         let _crc = self.crc.finish();
-        self.write(&_crc.to_be_bytes())?; // append crc result
+        self.flush()?;
+        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 flush policy. It will trigger sync to make sure data reaches disk.
     pub fn write_entry(&mut self, entry: Entry) -> Result<()> {
+        self.last_idx = entry.seq_id;
         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, .. } => {
+            &mut SyncCache::TimeBased { ref mut cache, .. } => {
                 cache.lock().unwrap().push(entry);
             }
-            &mut FlushCache::SizeBased {
+            &mut SyncCache::SizeBased {
                 ref mut cache,
                 ref limit,
                 ref mut size,
@@ -134,7 +155,7 @@ impl<W> Segment<W>
                 let entry_len = entry.len();
                 cache.push(entry);
                 if *size + entry_len >= *limit {
-                    self.flush();
+                    self.sync();
                 } else {
                     *size += entry_len;
                 }
@@ -147,11 +168,13 @@ impl<W> Segment<W>
 #[cfg(test)]
 mod tests {
     use crate::wal::segment::Segment;
-    use crate::wal::{Entry, EntryType, FlushPolicy, WAL_MAGIC_NUMBER};
+    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::io::SeekFrom::Start;
+    use std::io::{Read, Seek, Write};
     use tempfile::TempPath;
 
     #[test]
@@ -161,8 +184,8 @@ mod tests {
             .append(true)
             .create(true)
             .open(temp_path.path().join("test1"))?;
-        let mut segment = Segment::<File>::new(FlushPolicy::Immediate, temp_file)?;
-        let mut entry = Entry::new(1u64, EntryType::Default);
+        let mut segment = Segment::new(SyncPolicy::Immediate, temp_file)?;
+        let mut entry = Entry::new(1u64, EntryType::Default, vec![]);
         entry.push(vec![1, 2, 3, 4]);
         segment.write_entry(entry)?;
 
@@ -204,4 +227,53 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    pub fn test_write_entry_with_cache() -> Result<()> {
+        let temp_path = tempfile::tempdir()?;
+        let file_path = temp_path.path().join("test1");
+        let mut temp_file = OpenOptions::new()
+            .append(true)
+            .create(true)
+            .open(&file_path)?;
+        let mut segment = Segment::new(SyncPolicy::NumBased(5), temp_file)?;
+
+        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 temp_file = OpenOptions::new()
+            .append(true)
+            .create(true)
+            .open(&file_path)?;
+        let mut segment = Segment::new(SyncPolicy::NumBased(20), temp_file)?;
+        segment.write_entry(Entry::new(1u64, EntryType::Default, vec![]));
+
+        assert_eq!(fs::read_to_string(&file_path)?.len(), 8); // no entry yet.
+
+        segment.close();
+
+        assert_eq!(fs::read_to_string(&file_path)?.len(), 8 + 15 + 8); // flushed one entry(15 bytes) and crc 64
+
+        Ok(())
+    }
 }
diff --git a/src/wal/wal.rs b/src/wal/wal.rs
new file mode 100644
index 0000000..275d5c7
--- /dev/null
+++ b/src/wal/wal.rs
@@ -0,0 +1,61 @@
+use crate::wal::SyncPolicy;
+use crate::{
+    wal::{segment::Segment, 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<RwLock<Segment>>;
+
+pub struct WalConfig {
+    sync_policy: SyncPolicy,
+    initial_seq: u64,
+}
+
+/// Wal manages a dict of segment wal files.
+///
+pub struct Wal<'a> {
+    dir: &'a Path,
+    flush_policy: SyncPolicy,
+
+    // 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<Arc<Segment>>,
+}
+
+impl<'a> Default for Wal<'a> {
+    fn default() -> Self {
+        unimplemented!()
+    }
+}
+
+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<P: AsRef<Path>>(path: P, config: Option<WalConfig>) -> Result<Wal<'a>> {
+        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<Entry> {
+        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<W: Write> {
-    next_seq: AtomicU64, // generate the file name to maintain file order.
-    active_seg: Segment<W>, // active segment
-    segments: Vec<Segment<W>>, // cannot write again
-    crc: crc::crc64::Digest
-}
-
-impl<W> Write for WalWriter<W>
-    where W: Write{
-    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-        unimplemented!()
-    }
-
-    fn flush(&mut self) -> std::io::Result<()> {
-        unimplemented!()
-    }
-}
\ No newline at end of file

From 22143629a40cceb03e3abd6cbc0e1bb5817648c4 Mon Sep 17 00:00:00 2001
From: "zhongyang.wu" <zhongyang.wu@outlook.com>
Date: Sun, 26 Jul 2020 23:07:04 -0400
Subject: [PATCH 2/5] [wal] Update writeup

---
 doc/wal.md | 46 ++++++++++++++++++++++++++++++++++++++++++----
 1 file changed, 42 insertions(+), 4 deletions(-)

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. 

From 7013b216586d2d873dd8b15e21165f313bc4e7ee Mon Sep 17 00:00:00 2001
From: "zhongyang.wu" <zhongyang.wu@outlook.com>
Date: Mon, 27 Jul 2020 17:43:12 -0400
Subject: [PATCH 3/5] [wal] Allow segment to read existing file and resume.

---
 src/error.rs       |   9 +-
 src/wal/mod.rs     |  18 ++-
 src/wal/segment.rs | 308 ++++++++++++++++++++++++++++++++-------------
 src/wal/wal.rs     |  32 +++--
 4 files changed, 259 insertions(+), 108 deletions(-)

diff --git a/src/error.rs b/src/error.rs
index 08f635e..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,7 +33,7 @@ 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)]
+    #[fail(display = "[wal] {}", _0)]
     WalErr(crate::wal::WalErr),
 }
 
@@ -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)
+    }
+}
diff --git a/src/wal/mod.rs b/src/wal/mod.rs
index aefa8f2..323f09f 100644
--- a/src/wal/mod.rs
+++ b/src/wal/mod.rs
@@ -10,14 +10,6 @@ mod wal;
 /// 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) |
@@ -109,7 +101,8 @@ pub enum SyncCache {
     },
     SizeBased {
         limit: usize,
-        size: usize, // num of bytes
+        size: usize,
+        // num of bytes
         cache: Vec<Entry>,
     },
     None,
@@ -121,11 +114,14 @@ pub enum EntryType {
 
 #[derive(Debug, Fail)]
 pub enum WalErr {
-    #[fail(display = "wal: {}", _0)]
+    #[fail(display = "{}", _0)]
     InternalError(String),
 
-    #[fail(display = "wal: io error, {}", _0)]
+    #[fail(display = "io error, {}", _0)]
     FileIoErr(std::io::Error),
+
+    #[fail(display = "cannot found entry")]
+    NotFoundErr,
 }
 
 #[cfg(test)]
diff --git a/src/wal/segment.rs b/src/wal/segment.rs
index 6a6a32e..f61bcc3 100644
--- a/src/wal/segment.rs
+++ b/src/wal/segment.rs
@@ -1,8 +1,12 @@
+use crate::wal::WalErr::FileIoErr;
 use crate::wal::{Entry, SyncCache, SyncPolicy, WalErr, WAL_MAGIC_NUMBER};
-use crate::Result;
-use std::fs::File;
+use crate::{MonolithErr, Result};
+use futures::io::Error;
+use std::convert::TryInto;
+use std::fs::{File, OpenOptions};
 use std::hash::Hasher;
-use std::io::{Read, Write};
+use std::io::{ErrorKind, Read, Seek, SeekFrom, Write};
+use std::path::Path;
 use std::sync::atomic::AtomicU64;
 use std::sync::{Arc, Mutex};
 
@@ -10,80 +14,88 @@ type FileMutex = Arc<Mutex<File>>;
 
 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 writer for segment file
 ///
-/// Segment is **NOT** concurrent-safe.
-pub struct Segment {
+/// SegmentWriter is **NOT** concurrent-safe.
+pub struct SegmentWriter {
+    first_idx: u64,
     last_idx: u64,
+
     file: File,
     cache: SyncCache,
+
     crc: crc::crc64::Digest,
 }
 
-impl Write for Segment {
-    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
-        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<Entry> = 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::<Entry>::with_capacity(*idx);
-                _cache.extend_from_slice(&cache.as_slice()[0..*idx]);
-
-                *idx = 0;
-                *cache = vec![Entry::default(); *size];
-                _cache
-            }
-            &mut SyncCache::TimeBased { .. } => {
-                // fixme and above
-                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
+impl SegmentWriter {
+    /// Create an 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<P: AsRef<Path>>(sync_policy: SyncPolicy, path: P) -> Result<SegmentWriter> {
+        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)?;
+
+            // 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",
+                ))));
             }
-        };
 
-        if remaining.len() > 0 {
-            for entry in remaining {
-                self.write(entry.get_bytes().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))?;
 
-        Ok(())
-    }
-}
-
-impl Segment {
-    pub fn new(flush_policy: SyncPolicy, mut file: File) -> Result<Segment> {
-        // write magic number. Note that we don't include this when compute CRC
-        file.write(&WAL_MAGIC_NUMBER.to_be_bytes()[..]);
-        file.sync_all();
-
-        Ok(Segment {
-            last_idx: 0,
+        Ok(SegmentWriter {
+            first_idx,
+            last_idx,
             file,
-            cache: match flush_policy {
+            cache: match sync_policy {
                 SyncPolicy::Immediate => SyncCache::None,
                 SyncPolicy::TimeBased(dur) => {
                     let cache = Arc::new(Mutex::new(vec![]));
@@ -103,15 +115,22 @@ impl Segment {
                     cache: Vec::with_capacity(limit * AVG_NUM_BYTES_IN_ENTRY),
                 },
             },
-            crc: crc::crc64::Digest::new(crc::crc64::ECMA),
+            crc: crc::crc64::Digest::new_with_initial(crc::crc64::ECMA, crc_initial),
         })
     }
 
     /// 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.flush()?;
         self.file.write(&_crc.to_be_bytes())?; // append crc result
+
         self.sync(); // make sure all data synced.
         Ok(())
     }
@@ -125,9 +144,21 @@ impl Segment {
 
     /// Write an entry into segment file.
     ///
-    /// Based on different flush policy. It will trigger sync to make sure data reaches disk.
+    /// 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 SyncCache::None => {
                 self.write(entry.get_bytes().as_slice())?;
@@ -165,14 +196,104 @@ impl Segment {
     }
 }
 
+impl Write for SegmentWriter {
+    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+        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<Entry> = 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::<Entry>::with_capacity(*idx);
+                _cache.extend_from_slice(&cache.as_slice()[0..*idx]);
+
+                *idx = 0;
+                *cache = vec![Entry::default(); *size];
+                _cache
+            }
+            &mut SyncCache::TimeBased { .. } => {
+                // fixme and above
+                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) {
+        self.close();
+    }
+}
+
+/// SegmentReader if reader for segment file.
+pub struct SegmentReader {
+    first_idx: u64,
+    last_idx: u64,
+    file: File,
+}
+
+impl SegmentReader {
+    pub fn new<P: AsRef<Path>>(path: P) -> Result<SegmentReader> {
+        unimplemented!();
+    }
+}
+
+trait ReadExt<T> {
+    fn read_be_u64(&mut self) -> std::result::Result<u64, std::io::Error>;
+}
+
+impl<T> ReadExt<T> for T
+where
+    T: Read,
+{
+    fn read_be_u64(&mut self) -> std::result::Result<u64, std::io::Error> {
+        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::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::hash::Hasher;
     use std::io::SeekFrom::Start;
     use std::io::{Read, Seek, Write};
     use tempfile::TempPath;
@@ -180,11 +301,8 @@ mod tests {
     #[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(SyncPolicy::Immediate, temp_file)?;
+        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)?;
@@ -229,14 +347,10 @@ mod tests {
     }
 
     #[test]
-    pub fn test_write_entry_with_cache() -> Result<()> {
+    pub fn test_write_entry_with_num_cache() -> Result<()> {
         let temp_path = tempfile::tempdir()?;
         let file_path = temp_path.path().join("test1");
-        let mut temp_file = OpenOptions::new()
-            .append(true)
-            .create(true)
-            .open(&file_path)?;
-        let mut segment = Segment::new(SyncPolicy::NumBased(5), temp_file)?;
+        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![]));
@@ -261,11 +375,7 @@ mod tests {
     pub fn test_close_entry() -> Result<()> {
         let temp_path = tempfile::tempdir()?;
         let file_path = temp_path.path().join("test1");
-        let mut temp_file = OpenOptions::new()
-            .append(true)
-            .create(true)
-            .open(&file_path)?;
-        let mut segment = Segment::new(SyncPolicy::NumBased(20), temp_file)?;
+        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.
@@ -276,4 +386,32 @@ mod tests {
 
         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
index 275d5c7..377973b 100644
--- a/src/wal/wal.rs
+++ b/src/wal/wal.rs
@@ -1,6 +1,6 @@
 use crate::wal::SyncPolicy;
 use crate::{
-    wal::{segment::Segment, Entry, EntryType},
+    wal::{segment::SegmentWriter, Entry, EntryType},
     Result,
 };
 use std::fs::File;
@@ -9,31 +9,41 @@ use std::path::Path;
 use std::sync::atomic::AtomicU64;
 use std::sync::{Arc, Mutex, RwLock};
 
-type LockedSegment = Arc<RwLock<Segment>>;
+type LockedSegment = Arc<RwLock<SegmentWriter>>;
 
 pub struct WalConfig {
     sync_policy: SyncPolicy,
+    // must be larger than 1.
     initial_seq: u64,
 }
 
-/// Wal manages a dict of segment wal files.
+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,
-    flush_policy: SyncPolicy,
+    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<Arc<Segment>>,
-}
-
-impl<'a> Default for Wal<'a> {
-    fn default() -> Self {
-        unimplemented!()
-    }
+    segments: Vec<Arc<SegmentWriter>>,
 }
 
 impl<'a> Wal<'a> {

From 4497bf2ecb9e6e6646dd054e95b020894b76ae81 Mon Sep 17 00:00:00 2001
From: "zhongyang.wu" <zhongyang.wu@outlook.com>
Date: Mon, 27 Jul 2020 21:28:48 -0400
Subject: [PATCH 4/5] [wal] Fix test. Fix grammar errors.

---
 src/common/label.rs         |  1 -
 src/common/mod.rs           |  4 ++--
 src/compaction/mod.rs       |  6 +++---
 src/indexer/common.rs       |  4 ++--
 src/indexer/sled_indexer.rs |  4 ++--
 src/storage/common.rs       |  1 -
 src/wal/segment.rs          | 13 ++++++++++---
 7 files changed, 19 insertions(+), 14 deletions(-)

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<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<()>;
 }
 
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<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
     }
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<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 {
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/segment.rs b/src/wal/segment.rs
index f61bcc3..1475ad4 100644
--- a/src/wal/segment.rs
+++ b/src/wal/segment.rs
@@ -24,7 +24,7 @@ const AVG_NUM_BYTES_IN_ENTRY: usize = 32;
 ///! | Last seqid  | u64     |
 ///! | CRC64       | u64     |
 
-/// SegmentWriter is writer for segment file
+/// SegmentWriter is the writer for segment file
 ///
 /// SegmentWriter is **NOT** concurrent-safe.
 pub struct SegmentWriter {
@@ -38,7 +38,7 @@ pub struct SegmentWriter {
 }
 
 impl SegmentWriter {
-    /// Create an segment file, open with append-only
+    /// 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
@@ -382,7 +382,14 @@ mod tests {
 
         segment.close();
 
-        assert_eq!(fs::read_to_string(&file_path)?.len(), 8 + 15 + 8); // flushed one entry(15 bytes) and crc 64
+        // 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(())
     }

From 8776844011f3669860e8f180dc7cfddc25588e15 Mon Sep 17 00:00:00 2001
From: "zhongyang.wu" <zhongyang.wu@outlook.com>
Date: Sun, 6 Dec 2020 16:48:56 -0500
Subject: [PATCH 5/5] feat: Add time based SyncCache.

---
 README.md          |  1 -
 src/wal/mod.rs     | 22 ++++++++----
 src/wal/segment.rs | 83 +++++++++++++++++++++++++++++++++++++---------
 3 files changed, 83 insertions(+), 23 deletions(-)

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/src/wal/mod.rs b/src/wal/mod.rs
index 323f09f..95e6dd8 100644
--- a/src/wal/mod.rs
+++ b/src/wal/mod.rs
@@ -1,7 +1,9 @@
-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 wal;
 
@@ -88,11 +90,11 @@ pub enum SyncPolicy {
     Immediate,
 }
 
-/// FlushCache tells segment how to cache bytes
+/// 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,
@@ -108,6 +110,13 @@ pub enum SyncCache {
     None,
 }
 
+// Messages used in time based sync
+pub enum TimeSyncMessage {
+    Insert(Entry),
+    Shutdown,
+    Flush,
+}
+
 pub enum EntryType {
     Default = 0,
 }
@@ -126,9 +135,10 @@ pub enum WalErr {
 
 #[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 1475ad4..df5edeb 100644
--- a/src/wal/segment.rs
+++ b/src/wal/segment.rs
@@ -1,5 +1,5 @@
 use crate::wal::WalErr::FileIoErr;
-use crate::wal::{Entry, SyncCache, SyncPolicy, WalErr, WAL_MAGIC_NUMBER};
+use crate::wal::{Entry, SyncCache, SyncPolicy, WalErr, WAL_MAGIC_NUMBER, TimeSyncMessage};
 use crate::{MonolithErr, Result};
 use futures::io::Error;
 use std::convert::TryInto;
@@ -9,6 +9,7 @@ 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<Mutex<File>>;
 
@@ -32,9 +33,9 @@ pub struct SegmentWriter {
     last_idx: u64,
 
     file: File,
-    cache: SyncCache,
-
     crc: crc::crc64::Digest,
+
+    cache: SyncCache,
 }
 
 impl SegmentWriter {
@@ -89,19 +90,64 @@ impl SegmentWriter {
             file.sync_all();
             Ok((0, 0, crc_val, file))
         }
-        .map_err(|err| WalErr::FileIoErr(err))?;
+            .map_err(|err| WalErr::FileIoErr(err))?;
 
+        let mut crc = crc::crc64::Digest::new_with_initial(crc::crc64::ECMA, crc_initial);
         Ok(SegmentWriter {
             first_idx,
             last_idx,
-            file,
             cache: match sync_policy {
                 SyncPolicy::Immediate => SyncCache::None,
                 SyncPolicy::TimeBased(dur) => {
-                    let cache = Arc::new(Mutex::new(vec![]));
+                    let (main_sender, receiver) = std::sync::mpsc::channel::<TimeSyncMessage>();
+                    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<u8> = cache.iter().flat_map(|entry| entry.get_bytes()).collect();
+                                            crc.write(bytes.as_slice());
+                                            file.write(bytes.as_slice());
+
+                                            // 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<u8> = 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 {
-                        handler: std::thread::spawn(|| {}), //todo: figure out how to do time based flush
-                        cache,
+                        ticker: spawn(move || {
+                            loop {
+                                sleep(dur);
+                                let _ = ticker_sender.send(TimeSyncMessage::Flush);
+                            }
+                        }),
+                        message_handle: main_sender,
                     }
                 }
                 SyncPolicy::NumBased(limit) => SyncCache::NumBased {
@@ -115,7 +161,8 @@ impl SegmentWriter {
                     cache: Vec::with_capacity(limit * AVG_NUM_BYTES_IN_ENTRY),
                 },
             },
-            crc: crc::crc64::Digest::new_with_initial(crc::crc64::ECMA, crc_initial),
+            file,
+            crc,
         })
     }
 
@@ -175,8 +222,8 @@ impl SegmentWriter {
                     self.sync();
                 }
             }
-            &mut SyncCache::TimeBased { ref mut cache, .. } => {
-                cache.lock().unwrap().push(entry);
+            &mut SyncCache::TimeBased { ref message_handle, .. } => {
+                let _ = message_handle.send(TimeSyncMessage::Insert(entry));
             }
             &mut SyncCache::SizeBased {
                 ref mut cache,
@@ -224,8 +271,8 @@ impl Write for SegmentWriter {
                 *cache = vec![Entry::default(); *size];
                 _cache
             }
-            &mut SyncCache::TimeBased { .. } => {
-                // fixme and above
+            &mut SyncCache::TimeBased { ref message_handle, .. } => {
+                let _ = message_handle.send(TimeSyncMessage::Flush);
                 vec![]
             }
             &mut SyncCache::SizeBased {
@@ -252,7 +299,11 @@ impl Write for SegmentWriter {
 
 impl Drop for SegmentWriter {
     fn drop(&mut self) {
-        self.close();
+        if let SyncCache::TimeBased { ref message_handle, .. } = self.cache {
+            message_handle.send(TimeSyncMessage::Shutdown);
+        } else {
+            self.close();
+        }
     }
 }
 
@@ -274,8 +325,8 @@ trait ReadExt<T> {
 }
 
 impl<T> ReadExt<T> for T
-where
-    T: Read,
+    where
+        T: Read,
 {
     fn read_be_u64(&mut self) -> std::result::Result<u64, std::io::Error> {
         let mut buffer = vec![0; 8];