lsm 0.4.1

An implementation of log-structured merge trees in pure Rust
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

#[cfg(feature = "async-io")]
use tokio_uring::fs;

#[cfg(not(feature = "async-io"))]
use std::fs;

#[cfg(not(feature = "async-io"))]
use std::io::{Read, Seek, Write};

use std::path::Path;

use cfg_if::cfg_if;

/// Read from the offset to the end of the file
/// Not supported by tokio-uring yet, so added as a helper function here
#[cfg(feature = "async-io")]
async fn read_to_end(file: &fs::File, offset: u64) -> Result<Vec<u8>, std::io::Error> {
    let mut buffer = vec![0u8; 4096];
    let mut result = vec![];
    let mut pos = offset;

    loop {
        let (res, buf) = file.read_at(buffer, pos).await;

        match res {
            Ok(0) => return Ok(result),
            Ok(n) => {
                buffer = buf;
                result.extend_from_slice(&buffer[..n]);
                pos += n as u64;
            }
            Err(err) => return Err(err),
        }
    }
}

#[inline(always)]
#[tracing::instrument]
pub async fn read_uncompressed(fpath: &Path, offset: u64) -> Result<Vec<u8>, std::io::Error> {
    cfg_if! {
        if #[ cfg(feature="async-io") ] {
            let file = fs::File::open(fpath).await?;
            let buf = read_to_end(&file, offset).await?;
        } else {
            let mut file = fs::File::open(fpath)?;

            if offset > 0 {
                file.seek(std::io::SeekFrom::Start(offset))?;
            }

            let mut buf = vec![];
            file.read_to_end(&mut buf)?;

        }
    }

    Ok(buf)
}

#[inline(always)]
#[tracing::instrument]
pub async fn read(fpath: &Path, offset: u64) -> Result<Vec<u8>, std::io::Error> {
    let compressed = read_uncompressed(fpath, offset).await?;

    cfg_if! {
        if #[ cfg(feature="snappy-compression") ] {
            let mut decoder = snap::raw::Decoder::new();
            Ok(decoder.decompress_vec(&compressed)?)
        } else {
            Ok(compressed)
        }
    }
}

#[tracing::instrument(skip(data))]
#[inline(always)]
pub async fn write(fpath: &Path, data: &[u8]) -> Result<(), std::io::Error> {
    //TODO it might be worth investigating if encoding/decoding
    // chunks is more efficient

    cfg_if! {
        if #[cfg(feature="snappy-compression") ] {
            let mut encoder = snap::raw::Encoder::new();
            let compressed = encoder.compress_vec(data)
                .expect("Failed to compress data");
        } else {
            let mut compressed = vec![];
            compressed.extend_from_slice(data);
        }
    }

    write_uncompressed(fpath, compressed).await
}

#[tracing::instrument(skip(data))]
#[inline(always)]
pub async fn write_uncompressed(fpath: &Path, data: Vec<u8>) -> Result<(), std::io::Error> {
    cfg_if! {
        if #[ cfg(feature="async-io") ] {
            let file = fs::OpenOptions::new().create(true)
                .truncate(true).write(true)
                .open(fpath).await?;

            let (res, _buf) = file.write_all_at(data, 0).await;
            res?;
            file.sync_all().await?;
        } else {
            let mut file = fs::OpenOptions::new().create(true)
                .truncate(true).write(true)
                .open(fpath)?;

            file.write_all(&data)?;
            file.sync_all()?;
        }
    }

    Ok(())
}