Introduce the Env::nested_read_txn from an RwTxn

.github/workflows/rust.yml

@@ -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

.gitmodules

@@ -1,7 +1,7 @@
 [submodule "lmdb-master-sys/lmdb"]
 	path = lmdb-master-sys/lmdb
-	url = https://git.openldap.org/openldap/openldap
-	branch = mdb.master
+	url = https://github.com/meilisearch/lmdb.git
+	branch = allow-nested-rtxn-from-wtxn
 [submodule "lmdb-master3-sys/lmdb"]
 	path = lmdb-master3-sys/lmdb
 	url = https://git.openldap.org/openldap/openldap

heed/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "heed"
-version = "0.22.0"
+version = "0.22.1-nested-rtxns-7"
 authors = ["Kerollmops <renault.cle@gmail.com>"]
 description = "A fully typed LMDB (mdb.master) wrapper with minimum overhead"
 license = "MIT"
@@ -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-nested-rtxns-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"
 
@@ -122,6 +125,9 @@ name = "custom-dupsort-comparator"
 [[example]]
 name = "multi-env"
 
+[[example]]
+name = "nested-rtxns"
+
 [[example]]
 name = "nested"
 

heed/examples/nested-rtxns.rs

@@ -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..1000 {
+        let mut rng = StdRng::seed_from_u64(i as u64);
+        let max = rng.random_range(10_000..=100_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..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(10_000..=100_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 1000..10_000 {
+        let mut rng = StdRng::seed_from_u64(i as u64);
+        let max = rng.random_range(10_000..=100_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 =
+        (1000..10_000).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(10_000..=100_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(())
+}

heed/src/envs/env.rs

@@ -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;
@@ -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`].
@@ -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 fd of the inner file the file
+    /// is. Doing write operations on the file descriptor can lead to undefined behavior
+    /// and only read-only operations while no write operations are in progress is 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.
@@ -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)
     }
 
@@ -640,6 +658,61 @@ 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(()) }
+    /// ```
+    #[cfg(not(master3))]
+    pub fn nested_read_txn<'p>(&'p self, parent: &'p RwTxn) -> Result<RoTxn<'p, WithoutTls>> {
+        assert_eq_env_txn!(self, parent);
+
+        RoTxn::<WithoutTls>::nested(self, parent)
+    }
+}
+
 impl<T> Clone for Env<T> {
     fn clone(&self) -> Self {
         Env { inner: self.inner.clone(), _tls_marker: PhantomData }

heed/src/envs/mod.rs

@@ -1,9 +1,9 @@
 use std::cmp::Ordering;
 use std::collections::HashMap;
 use std::ffi::c_void;
-use std::fs::{File, Metadata};
+use std::fs::File;
 #[cfg(unix)]
-use std::os::unix::io::{AsRawFd, BorrowedFd, RawFd};
+use std::os::unix::io::{AsRawFd, RawFd};
 use std::panic::catch_unwind;
 use std::path::{Path, PathBuf};
 use std::process::abort;
@@ -12,7 +12,7 @@ use std::time::Duration;
 #[cfg(windows)]
 use std::{
     ffi::OsStr,
-    os::windows::io::{AsRawHandle as _, BorrowedHandle, RawHandle},
+    os::windows::io::{AsRawHandle as _, RawHandle},
 };
 use std::{fmt, io};
 
@@ -149,22 +149,6 @@ fn get_file_fd(file: &File) -> RawHandle {
     file.as_raw_handle()
 }
 
-#[cfg(unix)]
-/// Get metadata from a file descriptor.
-unsafe fn metadata_from_fd(raw_fd: RawFd) -> io::Result<Metadata> {
-    let fd = BorrowedFd::borrow_raw(raw_fd);
-    let owned = fd.try_clone_to_owned()?;
-    File::from(owned).metadata()
-}
-
-#[cfg(windows)]
-/// Get metadata from a file descriptor.
-unsafe fn metadata_from_fd(raw_fd: RawHandle) -> io::Result<Metadata> {
-    let fd = BorrowedHandle::borrow_raw(raw_fd);
-    let owned = fd.try_clone_to_owned()?;
-    File::from(owned).metadata()
-}
-
 /// A helper function that transforms the LMDB types into Rust types (`MDB_val` into slices)
 /// and vice versa, the Rust types into C types (`Ordering` into an integer).
 ///