Changes outside sqlx_rt

Author: martin-kolarik

Cargo.lock

@@ -22,6 +22,7 @@ json = ["serde", "serde_json"]
 _rt-async-global-executor = [
     "async-global-executor",
     "async-io-global-executor",
+    "async-lock",
     "async-net",
 ]
 _rt-async-std = ["async-std", "async-io-std"]
@@ -62,6 +63,7 @@ uuid = { workspace = true, optional = true }
 
 async-io-global-executor = { package = "async-io", version = "2.2", optional = true }
 async-io-std = { package = "async-io", version = "1.9.0", optional = true }
+async-lock = { version = "3.4.0", optional = true }
 async-net = { package = "async-net", version = "2.0.0", optional = true }
 base64 = { version = "0.22.0", default-features = false, features = ["std"] }
 bytes = "1.1.0"

sqlx-core/Cargo.toml

@@ -202,10 +202,16 @@ pub async fn connect_tcp<Ws: WithSocket>(
         return Ok(with_socket.with_socket(stream).await);
     }
 
-    #[cfg(feature = "_rt-async-global-executor")]
+    #[cfg(any(feature = "_rt-async-global-executor", feature = "_rt-smol"))]
     {
+        #[cfg(feature = "_rt-async-global-executor")]
         use async_io_global_executor::Async;
+        #[cfg(feature = "_rt-async-global-executor")]
         use async_net::resolve;
+        #[cfg(feature = "_rt-smol")]
+        use smol::Async;
+        #[cfg(feature = "_rt-smol")]
+        use smol::net::resolve;
         use std::net::TcpStream;
 
         let mut last_err = None;
@@ -270,7 +276,11 @@ pub async fn connect_tcp<Ws: WithSocket>(
         };
     }
 
-    #[cfg(not(all(feature = "_rt-async-global-executor", feature = "_rt-async-std")))]
+    #[cfg(not(all(
+        feature = "_rt-async-global-executor",
+        feature = "_rt-async-std",
+        feature = "_rt-smol"
+    )))]
     #[allow(unreachable_code)]
     {
         crate::rt::missing_rt((host, port, with_socket))
@@ -315,7 +325,21 @@ pub async fn connect_uds<P: AsRef<Path>, Ws: WithSocket>(
             return Ok(with_socket.with_socket(stream).await);
         }
 
-        #[cfg(not(all(feature = "_rt-async-global-executor", feature = "_rt-async-std")))]
+        #[cfg(feature = "_rt-smol")]
+        {
+            use smol::Async;
+            use std::os::unix::net::UnixStream;
+
+            let stream = Async::<UnixStream>::connect(path).await?;
+
+            return Ok(with_socket.with_socket(stream).await);
+        }
+
+        #[cfg(not(all(
+            feature = "_rt-async-global-executor",
+            feature = "_rt-async-std",
+            feature = "_rt-smol"
+        )))]
         #[allow(unreachable_code)]
         {
             crate::rt::missing_rt((path, with_socket))

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

@@ -4,14 +4,20 @@
 // We'll generally lean towards Tokio's types as those are more featureful
 // (including `tokio-console` support) and more widely deployed.
 
+#[cfg(all(feature = "_rt-async-global-executor", not(feature = "_rt-tokio")))]
+pub use async_lock::{Mutex as AsyncMutex, MutexGuard as AsyncMutexGuard};
+
 #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
 pub use async_std::sync::{Mutex as AsyncMutex, MutexGuard as AsyncMutexGuard};
 
+#[cfg(all(feature = "_rt-smol", not(feature = "_rt-tokio")))]
+pub use smol::lock::{Mutex as AsyncMutex, MutexGuard as AsyncMutexGuard};
+
 #[cfg(feature = "_rt-tokio")]
 pub use tokio::sync::{Mutex as AsyncMutex, MutexGuard as AsyncMutexGuard};
 
 pub struct AsyncSemaphore {
-    // We use the semaphore from futures-intrusive as the one from async-std
+    // We use the semaphore from futures-intrusive as the one from async-lock
     // is missing the ability to add arbitrary permits, and is not guaranteed to be fair:
     // * https://github.com/smol-rs/async-lock/issues/22
     // * https://github.com/smol-rs/async-lock/issues/23
@@ -20,7 +26,14 @@ pub struct AsyncSemaphore {
     // and there are some soundness concerns (although it turns out any intrusive future is unsound
     // in MIRI due to the necessitated mutable aliasing):
     // https://github.com/launchbadge/sqlx/issues/1668
-    #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
+    #[cfg(all(
+        any(
+            feature = "_rt-async-global-executor",
+            feature = "_rt-async-std",
+            feature = "_rt-smol"
+        ),
+        not(feature = "_rt-tokio")
+    ))]
     inner: futures_intrusive::sync::Semaphore,
 
     #[cfg(feature = "_rt-tokio")]
@@ -30,12 +43,24 @@ pub struct AsyncSemaphore {
 impl AsyncSemaphore {
     #[track_caller]
     pub fn new(fair: bool, permits: usize) -> Self {
-        if cfg!(not(any(feature = "_rt-async-std", feature = "_rt-tokio"))) {
+        if cfg!(not(any(
+            feature = "_rt-async-global-executor",
+            feature = "_rt-async-std",
+            feature = "_rt-smol",
+            feature = "_rt-tokio"
+        ))) {
             crate::rt::missing_rt((fair, permits));
         }
 
         AsyncSemaphore {
-            #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
+            #[cfg(all(
+                any(
+                    feature = "_rt-async-global-executor",
+                    feature = "_rt-async-std",
+                    feature = "_rt-smol"
+                ),
+                not(feature = "_rt-tokio")
+            ))]
             inner: futures_intrusive::sync::Semaphore::new(fair, permits),
             #[cfg(feature = "_rt-tokio")]
             inner: {
@@ -46,18 +71,39 @@ impl AsyncSemaphore {
     }
 
     pub fn permits(&self) -> usize {
-        #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
+        #[cfg(all(
+            any(
+                feature = "_rt-async-global-executor",
+                feature = "_rt-async-std",
+                feature = "_rt-smol"
+            ),
+            not(feature = "_rt-tokio")
+        ))]
         return self.inner.permits();
 
         #[cfg(feature = "_rt-tokio")]
         return self.inner.available_permits();
 
-        #[cfg(not(any(feature = "_rt-async-std", feature = "_rt-tokio")))]
+        #[cfg(not(any(
+            any(
+                feature = "_rt-async-global-executor",
+                feature = "_rt-async-std",
+                feature = "_rt-smol"
+            ),
+            feature = "_rt-tokio"
+        )))]
         crate::rt::missing_rt(())
     }
 
     pub async fn acquire(&self, permits: u32) -> AsyncSemaphoreReleaser<'_> {
-        #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
+        #[cfg(all(
+            any(
+                feature = "_rt-async-global-executor",
+                feature = "_rt-async-std",
+                feature = "_rt-smol"
+            ),
+            not(feature = "_rt-tokio")
+        ))]
         return AsyncSemaphoreReleaser {
             inner: self.inner.acquire(permits as usize).await,
         };
@@ -73,12 +119,26 @@ impl AsyncSemaphore {
                 .expect("BUG: we do not expose the `.close()` method"),
         };
 
-        #[cfg(not(any(feature = "_rt-async-std", feature = "_rt-tokio")))]
+        #[cfg(not(any(
+            any(
+                feature = "_rt-async-global-executor",
+                feature = "_rt-async-std",
+                feature = "_rt-smol"
+            ),
+            feature = "_rt-tokio"
+        )))]
         crate::rt::missing_rt(permits)
     }
 
     pub fn try_acquire(&self, permits: u32) -> Option<AsyncSemaphoreReleaser<'_>> {
-        #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
+        #[cfg(all(
+            any(
+                feature = "_rt-async-global-executor",
+                feature = "_rt-async-std",
+                feature = "_rt-smol"
+            ),
+            not(feature = "_rt-tokio")
+        ))]
         return Some(AsyncSemaphoreReleaser {
             inner: self.inner.try_acquire(permits as usize)?,
         });
@@ -88,18 +148,39 @@ impl AsyncSemaphore {
             inner: self.inner.try_acquire_many(permits).ok()?,
         });
 
-        #[cfg(not(any(feature = "_rt-async-std", feature = "_rt-tokio")))]
+        #[cfg(not(any(
+            any(
+                feature = "_rt-async-global-executor",
+                feature = "_rt-async-std",
+                feature = "_rt-smol"
+            ),
+            feature = "_rt-tokio"
+        )))]
         crate::rt::missing_rt(permits)
     }
 
     pub fn release(&self, permits: usize) {
-        #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
+        #[cfg(all(
+            any(
+                feature = "_rt-async-global-executor",
+                feature = "_rt-async-std",
+                feature = "_rt-smol"
+            ),
+            not(feature = "_rt-tokio")
+        ))]
         return self.inner.release(permits);
 
         #[cfg(feature = "_rt-tokio")]
         return self.inner.add_permits(permits);
 
-        #[cfg(not(any(feature = "_rt-async-std", feature = "_rt-tokio")))]
+        #[cfg(not(any(
+            any(
+                feature = "_rt-async-global-executor",
+                feature = "_rt-async-std",
+                feature = "_rt-smol"
+            ),
+            feature = "_rt-tokio"
+        )))]
         crate::rt::missing_rt(permits)
     }
 }
@@ -114,30 +195,58 @@ pub struct AsyncSemaphoreReleaser<'a> {
     // and there are some soundness concerns (although it turns out any intrusive future is unsound
     // in MIRI due to the necessitated mutable aliasing):
     // https://github.com/launchbadge/sqlx/issues/1668
-    #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
+    #[cfg(all(
+        any(
+            feature = "_rt-async-global-executor",
+            feature = "_rt-async-std",
+            feature = "_rt-smol"
+        ),
+        not(feature = "_rt-tokio")
+    ))]
     inner: futures_intrusive::sync::SemaphoreReleaser<'a>,
 
     #[cfg(feature = "_rt-tokio")]
     inner: tokio::sync::SemaphorePermit<'a>,
 
-    #[cfg(not(any(feature = "_rt-async-std", feature = "_rt-tokio")))]
+    #[cfg(not(any(
+        any(
+            feature = "_rt-async-global-executor",
+            feature = "_rt-async-std",
+            feature = "_rt-smol"
+        ),
+        feature = "_rt-tokio"
+    )))]
     _phantom: std::marker::PhantomData<&'a ()>,
 }
 
 impl AsyncSemaphoreReleaser<'_> {
     pub fn disarm(self) {
-        #[cfg(feature = "_rt-tokio")]
+        #[cfg(all(
+            any(
+                feature = "_rt-async-global-executor",
+                feature = "_rt-async-std",
+                feature = "_rt-smol"
+            ),
+            not(feature = "_rt-tokio")
+        ))]
         {
-            self.inner.forget();
+            let mut this = self;
+            this.inner.disarm();
         }
 
-        #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
+        #[cfg(feature = "_rt-tokio")]
         {
-            let mut this = self;
-            this.inner.disarm();
+            self.inner.forget();
         }
 
-        #[cfg(not(any(feature = "_rt-async-std", feature = "_rt-tokio")))]
+        #[cfg(not(any(
+            any(
+                feature = "_rt-async-global-executor",
+                feature = "_rt-async-std",
+                feature = "_rt-smol"
+            ),
+            feature = "_rt-tokio"
+        )))]
         crate::rt::missing_rt(())
     }
 }

sqlx-core/src/sync.rs

@@ -72,11 +72,24 @@ where
         TOKIO_RT.block_on(f)
     }
 
-    #[cfg(all(feature = "_rt-async-std", not(feature = "tokio")))]
+    #[cfg(all(
+        any(feature = "_rt-async-global-executor", feature = "_rt-smol"),
+        not(feature = "_rt-tokio")
+    ))]
+    {
+        sqlx_core::rt::test_block_on(f)
+    }
+
+    #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
     {
         async_std::task::block_on(f)
     }
 
-    #[cfg(not(any(feature = "_rt-async-std", feature = "tokio")))]
+    #[cfg(not(any(
+        feature = "_rt-async-global-executor",
+        feature = "_rt-async-std",
+        feature = "_rt-smol",
+        feature = "_rt-tokio"
+    )))]
     sqlx_core::rt::missing_rt(f)
 }