reth_libmdbx/
txn_manager.rs

1use crate::{
2    environment::EnvPtr,
3    error::{mdbx_result, Result},
4    CommitLatency,
5};
6use std::{
7    fmt, ptr,
8    sync::mpsc::{sync_channel, Receiver, SyncSender},
9};
10
11#[derive(Copy, Clone, Debug)]
12pub(crate) struct TxnPtr(pub(crate) *mut ffi::MDBX_txn);
13unsafe impl Send for TxnPtr {}
14unsafe impl Sync for TxnPtr {}
15
16pub(crate) enum TxnManagerMessage {
17    Begin { parent: TxnPtr, flags: ffi::MDBX_txn_flags_t, sender: SyncSender<Result<TxnPtr>> },
18    Abort { tx: TxnPtr, sender: SyncSender<Result<bool>> },
19    Commit { tx: TxnPtr, sender: SyncSender<Result<(bool, CommitLatency)>> },
20}
21
22impl fmt::Debug for TxnManagerMessage {
23    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
24        match self {
25            Self::Begin { parent, flags, sender: _ } => {
26                f.debug_struct("Begin").field("parent", parent).field("flags", flags).finish()
27            }
28            Self::Abort { tx, sender: _ } => f.debug_struct("Abort").field("tx", tx).finish(),
29            Self::Commit { tx, sender: _ } => f.debug_struct("Commit").field("tx", tx).finish(),
30        }
31    }
32}
33
34/// Manages transactions by doing two things:
35/// - Opening, aborting, and committing transactions using [`TxnManager::send_message`] with the
36///   corresponding [`TxnManagerMessage`]
37/// - Aborting long-lived read transactions (if the `read-tx-timeouts` feature is enabled and
38///   `TxnManager::with_max_read_transaction_duration` is called)
39#[derive(Debug)]
40pub(crate) struct TxnManager {
41    sender: SyncSender<TxnManagerMessage>,
42    #[cfg(feature = "read-tx-timeouts")]
43    read_transactions: Option<read_transactions::ReadTransactionMonitor>,
44}
45
46impl TxnManager {
47    pub(crate) fn new(env: EnvPtr) -> Self {
48        let (tx, rx) = sync_channel(0);
49        let txn_manager = Self {
50            sender: tx,
51            #[cfg(feature = "read-tx-timeouts")]
52            read_transactions: None,
53        };
54
55        txn_manager.start_message_listener(env, rx);
56
57        txn_manager
58    }
59
60    /// Spawns a new [`std::thread`] that listens to incoming [`TxnManagerMessage`] messages,
61    /// executes an FFI function, and returns the result on the provided channel.
62    ///
63    /// - [`TxnManagerMessage::Begin`] opens a new transaction with [`ffi::mdbx_txn_begin_ex`]
64    /// - [`TxnManagerMessage::Abort`] aborts a transaction with [`ffi::mdbx_txn_abort`]
65    /// - [`TxnManagerMessage::Commit`] commits a transaction with [`ffi::mdbx_txn_commit_ex`]
66    fn start_message_listener(&self, env: EnvPtr, rx: Receiver<TxnManagerMessage>) {
67        let task = move || {
68            let env = env;
69            loop {
70                let msg = rx.recv();
71                tracing::debug!(target: "libmdbx::txn", ?msg, "txn-mngr received");
72                match msg {
73                    Ok(msg) => match msg {
74                        TxnManagerMessage::Begin { parent, flags, sender } => {
75                            let _span =
76                                tracing::debug_span!(target: "libmdbx::txn", "begin", flags)
77                                    .entered();
78                            let mut txn: *mut ffi::MDBX_txn = ptr::null_mut();
79                            let res = mdbx_result(unsafe {
80                                ffi::mdbx_txn_begin_ex(
81                                    env.0,
82                                    parent.0,
83                                    flags,
84                                    &mut txn,
85                                    ptr::null_mut(),
86                                )
87                            })
88                            .map(|_| TxnPtr(txn));
89                            sender.send(res).unwrap();
90                        }
91                        TxnManagerMessage::Abort { tx, sender } => {
92                            let _span =
93                                tracing::debug_span!(target: "libmdbx::txn", "abort").entered();
94                            sender.send(mdbx_result(unsafe { ffi::mdbx_txn_abort(tx.0) })).unwrap();
95                        }
96                        TxnManagerMessage::Commit { tx, sender } => {
97                            let _span =
98                                tracing::debug_span!(target: "libmdbx::txn", "commit").entered();
99                            sender
100                                .send({
101                                    let mut latency = CommitLatency::new();
102                                    mdbx_result(unsafe {
103                                        ffi::mdbx_txn_commit_ex(tx.0, latency.mdbx_commit_latency())
104                                    })
105                                    .map(|v| (v, latency))
106                                })
107                                .unwrap();
108                        }
109                    },
110                    Err(_) => return,
111                }
112            }
113        };
114        std::thread::Builder::new().name("mdbx-rs-txn-mgr".to_string()).spawn(task).unwrap();
115    }
116
117    pub(crate) fn send_message(&self, message: TxnManagerMessage) {
118        self.sender.send(message).unwrap()
119    }
120}
121
122#[cfg(feature = "read-tx-timeouts")]
123mod read_transactions {
124    use crate::{
125        environment::EnvPtr, error::mdbx_result, transaction::TransactionPtr,
126        txn_manager::TxnManager,
127    };
128    use dashmap::{DashMap, DashSet};
129    use std::{
130        backtrace::Backtrace,
131        ops::Deref,
132        sync::{
133            mpsc::{channel, sync_channel, RecvTimeoutError, Sender},
134            Arc,
135        },
136        thread::JoinHandle,
137        time::{Duration, Instant},
138    };
139    use tracing::{error, trace, warn};
140
141    const READ_TRANSACTIONS_CHECK_INTERVAL: Duration = Duration::from_secs(5);
142
143    impl TxnManager {
144        /// Returns a new instance for which the maximum duration that a read transaction can be
145        /// open is set.
146        pub(crate) fn new_with_max_read_transaction_duration(
147            env: EnvPtr,
148            duration: Duration,
149        ) -> Self {
150            let read_transactions = Arc::new(ReadTransactions::new(duration));
151            let read_transactions = read_transactions.start_monitor();
152
153            let (tx, rx) = sync_channel(0);
154
155            let txn_manager = Self { sender: tx, read_transactions: Some(read_transactions) };
156
157            txn_manager.start_message_listener(env, rx);
158
159            txn_manager
160        }
161
162        /// Stops and joins the read transaction monitor before the environment is closed.
163        pub(crate) fn stop_read_transaction_monitor(&mut self) {
164            drop(self.read_transactions.take());
165        }
166
167        /// Adds a new transaction to the list of active read transactions.
168        pub(crate) fn add_active_read_transaction(
169            &self,
170            ptr: *mut ffi::MDBX_txn,
171            tx: TransactionPtr,
172        ) {
173            if let Some(read_transactions) = &self.read_transactions {
174                read_transactions.add_active(ptr, tx);
175            }
176        }
177
178        /// Removes a transaction from the list of active read transactions.
179        ///
180        /// Returns `true` if the transaction was found and removed.
181        pub(crate) fn remove_active_read_transaction(&self, ptr: *mut ffi::MDBX_txn) -> bool {
182            self.read_transactions.as_ref().is_some_and(|txs| txs.remove_active(ptr))
183        }
184
185        /// Returns the number of timed out transactions that were not aborted by the user yet.
186        pub(crate) fn timed_out_not_aborted_read_transactions(&self) -> Option<usize> {
187            self.read_transactions
188                .as_ref()
189                .map(|read_transactions| read_transactions.timed_out_not_aborted())
190        }
191    }
192
193    /// Owns the monitor thread and joins it when the environment is closed.
194    #[derive(Debug)]
195    pub(super) struct ReadTransactionMonitor {
196        transactions: Arc<ReadTransactions>,
197        shutdown: Option<Sender<()>>,
198        handle: Option<JoinHandle<()>>,
199    }
200
201    impl Deref for ReadTransactionMonitor {
202        type Target = ReadTransactions;
203
204        fn deref(&self) -> &Self::Target {
205            &self.transactions
206        }
207    }
208
209    impl Drop for ReadTransactionMonitor {
210        fn drop(&mut self) {
211            // Disconnecting wakes the monitor even if it has not started waiting yet.
212            drop(self.shutdown.take());
213            if let Some(handle) = self.handle.take() &&
214                handle.join().is_err()
215            {
216                error!(target: "libmdbx", "Read transaction monitor panicked during shutdown");
217            }
218        }
219    }
220
221    #[derive(Debug, Default)]
222    pub(super) struct ReadTransactions {
223        /// Maximum duration that a read transaction can be open until the
224        /// [`ReadTransactions::start_monitor`] aborts it.
225        max_duration: Duration,
226        /// List of currently active read transactions.
227        ///
228        /// We store `usize` instead of a raw pointer as a key, because pointers are not
229        /// comparable. The time of transaction opening is stored as a value.
230        ///
231        /// The backtrace of the transaction opening is recorded only when debug assertions are
232        /// enabled outside DST.
233        active: DashMap<usize, (TransactionPtr, Instant, Option<Arc<Backtrace>>)>,
234        /// List of timed out transactions that were not aborted by the user yet, hence have a
235        /// dangling read transaction pointer.
236        timed_out_not_aborted: DashSet<usize>,
237    }
238
239    impl ReadTransactions {
240        pub(super) fn new(max_duration: Duration) -> Self {
241            Self { max_duration, ..Default::default() }
242        }
243
244        /// Adds a new transaction to the list of active read transactions.
245        pub(super) fn add_active(&self, ptr: *mut ffi::MDBX_txn, tx: TransactionPtr) {
246            let _ = self.active.insert(
247                ptr as usize,
248                (
249                    tx,
250                    Instant::now(),
251                    cfg!(all(debug_assertions, not(feature = "fuzz")))
252                        .then(|| Arc::new(Backtrace::force_capture())),
253                ),
254            );
255        }
256
257        /// Removes a transaction from the list of active read transactions.
258        pub(super) fn remove_active(&self, ptr: *mut ffi::MDBX_txn) -> bool {
259            self.timed_out_not_aborted.remove(&(ptr as usize));
260            self.active.remove(&(ptr as usize)).is_some()
261        }
262
263        /// Returns `true` if `tx` is still the active read transaction registered for `ptr`.
264        ///
265        /// Pointers can be reused by new read transactions, so the transaction identity is
266        /// compared as well.
267        fn is_active(&self, ptr: usize, tx: &TransactionPtr) -> bool {
268            self.active.get(&ptr).is_some_and(|entry| entry.value().0.is_same(tx))
269        }
270
271        /// Returns the number of timed out transactions that were not aborted by the user yet.
272        pub(super) fn timed_out_not_aborted(&self) -> usize {
273            self.timed_out_not_aborted.len()
274        }
275
276        /// Spawns a new [`std::thread`] that monitors the list of active read transactions and
277        /// timeouts those that are open for longer than `ReadTransactions.max_duration`.
278        pub(super) fn start_monitor(self: Arc<Self>) -> ReadTransactionMonitor {
279            let transactions = self.clone();
280            let (shutdown, receiver) = channel();
281            let task = move || {
282                let mut expired = Vec::new();
283
284                loop {
285                    let now = Instant::now();
286                    let mut max_active_transaction_duration = None;
287
288                    // Collect active read transactions that are open for longer than
289                    // `self.max_duration`.
290                    //
291                    // The transaction locks must not be taken while iterating: commit and drop
292                    // hold the transaction lock and then remove the entry from `active`, which
293                    // needs the write lock of the shard that the iterator is holding.
294                    for entry in &self.active {
295                        let (tx, start, backtrace) = entry.value();
296                        let duration = now - *start;
297
298                        if duration > self.max_duration {
299                            expired.push((*entry.key(), tx.clone(), duration, backtrace.clone()));
300                        } else {
301                            max_active_transaction_duration = Some(
302                                duration.max(max_active_transaction_duration.unwrap_or_default()),
303                            );
304                        }
305                    }
306
307                    // Time out the collected transactions.
308                    for (ptr, tx, open_duration, backtrace) in &expired {
309                        let ptr = *ptr;
310                        let result = tx.txn_execute_fail_on_timeout(|txn_ptr| {
311                            // The transaction could have been committed or dropped after it was
312                            // collected, and its pointer could even have been reused by a new
313                            // read transaction from the pool. We hold the transaction lock, so
314                            // it can't be finished from now on. If the entry is still ours, the
315                            // transaction is live.
316                            if !self.is_active(ptr, tx) {
317                                return None
318                            }
319
320                            // Time out the transaction.
321                            //
322                            // We use `mdbx_txn_reset` instead of `mdbx_txn_abort` here to
323                            // prevent MDBX from reusing the pointer of the aborted
324                            // transaction for new read-only transactions. This is
325                            // important because we store the pointer in the `active` list
326                            // and assume that it is unique.
327                            //
328                            // See https://libmdbx.dqdkfa.ru/group__c__transactions.html#gae9f34737fe60b0ba538d5a09b6a25c8d for more info.
329                            let result = mdbx_result(unsafe { ffi::mdbx_txn_reset(txn_ptr) });
330                            if result.is_ok() {
331                                tx.set_timed_out();
332                            }
333
334                            // Delete the transaction from the list of active transactions while
335                            // still holding the transaction lock, so that it can't be dropped
336                            // in between.
337                            let was_in_active = self.remove_active(txn_ptr);
338                            match result {
339                                Ok(_) => {
340                                    // Happy path, the transaction has been timed out by us with
341                                    // no errors. Add it to the list of timed out transactions
342                                    // that were not aborted by the user yet.
343                                    self.timed_out_not_aborted.insert(txn_ptr as usize);
344                                    warn!(target: "libmdbx", ?open_duration, ?backtrace, "Long-lived read transaction has been timed out");
345                                }
346                                Err(err) if was_in_active => {
347                                    // If the transaction was in the list of active transactions,
348                                    // then user didn't abort it and we failed to do so.
349                                    error!(target: "libmdbx", %err, ?open_duration, ?backtrace, "Failed to time out the long-lived read transaction");
350                                }
351                                Err(_) => {}
352                            }
353                            Some(())
354                        });
355
356                        if let Err(err) = result {
357                            error!(target: "libmdbx", %err, ?backtrace, "Failed to abort the long-lived read transaction")
358                        }
359                    }
360
361                    // Clear the list, but not de-allocate the reserved capacity to save on
362                    // further pushes.
363                    expired.clear();
364
365                    if !self.active.is_empty() {
366                        trace!(
367                            target: "libmdbx",
368                            elapsed = ?now.elapsed(),
369                            active = ?self.active.iter().map(|entry| {
370                                let (tx, start, _) = entry.value();
371                                (tx.clone(), start.elapsed())
372                            }).collect::<Vec<_>>(),
373                            "Read transactions"
374                        );
375                    }
376
377                    // Sleep not more than `READ_TRANSACTIONS_CHECK_INTERVAL`, but at least until
378                    // the closest deadline of an active read transaction
379                    let sleep_duration = READ_TRANSACTIONS_CHECK_INTERVAL.min(
380                        self.max_duration - max_active_transaction_duration.unwrap_or_default(),
381                    );
382                    trace!(target: "libmdbx", ?sleep_duration, elapsed = ?now.elapsed(), "Putting transaction monitor to sleep");
383                    match receiver.recv_timeout(sleep_duration) {
384                        Err(RecvTimeoutError::Timeout) => {}
385                        Ok(()) | Err(RecvTimeoutError::Disconnected) => break,
386                    }
387                }
388            };
389            let handle = std::thread::Builder::new()
390                .name("mdbx-rs-read-tx-timeouts".to_string())
391                .spawn(task)
392                .unwrap();
393
394            ReadTransactionMonitor { transactions, shutdown: Some(shutdown), handle: Some(handle) }
395        }
396    }
397
398    #[cfg(test)]
399    mod tests {
400        use super::*;
401        use crate::{Environment, Error, MaxReadTransactionDuration};
402        use std::{
403            sync::mpsc,
404            thread::{self, sleep},
405            time::{Duration, Instant},
406        };
407        use tempfile::tempdir;
408
409        #[test]
410        fn txn_manager_read_transactions_duration_set() {
411            const MAX_DURATION: Duration = Duration::from_secs(1);
412
413            let dir = tempdir().unwrap();
414            let env = Environment::builder()
415                .set_max_read_transaction_duration(MaxReadTransactionDuration::Set(MAX_DURATION))
416                .open(dir.path())
417                .unwrap();
418
419            let read_transactions = env.txn_manager().read_transactions.as_ref().unwrap();
420
421            // Create a read-only transaction, successfully use it, close it by dropping.
422            {
423                let tx = env.begin_ro_txn().unwrap();
424                let tx_ptr = tx.txn() as usize;
425                assert!(read_transactions.active.contains_key(&tx_ptr));
426                assert_eq!(
427                    read_transactions.active.get(&tx_ptr).unwrap().2.is_some(),
428                    cfg!(all(debug_assertions, not(feature = "fuzz"))),
429                );
430
431                let snapshot = read_transactions.active.get(&tx_ptr).unwrap().0.clone();
432                assert!(read_transactions.is_active(tx_ptr, &snapshot));
433                tx.open_db(None).unwrap();
434                drop(tx);
435
436                assert!(!read_transactions.is_active(tx_ptr, &snapshot));
437
438                // A collected entry must not match a later transaction reusing its pooled handle.
439                let tx = env.begin_ro_txn().unwrap();
440                assert_eq!(tx.txn() as usize, tx_ptr);
441                assert!(!read_transactions.is_active(tx_ptr, &snapshot));
442                tx.open_db(None).unwrap();
443            }
444
445            // Create a read-only transaction, successfully use it, close it by committing.
446            {
447                let tx = env.begin_ro_txn().unwrap();
448                let tx_ptr = tx.txn() as usize;
449                assert!(read_transactions.active.contains_key(&tx_ptr));
450
451                tx.open_db(None).unwrap();
452                tx.commit().unwrap();
453
454                assert!(!read_transactions.active.contains_key(&tx_ptr));
455            }
456
457            {
458                // Create a read-only transaction and observe it's in the list of active
459                // transactions.
460                let tx = env.begin_ro_txn().unwrap();
461                let tx_ptr = tx.txn() as usize;
462                assert!(read_transactions.active.contains_key(&tx_ptr));
463
464                // Wait until the transaction is timed out by the manager.
465                let deadline = Instant::now() + MAX_DURATION + READ_TRANSACTIONS_CHECK_INTERVAL;
466                while !read_transactions.timed_out_not_aborted.contains(&tx_ptr) {
467                    assert!(Instant::now() < deadline, "transaction was not timed out");
468                    sleep(Duration::from_millis(10));
469                }
470
471                // Ensure that the transaction is not in the list of active transactions anymore,
472                // and is in the list of timed out but not aborted transactions.
473                assert!(!read_transactions.active.contains_key(&tx_ptr));
474                assert!(read_transactions.timed_out_not_aborted.contains(&tx_ptr));
475
476                // Use the timed out transaction and observe the `Error::ReadTransactionTimeout`
477                assert_eq!(tx.open_db(None).err(), Some(Error::ReadTransactionTimeout));
478                assert!(!read_transactions.active.contains_key(&tx_ptr));
479                assert!(read_transactions.timed_out_not_aborted.contains(&tx_ptr));
480
481                assert_eq!(tx.id().err(), Some(Error::ReadTransactionTimeout));
482                assert!(!read_transactions.active.contains_key(&tx_ptr));
483                assert!(read_transactions.timed_out_not_aborted.contains(&tx_ptr));
484
485                // Ensure that the transaction pointer is not reused when opening a new read-only
486                // transaction.
487                let new_tx = env.begin_ro_txn().unwrap();
488                let new_tx_ptr = new_tx.txn() as usize;
489                assert!(read_transactions.active.contains_key(&new_tx_ptr));
490                assert_ne!(tx_ptr, new_tx_ptr);
491
492                // Drop the transaction and ensure that it's not in the list of timed out but not
493                // aborted transactions anymore.
494                drop(tx);
495                assert!(!read_transactions.timed_out_not_aborted.contains(&tx_ptr));
496            }
497        }
498
499        /// Commits and drops read transactions around their timeout while the monitor is timing
500        /// them out. The monitor must not hold a lock on the active transactions while waiting
501        /// for a transaction lock, otherwise it deadlocks with commit and drop.
502        #[test]
503        fn txn_manager_read_transactions_timeout_commit_drop_race() {
504            const MAX_DURATION: Duration = Duration::from_millis(10);
505            const THREADS: usize = 16;
506            const RUN_FOR: Duration = Duration::from_secs(2);
507
508            let dir = tempdir().unwrap();
509            let env = Environment::builder()
510                .set_max_read_transaction_duration(MaxReadTransactionDuration::Set(MAX_DURATION))
511                .open(dir.path())
512                .unwrap();
513
514            let (done_tx, done_rx) = mpsc::channel();
515            for thread_id in 0..THREADS {
516                let env = env.clone();
517                let done_tx = done_tx.clone();
518                thread::spawn(move || {
519                    let deadline = Instant::now() + RUN_FOR;
520                    let mut i = thread_id;
521                    while Instant::now() < deadline {
522                        let tx = env.begin_ro_txn().unwrap();
523                        // Finish the transaction right around the moment it expires.
524                        sleep(
525                            MAX_DURATION - Duration::from_millis(3) +
526                                Duration::from_millis(i as u64 % 6),
527                        );
528                        // The transaction might have been timed out already.
529                        if i % 2 == 0 {
530                            if let Err(err) = tx.commit() {
531                                assert_eq!(err, Error::ReadTransactionTimeout);
532                            }
533                        } else {
534                            drop(tx);
535                        }
536                        i += 1;
537                    }
538                    done_tx.send(()).unwrap();
539                });
540            }
541            drop(done_tx);
542
543            for _ in 0..THREADS {
544                done_rx
545                    .recv_timeout(Duration::from_secs(30))
546                    .expect("read transaction timeout monitor deadlocked");
547            }
548
549            // Every transaction was finished, so nothing must be left behind.
550            let read_transactions = env.txn_manager().read_transactions.as_ref().unwrap();
551            assert!(read_transactions.active.is_empty());
552            assert!(read_transactions.timed_out_not_aborted.is_empty());
553        }
554
555        #[test]
556        fn txn_manager_read_transactions_duration_unbounded() {
557            let dir = tempdir().unwrap();
558            let env = Environment::builder()
559                .set_max_read_transaction_duration(MaxReadTransactionDuration::Unbounded)
560                .open(dir.path())
561                .unwrap();
562
563            assert!(env.txn_manager().read_transactions.is_none());
564
565            let tx = env.begin_ro_txn().unwrap();
566            assert!(tx.commit().is_ok())
567        }
568
569        #[test]
570        fn read_transaction_monitor_stops_on_last_environment_drop() {
571            let dir = tempdir().unwrap();
572            let env = Environment::builder()
573                .set_max_read_transaction_duration(MaxReadTransactionDuration::Set(
574                    Duration::from_secs(60),
575                ))
576                .open(dir.path())
577                .unwrap();
578            let transactions =
579                Arc::downgrade(&env.txn_manager().read_transactions.as_ref().unwrap().transactions);
580            let clone = env.clone();
581            drop(env);
582            assert!(transactions.upgrade().is_some());
583
584            drop(clone);
585            assert!(transactions.upgrade().is_none(), "monitor retained transaction state");
586        }
587
588        #[test]
589        fn read_transaction_monitor_shutdown_wakes_waiter() {
590            let transactions = Arc::new(ReadTransactions::new(Duration::from_secs(60)));
591            let monitor = transactions.start_monitor();
592            // Retaining the state proves shutdown does not depend on its reference count.
593            let transactions = monitor.transactions.clone();
594            let started = Instant::now();
595            drop(monitor);
596            assert!(started.elapsed() < READ_TRANSACTIONS_CHECK_INTERVAL);
597            assert_eq!(Arc::strong_count(&transactions), 1);
598        }
599    }
600}
reth_libmdbx/txn_manager.rs

reth_libmdbx/
txn_manager.rs
