Merge pull request #160 from WattSense/sync_wrapper_clear_mutex_poison

sync wrapper: Clear mutex poison and provide underlying mutex guar

Author: weiznich

examples/sync-wrapper/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-diesel = { version = "2.1.0", default-features = false }
+diesel = { version = "2.1.0", default-features = false, features = ["returning_clauses_for_sqlite_3_35"] }
 diesel-async = { version = "0.4.0", path = "../../", features = ["sync-connection-wrapper", "async-connection-wrapper"] }
 diesel_migrations = "2.1.0"
 futures-util = "0.3.21"

examples/sync-wrapper/src/main.rs

@@ -2,7 +2,7 @@ use diesel::prelude::*;
 use diesel::sqlite::{Sqlite, SqliteConnection};
 use diesel_async::async_connection_wrapper::AsyncConnectionWrapper;
 use diesel_async::sync_connection_wrapper::SyncConnectionWrapper;
-use diesel_async::{AsyncConnection, RunQueryDsl, SimpleAsyncConnection};
+use diesel_async::{AsyncConnection, RunQueryDsl};
 use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
 
 // ordinary diesel model setup
@@ -15,7 +15,7 @@ table! {
 }
 
 #[allow(dead_code)]
-#[derive(Debug, Queryable, Selectable)]
+#[derive(Debug, Queryable, QueryableByName, Selectable)]
 #[diesel(table_name = users)]
 struct User {
     id: i32,
@@ -47,6 +47,27 @@ where
     .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)
 }
 
+async fn transaction(
+    async_conn: &mut SyncConnectionWrapper<InnerConnection>,
+    old_name: &str,
+    new_name: &str,
+) -> Result<Vec<User>, diesel::result::Error> {
+    async_conn
+        .transaction::<Vec<User>, diesel::result::Error, _>(|c| {
+            Box::pin(async {
+                if old_name.is_empty() {
+                    Ok(Vec::new())
+                } else {
+                    diesel::update(users::table.filter(users::name.eq(old_name)))
+                        .set(users::name.eq(new_name))
+                        .load(c)
+                        .await
+                }
+            })
+        })
+        .await
+}
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let db_url = std::env::var("DATABASE_URL").expect("Env var `DATABASE_URL` not set");
@@ -57,10 +78,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let mut sync_wrapper: SyncConnectionWrapper<InnerConnection> = establish(&db_url).await?;
 
-    sync_wrapper.batch_execute("DELETE FROM users").await?;
+    diesel::delete(users::table)
+        .execute(&mut sync_wrapper)
+        .await?;
 
-    sync_wrapper
-        .batch_execute("INSERT INTO users(id, name) VALUES (3, 'toto')")
+    diesel::insert_into(users::table)
+        .values((users::id.eq(3), users::name.eq("toto")))
+        .execute(&mut sync_wrapper)
         .await?;
 
     let data: Vec<User> = users::table
@@ -86,5 +110,28 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .await?;
     println!("{data:?}");
 
+    // a quick test to check if we correctly handle transactions
+    let mut conn_a: SyncConnectionWrapper<InnerConnection> = establish(&db_url).await?;
+    let mut conn_b: SyncConnectionWrapper<InnerConnection> = establish(&db_url).await?;
+
+    let handle_1 = tokio::spawn(async move {
+        loop {
+            let changed = transaction(&mut conn_a, "iLuke", "JustLuke").await;
+            println!("Changed {changed:?}");
+            std::thread::sleep(std::time::Duration::from_secs(1));
+        }
+    });
+
+    let handle_2 = tokio::spawn(async move {
+        loop {
+            let changed = transaction(&mut conn_b, "JustLuke", "iLuke").await;
+            println!("Changed {changed:?}");
+            std::thread::sleep(std::time::Duration::from_secs(1));
+        }
+    });
+
+    let _ = handle_2.await;
+    let _ = handle_1.await;
+
     Ok(())
 }

src/sync_connection_wrapper.rs

@@ -129,10 +129,17 @@ where
             let mut cache = <<<C as LoadConnection>::Row<'_, '_> as IntoOwnedRow<
                 <C as Connection>::Backend,
             >>::Cache as Default>::default();
-            conn.load(&query).map(|c| {
-                c.map(|row| row.map(|r| IntoOwnedRow::into_owned(r, &mut cache)))
-                    .collect::<Vec<QueryResult<O>>>()
-            })
+            let cursor = conn.load(&query)?;
+
+            let size_hint = cursor.size_hint();
+            let mut out = Vec::with_capacity(size_hint.1.unwrap_or(size_hint.0));
+            // we use an explicit loop here to easily propagate possible errors
+            // as early as possible
+            for row in cursor {
+                out.push(Ok(IntoOwnedRow::into_owned(row?, &mut cache)));
+            }
+
+            Ok(out)
         })
         .map_ok(|rows| futures_util::stream::iter(rows).boxed())
         .boxed()
@@ -235,9 +242,11 @@ impl<C> SyncConnectionWrapper<C> {
     {
         let inner = self.inner.clone();
         tokio::task::spawn_blocking(move || {
-            let mut inner = inner
-                .lock()
-                .expect("Mutex is poisoned, a thread must have panicked holding it.");
+            let mut inner = inner.lock().unwrap_or_else(|poison| {
+                // try to be resilient by providing the guard
+                inner.clear_poison();
+                poison.into_inner()
+            });
             task(&mut inner)
         })
         .unwrap_or_else(|err| QueryResult::Err(from_tokio_join_error(err)))
@@ -268,9 +277,11 @@ impl<C> SyncConnectionWrapper<C> {
 
         let (collect_bind_result, collector_data) = {
             let exclusive = self.inner.clone();
-            let mut inner = exclusive
-                .lock()
-                .expect("Mutex is poisoned, a thread must have panicked holding it.");
+            let mut inner = exclusive.lock().unwrap_or_else(|poison| {
+                // try to be resilient by providing the guard
+                exclusive.clear_poison();
+                poison.into_inner()
+            });
             let mut bind_collector =
                 <<C::Backend as Backend>::BindCollector<'_> as Default>::default();
             let metadata_lookup = inner.metadata_lookup();