WIP: Unlock Notification

To do: unlock_notify_cb

src/lib.rs

@@ -125,8 +125,7 @@ pub mod limits;
 mod hooks;
 #[cfg(feature = "hooks")]
 pub use hooks::*;
-#[cfg(feature = "unlock_notify")]
+mod unlock_notify;
-pub mod unlock_notify;
 
 // Number of cached prepared statements we'll hold on to.
 const STATEMENT_CACHE_DEFAULT_CAPACITY: usize = 16;
@@ -864,16 +863,40 @@ impl InnerConnection {
         }
         let mut c_stmt: *mut ffi::sqlite3_stmt = unsafe { mem::uninitialized() };
         let c_sql = try!(str_to_cstring(sql));
+        let len_with_nul = (sql.len() + 1) as c_int;
         let r = unsafe {
-            let len_with_nul = (sql.len() + 1) as c_int;
+            if cfg!(feature = "unlock_notify") {
-            ffi::sqlite3_prepare_v2(self.db(),
+                let mut rc;
-                                    c_sql.as_ptr(),
+                loop {
-                                    len_with_nul,
+                    rc = ffi::sqlite3_prepare_v2(
-                                    &mut c_stmt,
+                        self.db(),
-                                    ptr::null_mut())
+                        c_sql.as_ptr(),
+                        len_with_nul,
+                        &mut c_stmt,
+                        ptr::null_mut(),
+                    );
+                    if rc != ffi::SQLITE_LOCKED {
+                        break;
+                    }
+                    rc = unlock_notify::wait_for_unlock_notify(self.db);
+                    if rc != ffi::SQLITE_OK {
+                        break;
+                    }
+                }
+                rc
+            } else {
+                ffi::sqlite3_prepare_v2(
+                    self.db(),
+                    c_sql.as_ptr(),
+                    len_with_nul,
+                    &mut c_stmt,
+                    ptr::null_mut(),
+                )
+            }
         };
-        self.decode_result(r)
+        self.decode_result(r).map(|_| {
-            .map(|_| Statement::new(conn, RawStatement::new(c_stmt)))
+            Statement::new(conn, RawStatement::new(c_stmt, self.db()))
+        })
     }
 
     fn changes(&mut self) -> c_int {

src/raw_statement.rs

@@ -2,14 +2,15 @@ use std::ffi::CStr;
 use std::ptr;
 use std::os::raw::c_int;
 use super::ffi;
+use super::unlock_notify;
 
 // Private newtype for raw sqlite3_stmts that finalize themselves when dropped.
 #[derive(Debug)]
-pub struct RawStatement(*mut ffi::sqlite3_stmt);
+pub struct RawStatement(*mut ffi::sqlite3_stmt, *mut ffi::sqlite3);
 
 impl RawStatement {
-    pub fn new(stmt: *mut ffi::sqlite3_stmt) -> RawStatement {
+    pub fn new(stmt: *mut ffi::sqlite3_stmt, db: *mut ffi::sqlite3) -> RawStatement {
-        RawStatement(stmt)
+        RawStatement(stmt, db)
     }
 
     pub unsafe fn ptr(&self) -> *mut ffi::sqlite3_stmt {
@@ -29,7 +30,29 @@ impl RawStatement {
     }
 
     pub fn step(&self) -> c_int {
-        unsafe { ffi::sqlite3_step(self.0) }
+        if cfg!(feature = "unlock_notify") {
+            let mut rc;
+            loop {
+                rc = unsafe { ffi::sqlite3_step(self.0) };
+                if rc == ffi::SQLITE_LOCKED {
+                    if unsafe { ffi::sqlite3_extended_errcode(self.1) } !=
+                        ffi::SQLITE_LOCKED_SHAREDCACHE
+                    {
+                        break;
+                    }
+                } else if rc != ffi::SQLITE_LOCKED_SHAREDCACHE {
+                    break;
+                }
+                rc = unlock_notify::wait_for_unlock_notify(self.1);
+                if rc != ffi::SQLITE_OK {
+                    break;
+                }
+                self.reset();
+            }
+            rc
+        } else {
+            unsafe { ffi::sqlite3_step(self.0) }
+        }
     }
 
     pub fn reset(&self) -> c_int {

src/statement.rs

@@ -462,7 +462,7 @@ impl<'conn> Statement<'conn> {
     }
 
     fn finalize_(&mut self) -> Result<()> {
-        let mut stmt = RawStatement::new(ptr::null_mut());
+        let mut stmt = RawStatement::new(ptr::null_mut(), ptr::null_mut());
         mem::swap(&mut stmt, &mut self.stmt);
         self.conn.decode_result(stmt.finalize())
     }
@@ -470,7 +470,7 @@ impl<'conn> Statement<'conn> {
 
 impl<'conn> Into<RawStatement> for Statement<'conn> {
     fn into(mut self) -> RawStatement {
-        let mut stmt = RawStatement::new(ptr::null_mut());
+        let mut stmt = RawStatement::new(ptr::null_mut(), ptr::null_mut());
         mem::swap(&mut stmt, &mut self.stmt);
         stmt
     }

src/unlock_notify.rs

@@ -1,16 +1,20 @@
 //! [Unlock Notification](http://sqlite.org/unlock_notify.html)
 
-use std::sync::{Mutex, Condvar};
+#[cfg(feature = "unlock_notify")]
-use std::os::raw::{c_char, c_int, c_void};
+use std::sync::{Condvar, Mutex};
+use std::os::raw::c_int;
+#[cfg(feature = "unlock_notify")]
+use std::os::raw::c_void;
 
 use ffi;
-use InnerConnection;
 
+#[cfg(feature = "unlock_notify")]
 struct UnlockNotification {
-    cond: Condvar, // Condition variable to wait on
+    cond: Condvar,      // Condition variable to wait on
     mutex: Mutex<bool>, // Mutex to protect structure
 }
 
+#[cfg(feature = "unlock_notify")]
 impl UnlockNotification {
     fn new() -> UnlockNotification {
         UnlockNotification {
@@ -34,8 +38,9 @@ impl UnlockNotification {
 }
 
 /// This function is an unlock-notify callback
+#[cfg(feature = "unlock_notify")]
 unsafe extern "C" fn unlock_notify_cb(ap_arg: *mut *mut c_void, n_arg: c_int) {
-  /*int i;
+    /*int i;
   for(i=0; i<nArg; i++){
     UnlockNotification *p = (UnlockNotification *)apArg[i];
     pthread_mutex_lock(&p->mutex);
@@ -45,36 +50,37 @@ unsafe extern "C" fn unlock_notify_cb(ap_arg: *mut *mut c_void, n_arg: c_int) {
   }*/
 }
 
-impl InnerConnection {
+/// This function assumes that an SQLite API call (either `sqlite3_prepare_v2()`
-    fn blocking_prepare(&mut self,
+/// or `sqlite3_step()`) has just returned `SQLITE_LOCKED`. The argument is the
-            z_sql: *const c_char,
+/// associated database connection.
-            n_byte: c_int,
+///
-            pp_stmt: *mut *mut ffi::sqlite3_stmt,
+/// This function calls `sqlite3_unlock_notify()` to register for an
-            pz_tail: *mut *const c_char) -> c_int {
+/// unlock-notify callback, then blocks until that callback is delivered
-        let mut rc;
+/// and returns `SQLITE_OK`. The caller should then retry the failed operation.
-        loop {
+///
-            rc = unsafe {
+/// Or, if `sqlite3_unlock_notify()` indicates that to block would deadlock
-                ffi::sqlite3_prepare_v2(self.db, z_sql, n_byte, pp_stmt, pz_tail)
+/// the system, then this function returns `SQLITE_LOCKED` immediately. In
-            };
+/// this case the caller should not retry the operation and should roll
-            if rc != ffi::SQLITE_LOCKED {
+/// back the current transaction (if any).
-                break;
+#[cfg(feature = "unlock_notify")]
-            }
+pub fn wait_for_unlock_notify(db: *mut ffi::sqlite3) -> c_int {
-            rc = self.wait_for_unlock_notify();
+    let mut un = UnlockNotification::new();
-            if rc != ffi::SQLITE_OK {
+    /* Register for an unlock-notify callback. */
-                break;
+    let rc = unsafe {
-            }
+        ffi::sqlite3_unlock_notify(
-        }
+            db,
-        rc
+            Some(unlock_notify_cb),
+            &mut un as *mut UnlockNotification as *mut c_void,
+        )
+    };
+    debug_assert!(rc == ffi::SQLITE_LOCKED || rc == ffi::SQLITE_OK);
+    if rc == ffi::SQLITE_OK {
+        un.wait();
     }
+    rc
+}
 
-    fn wait_for_unlock_notify(&mut self) -> c_int {
+#[cfg(not(feature = "unlock_notify"))]
-        let mut un = UnlockNotification::new();
+pub fn wait_for_unlock_notify(_db: *mut ffi::sqlite3) -> c_int {
-        /* Register for an unlock-notify callback. */
+    unreachable!()
-        let rc = unsafe { ffi::sqlite3_unlock_notify(self.db, Some(unlock_notify_cb), &mut un as *mut UnlockNotification as *mut c_void) };
+}
-        debug_assert!(rc == ffi::SQLITE_LOCKED || rc == ffi::SQLITE_OK);
-        if rc == ffi::SQLITE_OK {
-            un.wait();
-        }
-        rc
-    }
-}