add locks + file alloc in Writer::finish

2023-04-05 16:53:40 +03:00 · 2023-04-05 16:53:40 +03:00 · 39d4d6b610
commit 39d4d6b610
parent 27260695f7
4 changed files with 96 additions and 21 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@ -158,6 +158,7 @@ name = "chgk_ledb_lib"
 version = "1.1.0"
 dependencies = [
 "bincode",
+ "fs4",
 "memmap",
 "serde",
 "serde_derive",
@ -372,6 +373,17 @@ version = "1.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7fcaabb2fef8c910e7f4c7ce9f67a1283a1715879a7c230ca9d6d1ae31f16d91"

+[[package]]
+name = "errno"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f639046355ee4f37944e44f60642c6f3a7efa3cf6b78c78a0d989a8ce6c396a1"
+dependencies = [
+ "errno-dragonfly",
+ "libc",
+ "winapi",
+]
+
 [[package]]
 name = "errno"
 version = "0.3.0"
@ -380,7 +392,7 @@ checksum = "50d6a0976c999d473fe89ad888d5a284e55366d9dc9038b1ba2aa15128c4afa0"
 dependencies = [
 "errno-dragonfly",
 "libc",
- "windows-sys",
+ "windows-sys 0.45.0",
 ]

 [[package]]
@ -412,6 +424,17 @@ dependencies = [
 "miniz_oxide",
 ]

+[[package]]
+name = "fs4"
+version = "0.6.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8ea55201cc351fdb478217c0fb641b59813da9b4efe4c414a9d8f989a657d149"
+dependencies = [
+ "libc",
+ "rustix 0.35.13",
+ "winapi",
+]
+
 [[package]]
 name = "generic-array"
 version = "0.14.7"
@ -503,6 +526,12 @@ dependencies = [
 "cfg-if",
 ]

+[[package]]
+name = "io-lifetimes"
+version = "0.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "59ce5ef949d49ee85593fc4d3f3f95ad61657076395cbbce23e2121fc5542074"
+
 [[package]]
 name = "io-lifetimes"
 version = "1.0.9"
@ -511,7 +540,7 @@ checksum = "09270fd4fa1111bc614ed2246c7ef56239a3063d5be0d1ec3b589c505d400aeb"
 dependencies = [
 "hermit-abi 0.3.1",
 "libc",
- "windows-sys",
+ "windows-sys 0.45.0",
 ]

 [[package]]
@ -559,6 +588,12 @@ version = "0.2.140"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "99227334921fae1a979cf0bfdfcc6b3e5ce376ef57e16fb6fb3ea2ed6095f80c"

+[[package]]
+name = "linux-raw-sys"
+version = "0.0.46"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d4d2456c373231a208ad294c33dc5bff30051eafd954cd4caae83a712b12854d"
+
 [[package]]
 name = "linux-raw-sys"
 version = "0.3.1"
@ -826,6 +861,20 @@ version = "0.6.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f162c6dd7b008981e4d40210aca20b4bd0f9b60ca9271061b07f78537722f2e1"

+[[package]]
+name = "rustix"
+version = "0.35.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "727a1a6d65f786ec22df8a81ca3121107f235970dc1705ed681d3e6e8b9cd5f9"
+dependencies = [
+ "bitflags",
+ "errno 0.2.8",
+ "io-lifetimes 0.7.5",
+ "libc",
+ "linux-raw-sys 0.0.46",
+ "windows-sys 0.42.0",
+]
+
 [[package]]
 name = "rustix"
 version = "0.37.6"
@ -833,11 +882,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d097081ed288dfe45699b72f5b5d648e5f15d64d900c7080273baa20c16a6849"
 dependencies = [
 "bitflags",
- "errno",
- "io-lifetimes",
+ "errno 0.3.0",
+ "io-lifetimes 1.0.9",
 "libc",
- "linux-raw-sys",
- "windows-sys",
+ "linux-raw-sys 0.3.1",
+ "windows-sys 0.45.0",
 ]

 [[package]]
@ -957,8 +1006,8 @@ dependencies = [
 "cfg-if",
 "fastrand",
 "redox_syscall",
- "rustix",
- "windows-sys",
+ "rustix 0.37.6",
+ "windows-sys 0.45.0",
 ]

 [[package]]
@ -1148,6 +1197,21 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"

+[[package]]
+name = "windows-sys"
+version = "0.42.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a3e1820f08b8513f676f7ab6c1f99ff312fb97b553d30ff4dd86f9f15728aa7"
+dependencies = [
+ "windows_aarch64_gnullvm",
+ "windows_aarch64_msvc",
+ "windows_i686_gnu",
+ "windows_i686_msvc",
+ "windows_x86_64_gnu",
+ "windows_x86_64_gnullvm",
+ "windows_x86_64_msvc",
+]
+
 [[package]]
 name = "windows-sys"
 version = "0.45.0"
--- a/app/benches/db_bench.rs
+++ b/app/benches/db_bench.rs
@ -7,12 +7,12 @@ extern crate serde_json;
 extern crate tempfile;

 use chgk_ledb_lib::db;
-use std::path::PathBuf;
+use std::path::Path;

 use db::{Reader, Writer, WriterOpts};

 use criterion::{BatchSize, Criterion};
-use tempfile::tempdir;
+use tempfile::{tempdir, NamedTempFile};

 use serde_derive::{Deserialize, Serialize};

@ -37,17 +37,14 @@ struct TestData {
 const N: usize = 4096;

 fn gen_data(count: usize) -> impl Iterator<Item = TestData> {
-    (0..count)
-        .into_iter()
-        .map(|i| 143 + i as u64)
-        .map(|i| TestData {
-            num1: i,
-            num2: i * 100 ^ 0xDF0E441122334455,
-            test: "test ---- Test ____".repeat(123 + i as usize % 15),
-        })
+    (0..count).map(|i| 143 + i as u64).map(|i| TestData {
+        num1: i,
+        num2: (i * 100) ^ 0xDF0E441122334455,
+        test: "test ---- Test ____".repeat(123 + i as usize % 15),
+    })
 }

-fn prepare_db_writer(path: &PathBuf) -> Writer<TestData> {
+fn prepare_db_writer<P: AsRef<Path>>(path: P) -> Writer<TestData> {
    let opts = WriterOpts {
        compress_lvl: 1,
        data_buf_size: 100 * 1024 * 1024,
@ -60,7 +57,9 @@ fn prepare_db_writer(path: &PathBuf) -> Writer<TestData> {

 fn db_read(c: &mut Criterion) {
    let dir = tempdir().expect("tempdir");
-    let tmpfile = dir.path().join("test.tmp");
+    let tmpfile = NamedTempFile::new_in(dir.path())
+        .expect("new tempfile")
+        .into_temp_path();
    let mut writer = prepare_db_writer(&tmpfile);

    let mut items_iter = gen_data(N).collect::<Vec<TestData>>().into_iter();
@ -85,10 +84,13 @@ fn db_read(c: &mut Criterion) {

 fn db_write(c: &mut Criterion) {
    let dir = tempdir().expect("tempdir");
-    let tmpfile = dir.path().join("test.tmp");
+
    c.bench_function("write", |b| {
        b.iter_batched(
            || {
+                let tmpfile = NamedTempFile::new_in(dir.path())
+                    .expect("new tempfile")
+                    .into_temp_path();
                let src = gen_data(N).collect::<Vec<TestData>>().into_iter();
                let writer = prepare_db_writer(&tmpfile);
                (src, writer)
--- a/lib/Cargo.toml
+++ b/lib/Cargo.toml
@ -17,6 +17,7 @@ zip="0.6"
 bincode = "^2.0.0-rc.2"
 zstd = "^0.10"
 memmap = "0.7.0"
+fs4 = { version = "0.6.3", features = ["sync"] }

 [dev-dependencies]
 tempfile = "3.3"
--- a/lib/src/db.rs
+++ b/lib/src/db.rs
@ -8,6 +8,8 @@ use std::{

 use memmap::{Mmap, MmapOptions};

+use fs4::FileExt;
+
 type LSize = u32;
 const LEN_SIZE: usize = std::mem::size_of::<LSize>();
 const BINCODE_CFG: bincode::config::Configuration = bincode::config::standard();
@ -63,6 +65,7 @@ where
 {
    pub fn new<P: AsRef<Path>>(path: P, opts: WriterOpts) -> Result<Self, String> {
        let out = fs::File::create(path).str_err()?;
+        out.try_lock_exclusive().str_err()?;
        let out = io::BufWriter::with_capacity(opts.out_buf_size, out);
        let data_buf: Vec<u8> = Vec::with_capacity(opts.data_buf_size);
        let data_buf = Cursor::new(data_buf);
@ -133,6 +136,9 @@ where
        let pos: LSize = self.data_buf.position() as LSize;
        self.table.push(pos);

+        let output_size: u64 = (self.table.len() * LEN_SIZE) as u64 + self.data_buf.position();
+        self.out.get_ref().allocate(output_size).str_err()?;
+
        // write tab
        let tab_size = (self.table.len() * LEN_SIZE) as LSize;
        for pos in self.table {
@ -147,6 +153,7 @@ where
        io::copy(&mut data, &mut self.out).str_err()?;

        self.out.flush().str_err()?;
+        self.out.get_ref().unlock().str_err()?;
        Ok(())
    }
 }
@ -167,6 +174,7 @@ where
 {
    pub fn new<P: AsRef<Path>>(path: P, _buf_size: usize) -> Result<Self, String> {
        let file = fs::File::open(path).str_err()?;
+        file.try_lock_shared().str_err()?;
        let mmap = unsafe { MmapOptions::new().map(&file).str_err()? };

        // read first pos and records count