apply madsim patch to sqlx v0.8.2

Cargo.toml

@@ -151,8 +151,9 @@ dotenvy = { version = "0.15.0", default-features = false }
 version = "1.12"
 
 [workspace.dependencies.tokio]
-version = "1"
-features = ["time", "net", "sync", "fs", "io-util", "rt"]
+version = "0.2.30"
+package = "madsim-tokio"
+features = ["time", "net", "sync", "fs", "io-util", "rt", "macros", "rt-multi-thread"]
 default-features = false
 
 [dependencies]

examples/mysql/todos/Cargo.toml

@@ -9,4 +9,4 @@ anyhow = "1.0"
 futures = "0.3"
 sqlx = { path = "../../../", features = [ "mysql", "runtime-tokio", "tls-native-tls" ] }
 clap = { version = "4", features = ["derive"] }
-tokio = { version = "1.20.0", features = ["rt", "macros"]}
+tokio = { workspace = true, optional = true }

examples/postgres/axum-social-with-tests/Cargo.toml

@@ -9,7 +9,7 @@ edition = "2021"
 # Primary crates
 axum = { version = "0.5.13", features = ["macros"] }
 sqlx = { path = "../../../", features = [ "runtime-tokio", "tls-rustls-ring", "postgres", "time", "uuid" ] }
-tokio = { version = "1.20.1", features = ["rt-multi-thread", "macros"] }
+tokio = { workspace = true, optional = true }
 
 # Important secondary crates
 argon2 = "0.4.1"

examples/postgres/chat/Cargo.toml

@@ -7,7 +7,7 @@ workspace = "../../../"
 [dependencies]
 sqlx = { path = "../../../", features = [ "postgres", "runtime-tokio", "tls-native-tls" ] }
 futures = "0.3.1"
-tokio = { version = "1.20.0", features = [ "rt-multi-thread", "macros" ] }
+tokio = { workspace = true, optional = true }
 ratatui = "0.27.0"
 crossterm = "0.27.0"
 unicode-width = "0.1"

examples/postgres/files/Cargo.toml

@@ -8,5 +8,5 @@ edition = "2021"
 [dependencies]
 anyhow = "1.0"
 sqlx = { path = "../../../", features = [ "postgres", "runtime-tokio", "tls-native-tls" ] }
-tokio = { version = "1.20.0", features = ["rt", "macros"]}
+tokio = { workspace = true, optional = true }
 dotenvy = "0.15.0"

examples/postgres/json/Cargo.toml

@@ -12,4 +12,4 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 sqlx = { path = "../../../", features = [ "runtime-tokio", "postgres", "json" ] }
 clap = { version = "4", features = ["derive"] }
-tokio = { version = "1.20.0", features = ["rt", "macros"]}
+tokio = { workspace = true, optional = true }

examples/postgres/listen/Cargo.toml

@@ -7,4 +7,4 @@ workspace = "../../../"
 [dependencies]
 sqlx = { path = "../../../", features = [ "runtime-tokio", "postgres" ] }
 futures = "0.3.1"
-tokio = { version = "1.20.0", features = ["rt-multi-thread", "macros", "time"]}
+tokio = { workspace = true, optional = true }

examples/postgres/mockable-todos/Cargo.toml

@@ -9,7 +9,7 @@ anyhow = "1.0"
 futures = "0.3"
 sqlx = { path = "../../../", features = [ "postgres", "runtime-tokio", "tls-native-tls" ] }
 clap = { version = "4", features = ["derive"] }
-tokio = { version = "1.20.0", features = ["rt", "macros"]}
+tokio = { workspace = true, optional = true }
 dotenvy = "0.15.0"
 async-trait = "0.1.41"
 mockall = "0.11"

examples/postgres/todos/Cargo.toml

@@ -9,5 +9,5 @@ anyhow = "1.0"
 futures = "0.3"
 sqlx = { path = "../../../", features = [ "postgres", "runtime-tokio", "tls-native-tls" ] }
 clap = { version = "4", features = ["derive"] }
-tokio = { version = "1.20.0", features = ["rt", "macros"]}
+tokio = { workspace = true, optional = true }
 dotenvy = "0.15.0"

examples/postgres/transaction/Cargo.toml

@@ -7,4 +7,4 @@ workspace = "../../../"
 [dependencies]
 sqlx = { path = "../../../", features = [ "postgres", "runtime-tokio", "tls-native-tls" ] }
 futures = "0.3.1"
-tokio = { version = "1.20.0", features = ["rt-multi-thread", "macros"]}
+tokio = { workspace = true, optional = true }

examples/sqlite/todos/Cargo.toml

@@ -9,4 +9,4 @@ anyhow = "1.0"
 futures = "0.3"
 sqlx = { path = "../../../", features = [ "sqlite", "runtime-tokio", "tls-native-tls" ] }
 clap = { version = "4", features = ["derive"] }
-tokio = { version = "1.20.0", features = ["rt", "macros"]}
+tokio = { workspace = true, optional = true }

sqlx-cli/Cargo.toml

@@ -26,7 +26,7 @@ path = "src/bin/cargo-sqlx.rs"
 
 [dependencies]
 dotenvy = "0.15.0"
-tokio = { version = "1.15.0", features = ["macros", "rt", "rt-multi-thread"] }
+tokio = { workspace = true, optional = true }
 sqlx = { workspace = true, default-features = false, features = [
     "runtime-tokio",
     "migrate",

sqlx-core/Cargo.toml

@@ -93,7 +93,6 @@ hashbrown = "0.14.5"
 
 [dev-dependencies]
 sqlx = { workspace = true, features = ["postgres", "sqlite", "mysql", "migrate", "macros", "time", "uuid"] }
-tokio = { version = "1", features = ["rt"] }
 
 [lints]
 workspace = true

sqlx-core/src/net/socket/mod.rs

@@ -13,7 +13,7 @@ use crate::io::ReadBuf;
 
 mod buffered;
 
-pub trait Socket: Send + Sync + Unpin + 'static {
+pub trait Socket: Send + Unpin + 'static {
     fn try_read(&mut self, buf: &mut dyn ReadBuf) -> io::Result<usize>;
 
     fn try_write(&mut self, buf: &[u8]) -> io::Result<usize>;

sqlx-sqlite/Cargo.toml

@@ -27,7 +27,6 @@ regexp = ["dep:regex"]
 futures-core = { version = "0.3.19", default-features = false }
 futures-channel = { version = "0.3.19", default-features = false, features = ["sink", "alloc", "std"] }
 # used by the SQLite worker thread to block on the async mutex that locks the database handle
-futures-executor = { version = "0.3.19" }
 futures-intrusive = "0.5.0"
 futures-util = { version = "0.3.19", default-features = false, features = ["alloc", "sink"] }
 
@@ -48,6 +47,8 @@ tracing = { version = "0.1.37", features = ["log"] }
 
 serde = { version = "1.0.145", features = ["derive"], optional = true }
 regex = { version = "1.5.5", optional = true }
+tokio = { workspace = true }
+async-channel = "2.2.0"
 
 [dependencies.libsqlite3-sys]
 version = "0.30.1"

sqlx-sqlite/src/connection/worker.rs

@@ -2,7 +2,6 @@ use std::borrow::Cow;
 use std::future::Future;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
-use std::thread;
 
 use futures_channel::oneshot;
 use futures_intrusive::sync::{Mutex, MutexGuard};
@@ -79,9 +78,7 @@ impl ConnectionWorker {
     pub(crate) async fn establish(params: EstablishParams) -> Result<Self, Error> {
         let (establish_tx, establish_rx) = oneshot::channel();
 
-        thread::Builder::new()
-            .name(params.thread_name.clone())
-            .spawn(move || {
+            tokio::spawn(async move {
                 let (command_tx, command_rx) = flume::bounded(params.command_channel_size);
 
                 let conn = match params.establish() {
@@ -116,10 +113,11 @@ impl ConnectionWorker {
                 // would rollback an already completed transaction.
                 let mut ignore_next_start_rollback = false;
 
-                for (cmd, span) in command_rx {
+                while let Ok((cmd, span)) = command_rx.recv_async().await {
                     let _guard = span.enter();
                     match cmd {
                         Command::Prepare { query, tx } => {
+                            // TODO(kwannoel): Make this async?
                             tx.send(prepare(&mut conn, &query).map(|prepared| {
                                 update_cached_statements_size(
                                     &conn,
@@ -130,6 +128,7 @@ impl ConnectionWorker {
                             .ok();
                         }
                         Command::Describe { query, tx } => {
+                            // TODO(kwannoel): Make this async?
                             tx.send(describe(&mut conn, &query)).ok();
                         }
                         Command::Execute {
@@ -143,15 +142,15 @@ impl ConnectionWorker {
                             {
                                 Ok(iter) => iter,
                                 Err(e) => {
-                                    tx.send(Err(e)).ok();
+                                    tx.send_async(Err(e)).await.ok();
                                     continue;
                                 }
                             };
 
                             match limit {
                                 None => {
                                     for res in iter {
-                                        if tx.send(res).is_err() {
+                                        if tx.send_async(res).await.is_err() {
                                             break;
                                         }
                                     }
@@ -166,12 +165,12 @@ impl ConnectionWorker {
                                                 rows_returned += 1;
                                                 if rows_returned >= limit {
                                                     drop(iter);
-                                                    let _ = tx.send(res);
+                                                    let _ = tx.send_async(res).await;
                                                     break;
                                                 }
                                             }
                                         }
-                                        if tx.send(res).is_err() {
+                                        if tx.send_async(res).await.is_err() {
                                             break;
                                         }
                                     }
@@ -190,7 +189,7 @@ impl ConnectionWorker {
                                     });
                             let res_ok = res.is_ok();
 
-                            if tx.blocking_send(res).is_err() && res_ok {
+                            if tx.send(res).await.is_err() && res_ok {
                                 // The BEGIN was processed but not acknowledged. This means no
                                 // `Transaction` was created and so there is no way to commit /
                                 // rollback this transaction. We need to roll it back
@@ -224,7 +223,7 @@ impl ConnectionWorker {
                             };
                             let res_ok = res.is_ok();
 
-                            if tx.blocking_send(res).is_err() && res_ok {
+                            if tx.send(res).await.is_err() && res_ok {
                                 // The COMMIT was processed but not acknowledged. This means that
                                 // the `Transaction` doesn't know it was committed and will try to
                                 // rollback on drop. We need to ignore that rollback.
@@ -252,7 +251,7 @@ impl ConnectionWorker {
                             let res_ok = res.is_ok();
 
                             if let Some(tx) = tx {
-                                if tx.blocking_send(res).is_err() && res_ok {
+                                if tx.send(res).await.is_err() && res_ok {
                                     // The ROLLBACK was processed but not acknowledged. This means
                                     // that the `Transaction` doesn't know it was rolled back and
                                     // will try to rollback again on drop. We need to ignore that
@@ -268,7 +267,7 @@ impl ConnectionWorker {
                         }
                         Command::UnlockDb => {
                             drop(conn);
-                            conn = futures_executor::block_on(shared.conn.lock());
+                            conn = shared.conn.lock().await;
                         }
                         Command::Ping { tx } => {
                             tx.send(()).ok();
@@ -283,7 +282,7 @@ impl ConnectionWorker {
                         }
                     }
                 }
-            })?;
+            });
 
         establish_rx.await.map_err(|_| Error::WorkerCrashed)?
     }
@@ -408,12 +407,13 @@ impl ConnectionWorker {
     pub(crate) fn shutdown(&mut self) -> impl Future<Output = Result<(), Error>> {
         let (tx, rx) = oneshot::channel();
 
-        let send_res = self
-            .command_tx
-            .send((Command::Shutdown { tx }, Span::current()))
-            .map_err(|_| Error::WorkerCrashed);
+        let command_tx = self.command_tx.clone();
 
         async move {
+            let send_res = command_tx
+                .send_async((Command::Shutdown { tx }, Span::current()))
+                .await
+                .map_err(|_| Error::WorkerCrashed);
             send_res?;
 
             // wait for the response
@@ -471,10 +471,6 @@ mod rendezvous_oneshot {
             self.inner.send((value, ack_tx)).map_err(|_| Canceled)?;
             ack_rx.await
         }
-
-        pub fn blocking_send(self, value: T) -> Result<(), Canceled> {
-            futures_executor::block_on(self.send(value))
-        }
     }
 
     pub struct Receiver<T> {