rusqlite/
busy.rs

1///! Busy handler (when the database is locked)
2use std::convert::TryInto;
3use std::mem;
4use std::os::raw::{c_int, c_void};
5use std::panic::catch_unwind;
6use std::ptr;
7use std::time::Duration;
8
9use crate::ffi;
10use crate::{Connection, InnerConnection, Result};
11
12impl Connection {
13    /// Set a busy handler that sleeps for a specified amount of time when a
14    /// table is locked. The handler will sleep multiple times until at
15    /// least "ms" milliseconds of sleeping have accumulated.
16    ///
17    /// Calling this routine with an argument equal to zero turns off all busy
18    /// handlers.
19    ///
20    /// There can only be a single busy handler for a particular database
21    /// connection at any given moment. If another busy handler was defined
22    /// (using [`busy_handler`](Connection::busy_handler)) prior to calling this routine, that other
23    /// busy handler is cleared.
24    ///
25    /// Newly created connections currently have a default busy timeout of 5000ms, but this may be
26    /// subject to change.
27    pub fn busy_timeout(&self, timeout: Duration) -> Result<()> {
28        let ms: i32 = timeout
29            .as_secs()
30            .checked_mul(1000)
31            .and_then(|t| t.checked_add(timeout.subsec_millis().into()))
32            .and_then(|t| t.try_into().ok())
33            .expect("too big");
34        self.db.borrow_mut().busy_timeout(ms)
35    }
36
37    /// Register a callback to handle `SQLITE_BUSY` errors.
38    ///
39    /// If the busy callback is `None`, then `SQLITE_BUSY` is returned
40    /// immediately upon encountering the lock. The argument to the busy
41    /// handler callback is the number of times that the
42    /// busy handler has been invoked previously for the
43    /// same locking event. If the busy callback returns `false`, then no
44    /// additional attempts are made to access the
45    /// database and `SQLITE_BUSY` is returned to the
46    /// application. If the callback returns `true`, then another attempt
47    /// is made to access the database and the cycle repeats.
48    ///
49    /// There can only be a single busy handler defined for each database
50    /// connection. Setting a new busy handler clears any previously set
51    /// handler. Note that calling [`busy_timeout()`](Connection::busy_timeout) or evaluating `PRAGMA
52    /// busy_timeout=N` will change the busy handler and thus
53    /// clear any previously set busy handler.
54    ///
55    /// Newly created connections default to a [`busy_timeout()`](Connection::busy_timeout) handler
56    /// with a timeout of 5000ms, although this is subject to change.
57    pub fn busy_handler(&self, callback: Option<fn(i32) -> bool>) -> Result<()> {
58        unsafe extern "C" fn busy_handler_callback(p_arg: *mut c_void, count: c_int) -> c_int {
59            let handler_fn: fn(i32) -> bool = mem::transmute(p_arg);
60            if let Ok(true) = catch_unwind(|| handler_fn(count)) {
61                1
62            } else {
63                0
64            }
65        }
66        let mut c = self.db.borrow_mut();
67        let r = match callback {
68            Some(f) => unsafe {
69                ffi::sqlite3_busy_handler(c.db(), Some(busy_handler_callback), f as *mut c_void)
70            },
71            None => unsafe { ffi::sqlite3_busy_handler(c.db(), None, ptr::null_mut()) },
72        };
73        c.decode_result(r)
74    }
75}
76
77impl InnerConnection {
78    #[inline]
79    fn busy_timeout(&mut self, timeout: c_int) -> Result<()> {
80        let r = unsafe { ffi::sqlite3_busy_timeout(self.db, timeout) };
81        self.decode_result(r)
82    }
83}
84
85#[cfg(test)]
86mod test {
87    use std::sync::atomic::{AtomicBool, Ordering};
88    use std::sync::mpsc::sync_channel;
89    use std::thread;
90    use std::time::Duration;
91
92    use crate::{Connection, Error, ErrorCode, Result, TransactionBehavior};
93
94    #[test]
95    fn test_default_busy() -> Result<()> {
96        let temp_dir = tempfile::tempdir().unwrap();
97        let path = temp_dir.path().join("test.db3");
98
99        let mut db1 = Connection::open(&path)?;
100        let tx1 = db1.transaction_with_behavior(TransactionBehavior::Exclusive)?;
101        let db2 = Connection::open(&path)?;
102        let r: Result<()> = db2.query_row("PRAGMA schema_version", [], |_| unreachable!());
103        match r.unwrap_err() {
104            Error::SqliteFailure(err, _) => {
105                assert_eq!(err.code, ErrorCode::DatabaseBusy);
106            }
107            err => panic!("Unexpected error {}", err),
108        }
109        tx1.rollback()
110    }
111
112    #[test]
113    #[ignore] // FIXME: unstable
114    fn test_busy_timeout() {
115        let temp_dir = tempfile::tempdir().unwrap();
116        let path = temp_dir.path().join("test.db3");
117
118        let db2 = Connection::open(&path).unwrap();
119        db2.busy_timeout(Duration::from_secs(1)).unwrap();
120
121        let (rx, tx) = sync_channel(0);
122        let child = thread::spawn(move || {
123            let mut db1 = Connection::open(&path).unwrap();
124            let tx1 = db1
125                .transaction_with_behavior(TransactionBehavior::Exclusive)
126                .unwrap();
127            rx.send(1).unwrap();
128            thread::sleep(Duration::from_millis(100));
129            tx1.rollback().unwrap();
130        });
131
132        assert_eq!(tx.recv().unwrap(), 1);
133        let _ = db2
134            .query_row("PRAGMA schema_version", [], |row| row.get::<_, i32>(0))
135            .expect("unexpected error");
136
137        child.join().unwrap();
138    }
139
140    #[test]
141    #[ignore] // FIXME: unstable
142    fn test_busy_handler() {
143        static CALLED: AtomicBool = AtomicBool::new(false);
144        fn busy_handler(_: i32) -> bool {
145            CALLED.store(true, Ordering::Relaxed);
146            thread::sleep(Duration::from_millis(100));
147            true
148        }
149
150        let temp_dir = tempfile::tempdir().unwrap();
151        let path = temp_dir.path().join("test.db3");
152
153        let db2 = Connection::open(&path).unwrap();
154        db2.busy_handler(Some(busy_handler)).unwrap();
155
156        let (rx, tx) = sync_channel(0);
157        let child = thread::spawn(move || {
158            let mut db1 = Connection::open(&path).unwrap();
159            let tx1 = db1
160                .transaction_with_behavior(TransactionBehavior::Exclusive)
161                .unwrap();
162            rx.send(1).unwrap();
163            thread::sleep(Duration::from_millis(100));
164            tx1.rollback().unwrap();
165        });
166
167        assert_eq!(tx.recv().unwrap(), 1);
168        let _ = db2
169            .query_row("PRAGMA schema_version", [], |row| row.get::<_, i32>(0))
170            .expect("unexpected error");
171        assert_eq!(CALLED.load(Ordering::Relaxed), true);
172
173        child.join().unwrap();
174    }
175}
rusqlite/busy.rs

rusqlite/
busy.rs
