reth_libmdbx/
transaction.rs

Help
1use crate::{
2    database::Database,
3    environment::Environment,
4    error::{mdbx_result, Result},
5    flags::{DatabaseFlags, WriteFlags},
6    txn_manager::{TxnManagerMessage, TxnPtr},
7    Cursor, Error, Stat, TableObject,
8};
9use ffi::{MDBX_txn_flags_t, MDBX_TXN_RDONLY, MDBX_TXN_READWRITE};
10use indexmap::IndexSet;
11use parking_lot::{Mutex, MutexGuard};
12use std::{
13    ffi::{c_uint, c_void},
14    fmt::{self, Debug},
15    mem::size_of,
16    ptr, slice,
17    sync::{atomic::AtomicBool, mpsc::sync_channel, Arc},
18    time::Duration,
19};
20
21#[cfg(feature = "read-tx-timeouts")]
22use ffi::mdbx_txn_renew;
23
24mod private {
25    use super::*;
26
27    pub trait Sealed {}
28
29    impl Sealed for RO {}
30    impl Sealed for RW {}
31}
32
33pub trait TransactionKind: private::Sealed + Send + Sync + Debug + 'static {
34    #[doc(hidden)]
35    const OPEN_FLAGS: MDBX_txn_flags_t;
36
37    /// Convenience flag for distinguishing between read-only and read-write transactions.
38    #[doc(hidden)]
39    const IS_READ_ONLY: bool;
40}
41
42#[derive(Debug)]
43#[non_exhaustive]
44pub struct RO;
45
46#[derive(Debug)]
47#[non_exhaustive]
48pub struct RW;
49
50impl TransactionKind for RO {
51    const OPEN_FLAGS: MDBX_txn_flags_t = MDBX_TXN_RDONLY;
52    const IS_READ_ONLY: bool = true;
53}
54impl TransactionKind for RW {
55    const OPEN_FLAGS: MDBX_txn_flags_t = MDBX_TXN_READWRITE;
56    const IS_READ_ONLY: bool = false;
57}
58
59/// An MDBX transaction.
60///
61/// All database operations require a transaction.
62pub struct Transaction<K>
63where
64    K: TransactionKind,
65{
66    inner: Arc<TransactionInner<K>>,
67}
68
69impl<K> Transaction<K>
70where
71    K: TransactionKind,
72{
73    pub(crate) fn new(env: Environment) -> Result<Self> {
74        let mut txn: *mut ffi::MDBX_txn = ptr::null_mut();
75        unsafe {
76            mdbx_result(ffi::mdbx_txn_begin_ex(
77                env.env_ptr(),
78                ptr::null_mut(),
79                K::OPEN_FLAGS,
80                &mut txn,
81                ptr::null_mut(),
82            ))?;
83            Ok(Self::new_from_ptr(env, txn))
84        }
85    }
86
87    pub(crate) fn new_from_ptr(env: Environment, txn_ptr: *mut ffi::MDBX_txn) -> Self {
88        let txn = TransactionPtr::new(txn_ptr);
89
90        #[cfg(feature = "read-tx-timeouts")]
91        if K::IS_READ_ONLY {
92            env.txn_manager().add_active_read_transaction(txn_ptr, txn.clone())
93        }
94
95        let inner = TransactionInner {
96            txn,
97            primed_dbis: Mutex::new(IndexSet::new()),
98            committed: AtomicBool::new(false),
99            env,
100            _marker: Default::default(),
101        };
102
103        Self { inner: Arc::new(inner) }
104    }
105
106    /// Executes the given closure once the lock on the transaction is acquired.
107    ///
108    /// The caller **must** ensure that the pointer is not used after the
109    /// lifetime of the transaction.
110    #[inline]
111    pub fn txn_execute<F, T>(&self, f: F) -> Result<T>
112    where
113        F: FnOnce(*mut ffi::MDBX_txn) -> T,
114    {
115        self.inner.txn_execute(f)
116    }
117
118    /// Executes the given closure once the lock on the transaction is acquired. If the transaction
119    /// is timed out, it will be renewed first.
120    ///
121    /// Returns the result of the closure or an error if the transaction renewal fails.
122    #[inline]
123    pub(crate) fn txn_execute_renew_on_timeout<F, T>(&self, f: F) -> Result<T>
124    where
125        F: FnOnce(*mut ffi::MDBX_txn) -> T,
126    {
127        self.inner.txn_execute_renew_on_timeout(f)
128    }
129
130    /// Returns a copy of the raw pointer to the underlying MDBX transaction.
131    #[doc(hidden)]
132    #[cfg(test)]
133    pub fn txn(&self) -> *mut ffi::MDBX_txn {
134        self.inner.txn.txn
135    }
136
137    /// Returns a raw pointer to the MDBX environment.
138    pub fn env(&self) -> &Environment {
139        &self.inner.env
140    }
141
142    /// Returns the transaction id.
143    pub fn id(&self) -> Result<u64> {
144        self.txn_execute(|txn| unsafe { ffi::mdbx_txn_id(txn) })
145    }
146
147    /// Gets an item from a database.
148    ///
149    /// This function retrieves the data associated with the given key in the
150    /// database. If the database supports duplicate keys
151    /// ([`DatabaseFlags::DUP_SORT`]) then the first data item for the key will be
152    /// returned. Retrieval of other items requires the use of
153    /// [Cursor]. If the item is not in the database, then
154    /// [None] will be returned.
155    pub fn get<Key>(&self, dbi: ffi::MDBX_dbi, key: &[u8]) -> Result<Option<Key>>
156    where
157        Key: TableObject,
158    {
159        let key_val: ffi::MDBX_val =
160            ffi::MDBX_val { iov_len: key.len(), iov_base: key.as_ptr() as *mut c_void };
161        let mut data_val: ffi::MDBX_val = ffi::MDBX_val { iov_len: 0, iov_base: ptr::null_mut() };
162
163        self.txn_execute(|txn| unsafe {
164            match ffi::mdbx_get(txn, dbi, &key_val, &mut data_val) {
165                ffi::MDBX_SUCCESS => Key::decode_val::<K>(txn, data_val).map(Some),
166                ffi::MDBX_NOTFOUND => Ok(None),
167                err_code => Err(Error::from_err_code(err_code)),
168            }
169        })?
170    }
171
172    /// Commits the transaction.
173    ///
174    /// Any pending operations will be saved.
175    pub fn commit(self) -> Result<(bool, CommitLatency)> {
176        self.commit_and_rebind_open_dbs().map(|v| (v.0, v.1))
177    }
178
179    pub fn prime_for_permaopen(&self, db: Database) {
180        self.inner.primed_dbis.lock().insert(db.dbi());
181    }
182
183    /// Commits the transaction and returns table handles permanently open until dropped.
184    pub fn commit_and_rebind_open_dbs(self) -> Result<(bool, CommitLatency, Vec<Database>)> {
185        let result = {
186            let result = self.txn_execute(|txn| {
187                if K::IS_READ_ONLY {
188                    #[cfg(feature = "read-tx-timeouts")]
189                    self.env().txn_manager().remove_active_read_transaction(txn);
190
191                    let mut latency = CommitLatency::new();
192                    mdbx_result(unsafe {
193                        ffi::mdbx_txn_commit_ex(txn, latency.mdb_commit_latency())
194                    })
195                    .map(|v| (v, latency))
196                } else {
197                    let (sender, rx) = sync_channel(0);
198                    self.env()
199                        .txn_manager()
200                        .send_message(TxnManagerMessage::Commit { tx: TxnPtr(txn), sender });
201                    rx.recv().unwrap()
202                }
203            })?;
204
205            self.inner.set_committed();
206            result
207        };
208        result.map(|(v, latency)| {
209            (
210                v,
211                latency,
212                self.inner
213                    .primed_dbis
214                    .lock()
215                    .iter()
216                    .map(|&dbi| Database::new_from_ptr(dbi, self.env().clone()))
217                    .collect(),
218            )
219        })
220    }
221
222    /// Opens a handle to an MDBX database.
223    ///
224    /// If `name` is [None], then the returned handle will be for the default database.
225    ///
226    /// If `name` is not [None], then the returned handle will be for a named database. In this
227    /// case the environment must be configured to allow named databases through
228    /// [`EnvironmentBuilder::set_max_dbs()`](crate::EnvironmentBuilder::set_max_dbs).
229    ///
230    /// The returned database handle may be shared among any transaction in the environment.
231    ///
232    /// The database name may not contain the null character.
233    pub fn open_db(&self, name: Option<&str>) -> Result<Database> {
234        Database::new(self, name, 0)
235    }
236
237    /// Gets the option flags for the given database in the transaction.
238    pub fn db_flags(&self, db: &Database) -> Result<DatabaseFlags> {
239        let mut flags: c_uint = 0;
240        unsafe {
241            self.txn_execute(|txn| {
242                mdbx_result(ffi::mdbx_dbi_flags_ex(txn, db.dbi(), &mut flags, ptr::null_mut()))
243            })??;
244        }
245
246        // The types are not the same on Windows. Great!
247        #[cfg_attr(not(windows), allow(clippy::useless_conversion))]
248        Ok(DatabaseFlags::from_bits_truncate(flags.try_into().unwrap()))
249    }
250
251    /// Retrieves database statistics.
252    pub fn db_stat(&self, db: &Database) -> Result<Stat> {
253        self.db_stat_with_dbi(db.dbi())
254    }
255
256    /// Retrieves database statistics by the given dbi.
257    pub fn db_stat_with_dbi(&self, dbi: ffi::MDBX_dbi) -> Result<Stat> {
258        unsafe {
259            let mut stat = Stat::new();
260            self.txn_execute(|txn| {
261                mdbx_result(ffi::mdbx_dbi_stat(txn, dbi, stat.mdb_stat(), size_of::<Stat>()))
262            })??;
263            Ok(stat)
264        }
265    }
266
267    /// Open a new cursor on the given database.
268    pub fn cursor(&self, db: &Database) -> Result<Cursor<K>> {
269        Cursor::new(self.clone(), db.dbi())
270    }
271
272    /// Open a new cursor on the given dbi.
273    pub fn cursor_with_dbi(&self, dbi: ffi::MDBX_dbi) -> Result<Cursor<K>> {
274        Cursor::new(self.clone(), dbi)
275    }
276
277    /// Disables a timeout for this read transaction.
278    #[cfg(feature = "read-tx-timeouts")]
279    pub fn disable_timeout(&self) {
280        if K::IS_READ_ONLY {
281            self.env().txn_manager().remove_active_read_transaction(self.inner.txn.txn);
282        }
283    }
284}
285
286impl<K> Clone for Transaction<K>
287where
288    K: TransactionKind,
289{
290    fn clone(&self) -> Self {
291        Self { inner: Arc::clone(&self.inner) }
292    }
293}
294
295impl<K> fmt::Debug for Transaction<K>
296where
297    K: TransactionKind,
298{
299    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
300        f.debug_struct("RoTransaction").finish_non_exhaustive()
301    }
302}
303
304/// Internals of a transaction.
305struct TransactionInner<K>
306where
307    K: TransactionKind,
308{
309    /// The transaction pointer itself.
310    txn: TransactionPtr,
311    /// A set of database handles that are primed for permaopen.
312    primed_dbis: Mutex<IndexSet<ffi::MDBX_dbi>>,
313    /// Whether the transaction has committed.
314    committed: AtomicBool,
315    env: Environment,
316    _marker: std::marker::PhantomData<fn(K)>,
317}
318
319impl<K> TransactionInner<K>
320where
321    K: TransactionKind,
322{
323    /// Marks the transaction as committed.
324    fn set_committed(&self) {
325        self.committed.store(true, std::sync::atomic::Ordering::SeqCst);
326    }
327
328    fn has_committed(&self) -> bool {
329        self.committed.load(std::sync::atomic::Ordering::SeqCst)
330    }
331
332    #[inline]
333    fn txn_execute<F, T>(&self, f: F) -> Result<T>
334    where
335        F: FnOnce(*mut ffi::MDBX_txn) -> T,
336    {
337        self.txn.txn_execute_fail_on_timeout(f)
338    }
339
340    #[inline]
341    fn txn_execute_renew_on_timeout<F, T>(&self, f: F) -> Result<T>
342    where
343        F: FnOnce(*mut ffi::MDBX_txn) -> T,
344    {
345        self.txn.txn_execute_renew_on_timeout(f)
346    }
347}
348
349impl<K> Drop for TransactionInner<K>
350where
351    K: TransactionKind,
352{
353    fn drop(&mut self) {
354        // To be able to abort a timed out transaction, we need to renew it first.
355        // Hence the usage of `txn_execute_renew_on_timeout` here.
356        self.txn
357            .txn_execute_renew_on_timeout(|txn| {
358                if !self.has_committed() {
359                    if K::IS_READ_ONLY {
360                        #[cfg(feature = "read-tx-timeouts")]
361                        self.env.txn_manager().remove_active_read_transaction(txn);
362
363                        unsafe {
364                            ffi::mdbx_txn_abort(txn);
365                        }
366                    } else {
367                        let (sender, rx) = sync_channel(0);
368                        self.env
369                            .txn_manager()
370                            .send_message(TxnManagerMessage::Abort { tx: TxnPtr(txn), sender });
371                        rx.recv().unwrap().unwrap();
372                    }
373                }
374            })
375            .unwrap();
376    }
377}
378
379impl Transaction<RW> {
380    fn open_db_with_flags(&self, name: Option<&str>, flags: DatabaseFlags) -> Result<Database> {
381        Database::new(self, name, flags.bits())
382    }
383
384    /// Opens a handle to an MDBX database, creating the database if necessary.
385    ///
386    /// If the database is already created, the given option flags will be added to it.
387    ///
388    /// If `name` is [None], then the returned handle will be for the default database.
389    ///
390    /// If `name` is not [None], then the returned handle will be for a named database. In this
391    /// case the environment must be configured to allow named databases through
392    /// [`EnvironmentBuilder::set_max_dbs()`](crate::EnvironmentBuilder::set_max_dbs).
393    ///
394    /// This function will fail with [`Error::BadRslot`] if called by a thread with an open
395    /// transaction.
396    pub fn create_db(&self, name: Option<&str>, flags: DatabaseFlags) -> Result<Database> {
397        self.open_db_with_flags(name, flags | DatabaseFlags::CREATE)
398    }
399
400    /// Stores an item into a database.
401    ///
402    /// This function stores key/data pairs in the database. The default
403    /// behavior is to enter the new key/data pair, replacing any previously
404    /// existing key if duplicates are disallowed, or adding a duplicate data
405    /// item if duplicates are allowed ([`DatabaseFlags::DUP_SORT`]).
406    pub fn put(
407        &self,
408        dbi: ffi::MDBX_dbi,
409        key: impl AsRef<[u8]>,
410        data: impl AsRef<[u8]>,
411        flags: WriteFlags,
412    ) -> Result<()> {
413        let key = key.as_ref();
414        let data = data.as_ref();
415        let key_val: ffi::MDBX_val =
416            ffi::MDBX_val { iov_len: key.len(), iov_base: key.as_ptr() as *mut c_void };
417        let mut data_val: ffi::MDBX_val =
418            ffi::MDBX_val { iov_len: data.len(), iov_base: data.as_ptr() as *mut c_void };
419        mdbx_result(self.txn_execute(|txn| unsafe {
420            ffi::mdbx_put(txn, dbi, &key_val, &mut data_val, flags.bits())
421        })?)?;
422
423        Ok(())
424    }
425
426    /// Returns a buffer which can be used to write a value into the item at the
427    /// given key and with the given length. The buffer must be completely
428    /// filled by the caller.
429    #[allow(clippy::mut_from_ref)]
430    pub fn reserve(
431        &self,
432        db: &Database,
433        key: impl AsRef<[u8]>,
434        len: usize,
435        flags: WriteFlags,
436    ) -> Result<&mut [u8]> {
437        let key = key.as_ref();
438        let key_val: ffi::MDBX_val =
439            ffi::MDBX_val { iov_len: key.len(), iov_base: key.as_ptr() as *mut c_void };
440        let mut data_val: ffi::MDBX_val =
441            ffi::MDBX_val { iov_len: len, iov_base: ptr::null_mut::<c_void>() };
442        unsafe {
443            mdbx_result(self.txn_execute(|txn| {
444                ffi::mdbx_put(
445                    txn,
446                    db.dbi(),
447                    &key_val,
448                    &mut data_val,
449                    flags.bits() | ffi::MDBX_RESERVE,
450                )
451            })?)?;
452            Ok(slice::from_raw_parts_mut(data_val.iov_base as *mut u8, data_val.iov_len))
453        }
454    }
455
456    /// Delete items from a database.
457    /// This function removes key/data pairs from the database.
458    ///
459    /// The data parameter is NOT ignored regardless the database does support sorted duplicate data
460    /// items or not. If the data parameter is [Some] only the matching data item will be
461    /// deleted. Otherwise, if data parameter is [None], any/all value(s) for specified key will
462    /// be deleted.
463    ///
464    /// Returns `true` if the key/value pair was present.
465    pub fn del(
466        &self,
467        dbi: ffi::MDBX_dbi,
468        key: impl AsRef<[u8]>,
469        data: Option<&[u8]>,
470    ) -> Result<bool> {
471        let key = key.as_ref();
472        let key_val: ffi::MDBX_val =
473            ffi::MDBX_val { iov_len: key.len(), iov_base: key.as_ptr() as *mut c_void };
474        let data_val: Option<ffi::MDBX_val> = data.map(|data| ffi::MDBX_val {
475            iov_len: data.len(),
476            iov_base: data.as_ptr() as *mut c_void,
477        });
478
479        mdbx_result({
480            self.txn_execute(|txn| {
481                if let Some(d) = data_val {
482                    unsafe { ffi::mdbx_del(txn, dbi, &key_val, &d) }
483                } else {
484                    unsafe { ffi::mdbx_del(txn, dbi, &key_val, ptr::null()) }
485                }
486            })?
487        })
488        .map(|_| true)
489        .or_else(|e| match e {
490            Error::NotFound => Ok(false),
491            other => Err(other),
492        })
493    }
494
495    /// Empties the given database. All items will be removed.
496    pub fn clear_db(&self, dbi: ffi::MDBX_dbi) -> Result<()> {
497        mdbx_result(self.txn_execute(|txn| unsafe { ffi::mdbx_drop(txn, dbi, false) })?)?;
498
499        Ok(())
500    }
501
502    /// Drops the database from the environment.
503    ///
504    /// # Safety
505    /// Caller must close ALL other [Database] and [Cursor] instances pointing to the same dbi
506    /// BEFORE calling this function.
507    pub unsafe fn drop_db(&self, db: Database) -> Result<()> {
508        mdbx_result(self.txn_execute(|txn| ffi::mdbx_drop(txn, db.dbi(), true))?)?;
509
510        Ok(())
511    }
512}
513
514impl Transaction<RO> {
515    /// Closes the database handle.
516    ///
517    /// # Safety
518    /// Caller must close ALL other [Database] and [Cursor] instances pointing to the same dbi
519    /// BEFORE calling this function.
520    pub unsafe fn close_db(&self, db: Database) -> Result<()> {
521        mdbx_result(ffi::mdbx_dbi_close(self.env().env_ptr(), db.dbi()))?;
522
523        Ok(())
524    }
525}
526
527impl Transaction<RW> {
528    /// Begins a new nested transaction inside of this transaction.
529    pub fn begin_nested_txn(&mut self) -> Result<Self> {
530        if self.inner.env.is_write_map() {
531            return Err(Error::NestedTransactionsUnsupportedWithWriteMap)
532        }
533        self.txn_execute(|txn| {
534            let (tx, rx) = sync_channel(0);
535            self.env().txn_manager().send_message(TxnManagerMessage::Begin {
536                parent: TxnPtr(txn),
537                flags: RW::OPEN_FLAGS,
538                sender: tx,
539            });
540
541            rx.recv().unwrap().map(|ptr| Self::new_from_ptr(self.env().clone(), ptr.0))
542        })?
543    }
544}
545
546/// A shareable pointer to an MDBX transaction.
547#[derive(Debug, Clone)]
548pub(crate) struct TransactionPtr {
549    txn: *mut ffi::MDBX_txn,
550    #[cfg(feature = "read-tx-timeouts")]
551    timed_out: Arc<AtomicBool>,
552    lock: Arc<Mutex<()>>,
553}
554
555impl TransactionPtr {
556    fn new(txn: *mut ffi::MDBX_txn) -> Self {
557        Self {
558            txn,
559            #[cfg(feature = "read-tx-timeouts")]
560            timed_out: Arc::new(AtomicBool::new(false)),
561            lock: Arc::new(Mutex::new(())),
562        }
563    }
564
565    /// Returns `true` if the transaction is timed out.
566    ///
567    /// When transaction is timed out via `TxnManager`, it's actually reset using
568    /// `mdbx_txn_reset`. It makes the transaction unusable (MDBX fails on any usages of such
569    /// transactions).
570    ///
571    /// Importantly, we can't rely on `MDBX_TXN_FINISHED` flag to check if the transaction is timed
572    /// out using `mdbx_txn_reset`, because MDBX uses it in other cases too.
573    #[cfg(feature = "read-tx-timeouts")]
574    fn is_timed_out(&self) -> bool {
575        self.timed_out.load(std::sync::atomic::Ordering::SeqCst)
576    }
577
578    #[cfg(feature = "read-tx-timeouts")]
579    pub(crate) fn set_timed_out(&self) {
580        self.timed_out.store(true, std::sync::atomic::Ordering::SeqCst);
581    }
582
583    /// Acquires the inner transaction lock to guarantee exclusive access to the transaction
584    /// pointer.
585    fn lock(&self) -> MutexGuard<'_, ()> {
586        if let Some(lock) = self.lock.try_lock() {
587            lock
588        } else {
589            tracing::trace!(
590                target: "libmdbx",
591                txn = %self.txn as usize,
592                backtrace = %std::backtrace::Backtrace::capture(),
593                "Transaction lock is already acquired, blocking...
594                To display the full backtrace, run with `RUST_BACKTRACE=full` env variable."
595            );
596            self.lock.lock()
597        }
598    }
599
600    /// Executes the given closure once the lock on the transaction is acquired.
601    ///
602    /// Returns the result of the closure or an error if the transaction is timed out.
603    #[inline]
604    pub(crate) fn txn_execute_fail_on_timeout<F, T>(&self, f: F) -> Result<T>
605    where
606        F: FnOnce(*mut ffi::MDBX_txn) -> T,
607    {
608        let _lck = self.lock();
609
610        // No race condition with the `TxnManager` timing out the transaction is possible here,
611        // because we're taking a lock for any actions on the transaction pointer, including a call
612        // to the `mdbx_txn_reset`.
613        #[cfg(feature = "read-tx-timeouts")]
614        if self.is_timed_out() {
615            return Err(Error::ReadTransactionTimeout)
616        }
617
618        Ok((f)(self.txn))
619    }
620
621    /// Executes the given closure once the lock on the transaction is acquired. If the transaction
622    /// is timed out, it will be renewed first.
623    ///
624    /// Returns the result of the closure or an error if the transaction renewal fails.
625    #[inline]
626    pub(crate) fn txn_execute_renew_on_timeout<F, T>(&self, f: F) -> Result<T>
627    where
628        F: FnOnce(*mut ffi::MDBX_txn) -> T,
629    {
630        let _lck = self.lock();
631
632        // To be able to do any operations on the transaction, we need to renew it first.
633        #[cfg(feature = "read-tx-timeouts")]
634        if self.is_timed_out() {
635            mdbx_result(unsafe { mdbx_txn_renew(self.txn) })?;
636        }
637
638        Ok((f)(self.txn))
639    }
640}
641
642/// Commit latencies info.
643///
644/// Contains information about latency of commit stages.
645/// Inner struct stores this info in 1/65536 of seconds units.
646#[derive(Debug)]
647#[repr(transparent)]
648pub struct CommitLatency(ffi::MDBX_commit_latency);
649
650impl CommitLatency {
651    /// Create a new `CommitLatency` with zero'd inner struct `ffi::MDBX_commit_latency`.
652    pub(crate) const fn new() -> Self {
653        unsafe { Self(std::mem::zeroed()) }
654    }
655
656    /// Returns a mut pointer to `ffi::MDBX_commit_latency`.
657    pub(crate) const fn mdb_commit_latency(&mut self) -> *mut ffi::MDBX_commit_latency {
658        &mut self.0
659    }
660}
661
662impl CommitLatency {
663    /// Duration of preparation (commit child transactions, update
664    /// sub-databases records and cursors destroying).
665    #[inline]
666    pub const fn preparation(&self) -> Duration {
667        Self::time_to_duration(self.0.preparation)
668    }
669
670    /// Duration of GC update by wall clock.
671    #[inline]
672    pub const fn gc_wallclock(&self) -> Duration {
673        Self::time_to_duration(self.0.gc_wallclock)
674    }
675
676    /// Duration of internal audit if enabled.
677    #[inline]
678    pub const fn audit(&self) -> Duration {
679        Self::time_to_duration(self.0.audit)
680    }
681
682    /// Duration of writing dirty/modified data pages to a filesystem,
683    /// i.e. the summary duration of a `write()` syscalls during commit.
684    #[inline]
685    pub const fn write(&self) -> Duration {
686        Self::time_to_duration(self.0.write)
687    }
688
689    /// Duration of syncing written data to the disk/storage, i.e.
690    /// the duration of a `fdatasync()` or a `msync()` syscall during commit.
691    #[inline]
692    pub const fn sync(&self) -> Duration {
693        Self::time_to_duration(self.0.sync)
694    }
695
696    /// Duration of transaction ending (releasing resources).
697    #[inline]
698    pub const fn ending(&self) -> Duration {
699        Self::time_to_duration(self.0.ending)
700    }
701
702    /// The total duration of a commit.
703    #[inline]
704    pub const fn whole(&self) -> Duration {
705        Self::time_to_duration(self.0.whole)
706    }
707
708    /// User-mode CPU time spent on GC update.
709    #[inline]
710    pub const fn gc_cputime(&self) -> Duration {
711        Self::time_to_duration(self.0.gc_cputime)
712    }
713
714    #[inline]
715    const fn time_to_duration(time: u32) -> Duration {
716        Duration::from_nanos(time as u64 * (1_000_000_000 / 65_536))
717    }
718}
719
720// SAFETY: Access to the transaction is synchronized by the lock.
721unsafe impl Send for TransactionPtr {}
722
723// SAFETY: Access to the transaction is synchronized by the lock.
724unsafe impl Sync for TransactionPtr {}
725
726#[cfg(test)]
727mod tests {
728    use super::*;
729
730    const fn assert_send_sync<T: Send + Sync>() {}
731
732    #[expect(dead_code)]
733    const fn test_txn_send_sync() {
734        assert_send_sync::<Transaction<RO>>();
735        assert_send_sync::<Transaction<RW>>();
736    }
737}
reth_libmdbx/transaction.rs

reth_libmdbx/
transaction.rs
