Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion .github/workflows/rust.yml
Original file line number Diff line number Diff line change
Expand Up @@ -159,8 +159,10 @@ jobs:
run: |
cargo clean
# rmp-serde needs a feature activated, so we'll just run it separately.
cargo run --example 2>&1 | grep -E '^ ' | awk '!/rmp-serde/' | xargs -n1 cargo run --example
# We also want to run the nested-rtxns example in release.
cargo run --example 2>&1 | grep -E '^ ' | awk '!/rmp-serde|nested-rtxns/' | xargs -n1 cargo run --example
cargo run --example rmp-serde --features serde-rmp
cargo run --release --example nested-rtxns

heed3-examples:
name: Run the heed3 examples
Expand Down
4 changes: 2 additions & 2 deletions .gitmodules
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
[submodule "lmdb-master-sys/lmdb"]
path = lmdb-master-sys/lmdb
url = https://github.com/openldap/openldap.git
branch = mdb.master
url = https://github.com/meilisearch/lmdb.git
branch = mdb.master.nested-rtxns
[submodule "lmdb-master3-sys/lmdb"]
path = lmdb-master3-sys/lmdb
url = https://github.com/openldap/openldap.git
Expand Down
10 changes: 8 additions & 2 deletions heed/Cargo.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[package]
name = "heed"
version = "0.22.0"
version = "0.22.1"
authors = ["Kerollmops <renault.cle@gmail.com>"]
description = "A fully typed LMDB (mdb.master) wrapper with minimum overhead"
license = "MIT"
Expand All @@ -16,14 +16,17 @@ byteorder = { version = "1.5.0", default-features = false }
heed-traits = { version = "0.20.0", path = "../heed-traits" }
heed-types = { version = "0.21.0", default-features = false, path = "../heed-types" }
libc = "0.2.175"
lmdb-master-sys = { version = "0.2.5", path = "../lmdb-master-sys" }
lmdb-master-sys = { version = "0.2.6", path = "../lmdb-master-sys" }
once_cell = "1.21.3"
page_size = "0.6.0"
serde = { version = "1.0.223", features = ["derive"], optional = true }
synchronoise = "1.0.1"

[dev-dependencies]
memchr = "2.7.5"
rand = "0.9.0"
rayon = "1.10.0"
roaring = "0.10.10"
serde = { version = "1.0.223", features = ["derive"] }
tempfile = "3.22.0"

Expand Down Expand Up @@ -122,6 +125,9 @@ name = "custom-dupsort-comparator"
[[example]]
name = "multi-env"

[[example]]
name = "nested-rtxns"

[[example]]
name = "nested"

Expand Down
80 changes: 80 additions & 0 deletions heed/examples/nested-rtxns.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
use heed::types::*;
use heed::{Database, EnvFlags, EnvOpenOptions};
use rand::prelude::*;
use rayon::prelude::*;
use roaring::RoaringBitmap;

fn main() -> Result<(), Box<dyn std::error::Error>> {
let dir = tempfile::tempdir()?;
let env = unsafe {
let mut options = EnvOpenOptions::new().read_txn_without_tls();
#[cfg(not(windows))]
options.flags(EnvFlags::WRITE_MAP);
options
.map_size(2 * 1024 * 1024 * 1024) // 2 GiB
.open(dir.path())?
};

// opening a write transaction
let mut wtxn = env.write_txn()?;
// we will open the default unnamed database
let db: Database<U32<byteorder::BigEndian>, Bytes> = env.create_database(&mut wtxn, None)?;

let mut buffer = Vec::new();
for i in 0..100 {
let mut rng = StdRng::seed_from_u64(i as u64);
let max = rng.random_range(1000..=10_000);
let roaring = RoaringBitmap::from_sorted_iter(0..max)?;
buffer.clear();
roaring.serialize_into(&mut buffer)?;
db.put(&mut wtxn, &i, &buffer)?;
}

// opening multiple read-only transactions
// to check if those values are now available
// without committing beforehand
let rtxns = (0..100).map(|_| env.nested_read_txn(&wtxn)).collect::<heed::Result<Vec<_>>>()?;

rtxns.into_par_iter().enumerate().for_each(|(i, rtxn)| {
let mut rng = StdRng::seed_from_u64(i as u64);
let max = rng.random_range(1000..=10_000);
let roaring = RoaringBitmap::from_sorted_iter(0..max).unwrap();

let mut buffer = Vec::new();
roaring.serialize_into(&mut buffer).unwrap();

let i = i as u32;
let ret = db.get(&rtxn, &i).unwrap();
assert_eq!(ret, Some(&buffer[..]));
});

for i in 100..1000 {
let mut rng = StdRng::seed_from_u64(i as u64);
let max = rng.random_range(1000..=10_000);
let roaring = RoaringBitmap::from_sorted_iter(0..max)?;
buffer.clear();
roaring.serialize_into(&mut buffer)?;
db.put(&mut wtxn, &i, &buffer)?;
}

// opening multiple read-only transactions
// to check if those values are now available
// without committing beforehand
let rtxns =
(100..1000).map(|_| env.nested_read_txn(&wtxn)).collect::<heed::Result<Vec<_>>>()?;

rtxns.into_par_iter().enumerate().for_each(|(i, rtxn)| {
let mut rng = StdRng::seed_from_u64(i as u64);
let max = rng.random_range(1000..=10_000);
let roaring = RoaringBitmap::from_sorted_iter(0..max).unwrap();

let mut buffer = Vec::new();
roaring.serialize_into(&mut buffer).unwrap();

let i = i as u32;
let ret = db.get(&rtxn, &i).unwrap();
assert_eq!(ret, Some(&buffer[..]));
});

Ok(())
}
73 changes: 72 additions & 1 deletion heed/src/envs/encrypted_env.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ use super::{Env, EnvClosingEvent, EnvInfo, FlagSetMode};
use crate::databases::{EncryptedDatabase, EncryptedDatabaseOpenOptions};
use crate::envs::EnvStat;
use crate::mdb::ffi::{self};
use crate::{CompactionOption, EnvFlags, Result, RoTxn, RwTxn, Unspecified, WithTls};
use crate::{CompactionOption, EnvFlags, Result, RoTxn, RwTxn, Unspecified, WithTls, WithoutTls};
#[allow(unused)] // fro cargo auto doc links
use crate::{Database, EnvOpenOptions};

Expand Down Expand Up @@ -373,6 +373,77 @@ impl<T> EncryptedEnv<T> {
}
}

impl EncryptedEnv<WithoutTls> {
/// Create a nested read transaction that is capable of reading uncommitted changes.
///
/// The new transaction will be a nested transaction, with the transaction indicated by parent
/// as its parent. Transactions may be nested to any level.
///
/// This is a custom LMDB fork feature that allows reading uncommitted changes.
/// It enables parallel processing of data across multiple threads through
/// concurrent read-only transactions. You can [read more in this PR](https://github.com/meilisearch/heed/pull/307).
///
/// ```
/// use std::fs;
/// use std::path::Path;
/// use argon2::Argon2;
/// use chacha20poly1305::{ChaCha20Poly1305, Key};
/// use heed::{EnvOpenOptions, EncryptedDatabase};
/// use heed::types::*;
///
/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let dir = tempfile::tempdir()?;
/// let env = unsafe {
/// EnvOpenOptions::new()
/// .read_txn_without_tls()
/// .map_size(2 * 1024 * 1024) // 2 MiB
/// .open(dir.path())?
/// };
///
/// let env_path = tempfile::tempdir()?;
/// let password = "This is the password that will be hashed by the argon2 algorithm";
/// let salt = "The salt added to the password hashes to add more security when stored";
///
/// // We choose to use argon2 as our Key Derivation Function, but you can choose whatever you want.
/// // <https://github.com/RustCrypto/traits/tree/master/password-hash#supported-crates>
/// let mut key = Key::default();
/// Argon2::default().hash_password_into(password.as_bytes(), salt.as_bytes(), &mut key)?;
///
/// // We open the environment
/// let env = unsafe {
/// let mut options = EnvOpenOptions::new().read_txn_without_tls();
/// options
/// .map_size(2 * 1024 * 1024 * 1024) // 2 GiB
/// .open_encrypted::<ChaCha20Poly1305, _>(key, &env_path)?
/// };
///
/// // we will open the default unnamed database
/// let mut wtxn = env.write_txn()?;
/// let db: EncryptedDatabase<U32<byteorder::BigEndian>, U32<byteorder::BigEndian>> = env.create_database(&mut wtxn, None)?;
///
/// // opening a write transaction
/// for i in 0..1000 {
/// db.put(&mut wtxn, &i, &i)?;
/// }
///
/// // opening multiple read-only transactions
/// // to check if those values are now available
/// // without committing beforehand
/// let rtxns = (0..1000).map(|_| env.nested_read_txn(&wtxn)).collect::<heed::Result<Vec<_>>>()?;
///
/// for (i, mut rtxn) in rtxns.into_iter().enumerate() {
/// let i = i as u32;
/// let ret = db.get(&mut rtxn, &i)?;
/// assert_eq!(ret, Some(i));
/// }
///
/// # Ok(()) }
/// ```
pub fn nested_read_txn<'p>(&'p self, parent: &'p RwTxn) -> Result<RoTxn<'p, WithoutTls>> {
self.inner.nested_read_txn(parent)
}
}

unsafe impl<T> Send for EncryptedEnv<T> {}
unsafe impl<T> Sync for EncryptedEnv<T> {}

Expand Down
86 changes: 79 additions & 7 deletions heed/src/envs/env.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,8 @@ use heed_traits::Comparator;
use synchronoise::SignalEvent;

use super::{
custom_key_cmp_wrapper, get_file_fd, metadata_from_fd, DefaultComparator, EnvClosingEvent,
EnvInfo, FlagSetMode, IntegerComparator, OPENED_ENV,
custom_key_cmp_wrapper, get_file_fd, DefaultComparator, EnvClosingEvent, EnvInfo, FlagSetMode,
IntegerComparator, OPENED_ENV,
};
use crate::cursor::{MoveOperation, RoCursor};
use crate::envs::EnvStat;
Expand All @@ -23,8 +23,8 @@ use crate::mdb::lmdb_flags::AllDatabaseFlags;
#[allow(unused)] // for cargo auto doc links
use crate::EnvOpenOptions;
use crate::{
CompactionOption, Database, DatabaseOpenOptions, EnvFlags, Error, Result, RoTxn, RwTxn,
Unspecified, WithTls,
assert_eq_env_txn, CompactionOption, Database, DatabaseOpenOptions, EnvFlags, Error, Result,
RoTxn, RwTxn, Unspecified, WithTls, WithoutTls,
};

/// An environment handle constructed by using [`EnvOpenOptions::open`].
Expand Down Expand Up @@ -64,11 +64,27 @@ impl<T> Env<T> {
/// # Ok(()) }
/// ```
pub fn real_disk_size(&self) -> Result<u64> {
Ok(self.try_clone_inner_file()?.metadata()?.len())
}

/// Try cloning the inner file used in the environment and return a `File`
/// corresponding to the environment file.
///
/// # Safety
///
/// This function is safe as we are creating a cloned file descriptor of the inner file.
/// Doing write operations on the file descriptor can lead to undefined behavior,
/// and only read-only operations while no write operations are in progress are safe.
pub fn try_clone_inner_file(&self) -> Result<File> {
let mut fd = mem::MaybeUninit::uninit();
unsafe { mdb_result(ffi::mdb_env_get_fd(self.env_mut_ptr().as_mut(), fd.as_mut_ptr()))? };
let fd = unsafe { fd.assume_init() };
let metadata = unsafe { metadata_from_fd(fd)? };
Ok(metadata.len())
let raw_fd = unsafe { fd.assume_init() };
#[cfg(unix)]
let fd = unsafe { std::os::fd::BorrowedFd::borrow_raw(raw_fd) };
#[cfg(windows)]
let fd = unsafe { std::os::windows::io::BorrowedHandle::borrow_raw(raw_fd) };
let owned = fd.try_clone_to_owned()?;
Ok(File::from(owned))
}

/// Return the raw flags the environment was opened with.
Expand Down Expand Up @@ -376,6 +392,8 @@ impl<T> Env<T> {
/// A parent transaction and its cursors may not issue any other operations than _commit_ and
/// _abort_ while it has active child transactions.
pub fn nested_write_txn<'p>(&'p self, parent: &'p mut RwTxn) -> Result<RwTxn<'p>> {
assert_eq_env_txn!(self, parent);

RwTxn::nested(self, parent)
}

Expand Down Expand Up @@ -640,6 +658,60 @@ impl<T> Env<T> {
}
}

impl Env<WithoutTls> {
/// Create a nested read transaction that is capable of reading uncommitted changes.
///
/// The new transaction will be a nested transaction, with the transaction indicated by parent
/// as its parent. Transactions may be nested to any level.
///
/// This is a custom LMDB fork feature that allows reading uncommitted changes.
/// It enables parallel processing of data across multiple threads through
/// concurrent read-only transactions. You can [read more in this PR](https://github.com/meilisearch/heed/pull/307).
///
/// ```
/// use std::fs;
/// use std::path::Path;
/// use heed::{EnvOpenOptions, Database};
/// use heed::types::*;
///
/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let dir = tempfile::tempdir()?;
/// let env = unsafe {
/// EnvOpenOptions::new()
/// .read_txn_without_tls()
/// .map_size(2 * 1024 * 1024) // 2 MiB
/// .open(dir.path())?
/// };
///
/// // we will open the default unnamed database
/// let mut wtxn = env.write_txn()?;
/// let db: Database<U32<byteorder::BigEndian>, U32<byteorder::BigEndian>> = env.create_database(&mut wtxn, None)?;
///
/// // opening a write transaction
/// for i in 0..1000 {
/// db.put(&mut wtxn, &i, &i)?;
/// }
///
/// // opening multiple read-only transactions
/// // to check if those values are now available
/// // without committing beforehand
/// let rtxns = (0..1000).map(|_| env.nested_read_txn(&wtxn)).collect::<heed::Result<Vec<_>>>()?;
///
/// for (i, rtxn) in rtxns.iter().enumerate() {
/// let i = i as u32;
/// let ret = db.get(&rtxn, &i)?;
/// assert_eq!(ret, Some(i));
/// }
///
/// # Ok(()) }
/// ```
pub fn nested_read_txn<'p>(&'p self, parent: &'p RwTxn) -> Result<RoTxn<'p, WithoutTls>> {
assert_eq_env_txn!(self, parent);

parent.nested_read_txn()
}
}

impl<T> Clone for Env<T> {
fn clone(&self) -> Self {
Env { inner: self.inner.clone(), _tls_marker: PhantomData }
Expand Down
10 changes: 5 additions & 5 deletions heed/src/envs/env_open_options.rs
Original file line number Diff line number Diff line change
Expand Up @@ -332,8 +332,8 @@ impl<T: TlsUsage> EnvOpenOptions<T> {
/// use std::path::Path;
/// use argon2::Argon2;
/// use chacha20poly1305::{ChaCha20Poly1305, Key};
/// use heed3_encryption::types::*;
/// use heed3_encryption::{EnvOpenOptions, Database};
/// use heed3::types::*;
/// use heed3::{EnvOpenOptions, EncryptedDatabase;};
///
/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let env_path = tempfile::tempdir()?;
Expand All @@ -346,20 +346,20 @@ impl<T: TlsUsage> EnvOpenOptions<T> {
/// Argon2::default().hash_password_into(password.as_bytes(), salt.as_bytes(), &mut key)?;
///
/// // We open the environment
/// let mut options = EnvOpenOptions::<ChaCha20Poly1305>::new_encrypted_with(key);
/// let mut options = EnvOpenOptions::new();
/// let env = unsafe {
/// options
/// .map_size(10 * 1024 * 1024) // 10MB
/// .max_dbs(3)
/// .open(&env_path)?
/// .open_encrypted::<ChaCha20Poly1305, _>(key, &env_path)?
/// };
///
/// let key1 = "first-key";
/// let key2 = "second-key";
///
/// // We create the database
/// let mut wtxn = env.write_txn()?;
/// let db: Database<Str, Str> = env.create_database(&mut wtxn, Some("first"))?;
/// let db: EncryptedDatabase<Str, Str> = env.create_database(&mut wtxn, Some("first"))?;
/// wtxn.commit()?;
///
/// // Declare the read transaction as mutable because LMDB, when using encryption,
Expand Down
Loading
Loading