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<std::sync::Arc<read_transactions::ReadTransactions>>,
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        sync::{mpsc::sync_channel, Arc},
132        time::{Duration, Instant},
133    };
134    use tracing::{error, trace, warn};
135
136    const READ_TRANSACTIONS_CHECK_INTERVAL: Duration = Duration::from_secs(5);
137
138    impl TxnManager {
139        /// Returns a new instance for which the maximum duration that a read transaction can be
140        /// open is set.
141        pub(crate) fn new_with_max_read_transaction_duration(
142            env: EnvPtr,
143            duration: Duration,
144        ) -> Self {
145            let read_transactions = Arc::new(ReadTransactions::new(duration));
146            read_transactions.clone().start_monitor();
147
148            let (tx, rx) = sync_channel(0);
149
150            let txn_manager = Self { sender: tx, read_transactions: Some(read_transactions) };
151
152            txn_manager.start_message_listener(env, rx);
153
154            txn_manager
155        }
156
157        /// Adds a new transaction to the list of active read transactions.
158        pub(crate) fn add_active_read_transaction(
159            &self,
160            ptr: *mut ffi::MDBX_txn,
161            tx: TransactionPtr,
162        ) {
163            if let Some(read_transactions) = &self.read_transactions {
164                read_transactions.add_active(ptr, tx);
165            }
166        }
167
168        /// Removes a transaction from the list of active read transactions.
169        ///
170        /// Returns `true` if the transaction was found and removed.
171        pub(crate) fn remove_active_read_transaction(&self, ptr: *mut ffi::MDBX_txn) -> bool {
172            self.read_transactions.as_ref().is_some_and(|txs| txs.remove_active(ptr))
173        }
174
175        /// Returns the number of timed out transactions that were not aborted by the user yet.
176        pub(crate) fn timed_out_not_aborted_read_transactions(&self) -> Option<usize> {
177            self.read_transactions
178                .as_ref()
179                .map(|read_transactions| read_transactions.timed_out_not_aborted())
180        }
181    }
182
183    #[derive(Debug, Default)]
184    pub(super) struct ReadTransactions {
185        /// Maximum duration that a read transaction can be open until the
186        /// [`ReadTransactions::start_monitor`] aborts it.
187        max_duration: Duration,
188        /// List of currently active read transactions.
189        ///
190        /// We store `usize` instead of a raw pointer as a key, because pointers are not
191        /// comparable. The time of transaction opening is stored as a value.
192        ///
193        /// The backtrace of the transaction opening is recorded only when debug assertions are
194        /// enabled outside DST.
195        active: DashMap<usize, (TransactionPtr, Instant, Option<Arc<Backtrace>>)>,
196        /// List of timed out transactions that were not aborted by the user yet, hence have a
197        /// dangling read transaction pointer.
198        timed_out_not_aborted: DashSet<usize>,
199    }
200
201    impl ReadTransactions {
202        pub(super) fn new(max_duration: Duration) -> Self {
203            Self { max_duration, ..Default::default() }
204        }
205
206        /// Adds a new transaction to the list of active read transactions.
207        pub(super) fn add_active(&self, ptr: *mut ffi::MDBX_txn, tx: TransactionPtr) {
208            let _ = self.active.insert(
209                ptr as usize,
210                (
211                    tx,
212                    Instant::now(),
213                    cfg!(all(debug_assertions, not(feature = "dst")))
214                        .then(|| Arc::new(Backtrace::force_capture())),
215                ),
216            );
217        }
218
219        /// Removes a transaction from the list of active read transactions.
220        pub(super) fn remove_active(&self, ptr: *mut ffi::MDBX_txn) -> bool {
221            self.timed_out_not_aborted.remove(&(ptr as usize));
222            self.active.remove(&(ptr as usize)).is_some()
223        }
224
225        /// Returns the number of timed out transactions that were not aborted by the user yet.
226        pub(super) fn timed_out_not_aborted(&self) -> usize {
227            self.timed_out_not_aborted.len()
228        }
229
230        /// Spawns a new [`std::thread`] that monitors the list of active read transactions and
231        /// timeouts those that are open for longer than `ReadTransactions.max_duration`.
232        pub(super) fn start_monitor(self: Arc<Self>) {
233            let task = move || {
234                let mut timed_out_active = Vec::new();
235
236                loop {
237                    let now = Instant::now();
238                    let mut max_active_transaction_duration = None;
239
240                    // Iterate through active read transactions and time out those that's open for
241                    // longer than `self.max_duration`.
242                    for entry in &self.active {
243                        let (tx, start, backtrace) = entry.value();
244                        let duration = now - *start;
245
246                        if duration > self.max_duration {
247                            let result = tx.txn_execute_fail_on_timeout(|txn_ptr| {
248                                // Time out the transaction.
249                                //
250                                // We use `mdbx_txn_reset` instead of `mdbx_txn_abort` here to
251                                // prevent MDBX from reusing the pointer of the aborted
252                                // transaction for new read-only transactions. This is
253                                // important because we store the pointer in the `active` list
254                                // and assume that it is unique.
255                                //
256                                // See https://libmdbx.dqdkfa.ru/group__c__transactions.html#gae9f34737fe60b0ba538d5a09b6a25c8d for more info.
257                                let result = mdbx_result(unsafe { ffi::mdbx_txn_reset(txn_ptr) });
258                                if result.is_ok() {
259                                    tx.set_timed_out();
260                                }
261                                (txn_ptr, duration, result)
262                            });
263
264                            match result {
265                                Ok((txn_ptr, duration, error)) => {
266                                    // Add the transaction to `timed_out_active`. We can't remove it
267                                    // instantly from the list of active transactions, because we
268                                    // iterate through it.
269                                    timed_out_active.push((
270                                        txn_ptr,
271                                        duration,
272                                        backtrace.clone(),
273                                        error,
274                                    ));
275                                }
276                                Err(err) => {
277                                    error!(target: "libmdbx", %err, ?backtrace, "Failed to abort the long-lived read transaction")
278                                }
279                            }
280                        } else {
281                            max_active_transaction_duration = Some(
282                                duration.max(max_active_transaction_duration.unwrap_or_default()),
283                            );
284                        }
285                    }
286
287                    // Walk through timed out transactions, and delete them from the list of active
288                    // transactions.
289                    for (ptr, open_duration, backtrace, err) in timed_out_active.iter().cloned() {
290                        // Try deleting the transaction from the list of active transactions.
291                        let was_in_active = self.remove_active(ptr);
292                        if let Err(err) = err {
293                            if was_in_active {
294                                // If the transaction was in the list of active transactions,
295                                // then user didn't abort it and we failed to do so.
296                                error!(target: "libmdbx", %err, ?open_duration, ?backtrace, "Failed to time out the long-lived read transaction");
297                            }
298                        } else {
299                            // Happy path, the transaction has been timed out by us with no errors.
300                            warn!(target: "libmdbx", ?open_duration, ?backtrace, "Long-lived read transaction has been timed out");
301                            // Add transaction to the list of timed out transactions that were not
302                            // aborted by the user yet.
303                            self.timed_out_not_aborted.insert(ptr as usize);
304                        }
305                    }
306
307                    // Clear the list of timed out transactions, but not de-allocate the reserved
308                    // capacity to save on further pushes.
309                    timed_out_active.clear();
310
311                    if !self.active.is_empty() {
312                        trace!(
313                            target: "libmdbx",
314                            elapsed = ?now.elapsed(),
315                            active = ?self.active.iter().map(|entry| {
316                                let (tx, start, _) = entry.value();
317                                (tx.clone(), start.elapsed())
318                            }).collect::<Vec<_>>(),
319                            "Read transactions"
320                        );
321                    }
322
323                    // Sleep not more than `READ_TRANSACTIONS_CHECK_INTERVAL`, but at least until
324                    // the closest deadline of an active read transaction
325                    let sleep_duration = READ_TRANSACTIONS_CHECK_INTERVAL.min(
326                        self.max_duration - max_active_transaction_duration.unwrap_or_default(),
327                    );
328                    trace!(target: "libmdbx", ?sleep_duration, elapsed = ?now.elapsed(), "Putting transaction monitor to sleep");
329                    std::thread::sleep(sleep_duration);
330                }
331            };
332            std::thread::Builder::new()
333                .name("mdbx-rs-read-tx-timeouts".to_string())
334                .spawn(task)
335                .unwrap();
336        }
337    }
338
339    #[cfg(test)]
340    mod tests {
341        use crate::{
342            txn_manager::read_transactions::READ_TRANSACTIONS_CHECK_INTERVAL, Environment, Error,
343            MaxReadTransactionDuration,
344        };
345        use std::{
346            thread::sleep,
347            time::{Duration, Instant},
348        };
349        use tempfile::tempdir;
350
351        #[test]
352        fn txn_manager_read_transactions_duration_set() {
353            const MAX_DURATION: Duration = Duration::from_secs(1);
354
355            let dir = tempdir().unwrap();
356            let env = Environment::builder()
357                .set_max_read_transaction_duration(MaxReadTransactionDuration::Set(MAX_DURATION))
358                .open(dir.path())
359                .unwrap();
360
361            let read_transactions = env.txn_manager().read_transactions.as_ref().unwrap();
362
363            // Create a read-only transaction, successfully use it, close it by dropping.
364            {
365                let tx = env.begin_ro_txn().unwrap();
366                let tx_ptr = tx.txn() as usize;
367                assert!(read_transactions.active.contains_key(&tx_ptr));
368                assert_eq!(
369                    read_transactions.active.get(&tx_ptr).unwrap().2.is_some(),
370                    cfg!(all(debug_assertions, not(feature = "dst"))),
371                );
372
373                tx.open_db(None).unwrap();
374                drop(tx);
375
376                assert!(!read_transactions.active.contains_key(&tx_ptr));
377            }
378
379            // Create a read-only transaction, successfully use it, close it by committing.
380            {
381                let tx = env.begin_ro_txn().unwrap();
382                let tx_ptr = tx.txn() as usize;
383                assert!(read_transactions.active.contains_key(&tx_ptr));
384
385                tx.open_db(None).unwrap();
386                tx.commit().unwrap();
387
388                assert!(!read_transactions.active.contains_key(&tx_ptr));
389            }
390
391            {
392                // Create a read-only transaction and observe it's in the list of active
393                // transactions.
394                let tx = env.begin_ro_txn().unwrap();
395                let tx_ptr = tx.txn() as usize;
396                assert!(read_transactions.active.contains_key(&tx_ptr));
397
398                // Wait until the transaction is timed out by the manager.
399                let deadline = Instant::now() + MAX_DURATION + READ_TRANSACTIONS_CHECK_INTERVAL;
400                while !read_transactions.timed_out_not_aborted.contains(&tx_ptr) {
401                    assert!(Instant::now() < deadline, "transaction was not timed out");
402                    sleep(Duration::from_millis(10));
403                }
404
405                // Ensure that the transaction is not in the list of active transactions anymore,
406                // and is in the list of timed out but not aborted transactions.
407                assert!(!read_transactions.active.contains_key(&tx_ptr));
408                assert!(read_transactions.timed_out_not_aborted.contains(&tx_ptr));
409
410                // Use the timed out transaction and observe the `Error::ReadTransactionTimeout`
411                assert_eq!(tx.open_db(None).err(), Some(Error::ReadTransactionTimeout));
412                assert!(!read_transactions.active.contains_key(&tx_ptr));
413                assert!(read_transactions.timed_out_not_aborted.contains(&tx_ptr));
414
415                assert_eq!(tx.id().err(), Some(Error::ReadTransactionTimeout));
416                assert!(!read_transactions.active.contains_key(&tx_ptr));
417                assert!(read_transactions.timed_out_not_aborted.contains(&tx_ptr));
418
419                // Ensure that the transaction pointer is not reused when opening a new read-only
420                // transaction.
421                let new_tx = env.begin_ro_txn().unwrap();
422                let new_tx_ptr = new_tx.txn() as usize;
423                assert!(read_transactions.active.contains_key(&new_tx_ptr));
424                assert_ne!(tx_ptr, new_tx_ptr);
425
426                // Drop the transaction and ensure that it's not in the list of timed out but not
427                // aborted transactions anymore.
428                drop(tx);
429                assert!(!read_transactions.timed_out_not_aborted.contains(&tx_ptr));
430            }
431        }
432
433        #[test]
434        fn txn_manager_read_transactions_duration_unbounded() {
435            let dir = tempdir().unwrap();
436            let env = Environment::builder()
437                .set_max_read_transaction_duration(MaxReadTransactionDuration::Unbounded)
438                .open(dir.path())
439                .unwrap();
440
441            assert!(env.txn_manager().read_transactions.is_none());
442
443            let tx = env.begin_ro_txn().unwrap();
444            assert!(tx.commit().is_ok())
445        }
446    }
447}
reth_libmdbx/txn_manager.rs

reth_libmdbx/
txn_manager.rs
