baildon 0.1.2

B+Tree library
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

//! WAL functions
//!

use std::path::Path;
use std::time::Duration;

use anyhow::Result;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tokio::fs::{File, OpenOptions};
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[derive(Debug)]
pub(crate) struct WalFile {
    file: File,
    sync_allowed: Arc<AtomicBool>,
}

impl WalFile {
    pub(crate) async fn try_open(path: &Path) -> Result<Self> {
        let file = OpenOptions::new()
            .create(false)
            .read(true)
            .write(false)
            .open(path)
            .await?;

        Ok(Self {
            file,
            sync_allowed: Arc::new(AtomicBool::default()),
        })
    }

    pub(crate) async fn try_new(path: &Path) -> Result<Self> {
        let file = OpenOptions::new()
            .create_new(true)
            .read(true)
            .append(true)
            .open(path)
            .await?;

        let sync_allowed = Arc::new(AtomicBool::default());
        let shared_sync = sync_allowed.clone();
        tokio::spawn(async move {
            // Re-enable flushing every 2 seconds
            let mut timer = tokio::time::interval(Duration::from_secs(2));
            loop {
                timer.tick().await;
                shared_sync.store(true, Ordering::Release);
            }
            #[allow(unreachable_code)]
            Ok::<(), anyhow::Error>(()) // <- note the explicit type annotation here
        });
        Ok(Self { file, sync_allowed })
    }

    pub(crate) async fn flush(&mut self) -> Result<()> {
        // To prevent excessive flushing, we only flush if our allowed flag is true
        if self
            .sync_allowed
            .compare_exchange(true, false, Ordering::Acquire, Ordering::Relaxed)
            .is_ok()
        {
            self.file.sync_all().await.map_err(|e| e.into())
        } else {
            Ok(())
        }
    }

    pub(crate) async fn write_data(&mut self, data: &[u8]) -> Result<()> {
        self.file.write_u64(data.len() as u64).await?;
        self.file.write_all(data).await?;
        self.flush().await
    }

    pub(crate) async fn read_data(&mut self) -> Result<Vec<u8>> {
        let len = self.file.read_u64().await?;
        let mut buf = vec![0; len as usize];
        let _ = self.file.read_exact(&mut buf).await?;
        Ok(buf)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use crate::command::Command;

    #[tokio::test]
    async fn it_creates_wal_file() {
        let _wal = WalFile::try_new(Path::new("wal_file_create.db"))
            .await
            .expect("creates wal file");
        std::fs::remove_file("wal_file_create.db").expect("cleanup");
    }

    #[tokio::test]
    async fn it_opens_wal_file() {
        let mut wal = WalFile::try_new(Path::new("wal_file_open.db"))
            .await
            .expect("creates wal file");
        wal.flush().await.expect("flushed away");
        drop(wal);
        let _wal = WalFile::try_open(Path::new("wal_file_open.db"))
            .await
            .expect("opens wal file");
        std::fs::remove_file("wal_file_open.db").expect("cleanup");
    }

    #[tokio::test]
    async fn it_writes_to_wal_file() {
        let mut wal = WalFile::try_new(Path::new("wal_file_write.db"))
            .await
            .expect("creates wal file");
        let upsert = Command::Upsert("key".to_string(), "value".to_string());
        let data = upsert.serialize().expect("serializes");
        wal.write_data(&data).await.expect("write data");

        drop(wal);

        let mut wal = WalFile::try_open(Path::new("wal_file_write.db"))
            .await
            .expect("opens wal file");
        let new_upsert = Command::deserialize(&wal.read_data().await.expect("reads data"))
            .expect("deserializes");
        assert_eq!(upsert, new_upsert);
        std::fs::remove_file("wal_file_write.db").expect("cleanup");
    }
}