Unsafe fixes, smol executor added

Author: martin-kolarik

Cargo.lock

@@ -77,6 +77,7 @@ _unstable-all-types = [
 # Base runtime features without TLS
 runtime-async-global-executor = ["_rt-async-global-executor", "sqlx-core/_rt-async-global-executor", "sqlx-macros?/_rt-async-global-executor"]
 runtime-async-std = ["_rt-async-std", "sqlx-core/_rt-async-std", "sqlx-macros?/_rt-async-std"]
+runtime-smol = ["_rt-smol", "sqlx-core/_rt-smol", "sqlx-macros?/_rt-smol"]
 runtime-tokio = ["_rt-tokio", "sqlx-core/_rt-tokio", "sqlx-macros?/_rt-tokio"]
 
 # TLS features
@@ -98,12 +99,16 @@ runtime-async-global-executor-rustls = ["runtime-async-global-executor", "tls-ru
 runtime-async-std-native-tls = ["runtime-async-std", "tls-native-tls"]
 runtime-async-std-rustls = ["runtime-async-std", "tls-rustls-ring"]
 
+runtime-smol-native-tls = ["runtime-smol", "tls-native-tls"]
+runtime-smol-rustls = ["runtime-smol", "tls-rustls-ring"]
+
 runtime-tokio-native-tls = ["runtime-tokio", "tls-native-tls"]
 runtime-tokio-rustls = ["runtime-tokio", "tls-rustls-ring"]
 
 # for conditional compilation
 _rt-async-global-executor = []
 _rt-async-std = []
+_rt-smol = []
 _rt-tokio = []
 _sqlite = []
 
@@ -166,6 +171,10 @@ features = ["async-io"]
 [workspace.dependencies.async-std]
 version = "1.12"
 
+[workspace.dependencies.smol]
+version = "2.0"
+default-features = false
+
 [workspace.dependencies.tokio]
 version = "1"
 features = ["time", "net", "sync", "fs", "io-util", "rt"]

Cargo.toml

@@ -25,6 +25,7 @@ _rt-async-global-executor = [
     "async-net",
 ]
 _rt-async-std = ["async-std", "async-io-std"]
+_rt-smol = ["smol"]
 _rt-tokio = ["tokio", "tokio-stream"]
 _tls-native-tls = ["native-tls"]
 _tls-rustls-aws-lc-rs = ["_tls-rustls", "rustls/aws-lc-rs", "webpki-roots"]
@@ -40,6 +41,7 @@ offline = ["serde", "either/serde"]
 # Runtimes
 async-global-executor = { workspace = true, optional = true }
 async-std = { workspace = true, optional = true }
+smol = { workspace = true, optional = true }
 tokio = { workspace = true, optional = true }
 
 # TLS

sqlx-core/Cargo.toml

@@ -18,7 +18,7 @@
 // The only unsafe code in SQLx is that necessary to interact with native APIs like with SQLite,
 // and that can live in its own separate driver crate.
 // temporary
-// #![forbid(unsafe_code)]
+#![forbid(unsafe_code)]
 // Allows an API be documented as only available in some specific platforms.
 // <https://doc.rust-lang.org/unstable-book/language-features/doc-cfg.html>
 #![cfg_attr(docsrs, feature(doc_cfg))]

sqlx-core/src/lib.rs

@@ -10,6 +10,9 @@ pub mod rt_async_global_executor;
 #[cfg(feature = "_rt-async-std")]
 pub mod rt_async_std;
 
+#[cfg(feature = "_rt-smol")]
+pub mod rt_smol;
+
 #[cfg(feature = "_rt-tokio")]
 pub mod rt_tokio;
 
@@ -22,6 +25,8 @@ pub enum JoinHandle<T> {
     AsyncGlobalExecutor(rt_async_global_executor::JoinHandle<T>),
     #[cfg(feature = "_rt-async-std")]
     AsyncStd(async_std::task::JoinHandle<T>),
+    #[cfg(feature = "_rt-smol")]
+    Smol(rt_smol::JoinHandle<T>),
     #[cfg(feature = "_rt-tokio")]
     Tokio(tokio::task::JoinHandle<T>),
     // `PhantomData<T>` requires `T: Unpin`
@@ -41,14 +46,23 @@ pub async fn timeout<F: Future>(duration: Duration, f: F) -> Result<F::Output, T
         return rt_async_global_executor::timeout(duration, f).await;
     }
 
+    #[cfg(feature = "_rt-smol")]
+    {
+        return rt_smol::timeout(duration, f).await;
+    }
+
     #[cfg(feature = "_rt-async-std")]
     {
         return async_std::future::timeout(duration, f)
             .await
             .map_err(|_| TimeoutError);
     }
 
-    #[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)]
     missing_rt((duration, f))
 }
@@ -64,12 +78,21 @@ pub async fn sleep(duration: Duration) {
         return rt_async_global_executor::sleep(duration).await;
     }
 
+    #[cfg(feature = "_rt-smol")]
+    {
+        return rt_smol::sleep(duration).await;
+    }
+
     #[cfg(feature = "_rt-async-std")]
     {
         return async_std::task::sleep(duration).await;
     }
 
-    #[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)]
     missing_rt(duration)
 }
@@ -97,7 +120,11 @@ where
         return JoinHandle::AsyncStd(async_std::task::spawn(fut));
     }
 
-    #[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)]
     missing_rt(fut)
 }
@@ -125,7 +152,18 @@ where
         return JoinHandle::AsyncStd(async_std::task::spawn_blocking(f));
     }
 
-    #[cfg(not(all(feature = "_rt-async-global-executor", feature = "_rt-async-std",)))]
+    #[cfg(feature = "_rt-smol")]
+    {
+        return JoinHandle::Smol(rt_smol::JoinHandle {
+            task: Some(smol::unblock(f)),
+        });
+    }
+
+    #[cfg(not(all(
+        feature = "_rt-async-global-executor",
+        feature = "_rt-async-std",
+        feature = "_rt-smol"
+    )))]
     #[allow(unreachable_code)]
     missing_rt(f)
 }
@@ -146,7 +184,16 @@ pub async fn yield_now() {
         return async_std::task::yield_now().await;
     }
 
-    #[cfg(not(all(feature = "_rt-async-global-executor", feature = "_rt-async-std",)))]
+    #[cfg(feature = "_rt-smol")]
+    {
+        return smol::future::yield_now().await;
+    }
+
+    #[cfg(not(all(
+        feature = "_rt-async-global-executor",
+        feature = "_rt-async-std",
+        feature = "_rt-smol"
+    )))]
     #[allow(unreachable_code)]
     missing_rt(())
 }
@@ -155,11 +202,13 @@ pub async fn yield_now() {
 pub fn test_block_on<F: Future>(f: F) -> F::Output {
     #[cfg(feature = "_rt-tokio")]
     {
-        return tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("failed to start Tokio runtime")
-            .block_on(f);
+        if rt_tokio::available() {
+            return tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("failed to start Tokio runtime")
+                .block_on(f);
+        }
     }
 
     #[cfg(feature = "_rt-async-global-executor")]
@@ -172,7 +221,16 @@ pub fn test_block_on<F: Future>(f: F) -> F::Output {
         return async_std::task::block_on(f);
     }
 
-    #[cfg(not(all(feature = "_rt-async-global-executor", feature = "_rt-async-std",)))]
+    #[cfg(feature = "_rt-smol")]
+    {
+        return smol::block_on(f);
+    }
+
+    #[cfg(not(all(
+        feature = "_rt-async-global-executor",
+        feature = "_rt-async-std",
+        feature = "_rt-smol"
+    )))]
     #[allow(unreachable_code)]
     missing_rt(f)
 }
@@ -183,7 +241,7 @@ pub fn missing_rt<T>(_unused: T) -> ! {
         panic!("this functionality requires a Tokio context")
     }
 
-    panic!("one of the `runtime-async-global-executor`, `runtime-async-std`, or `runtime-tokio` feature must be enabled")
+    panic!("one of the `runtime-async-global-executor`, `runtime-async-std`, `runtime-smol`, or `runtime-tokio` feature must be enabled")
 }
 
 impl<T: Send + 'static> Future for JoinHandle<T> {
@@ -196,6 +254,8 @@ impl<T: Send + 'static> Future for JoinHandle<T> {
             Self::AsyncGlobalExecutor(handle) => Pin::new(handle).poll(cx),
             #[cfg(feature = "_rt-async-std")]
             Self::AsyncStd(handle) => Pin::new(handle).poll(cx),
+            #[cfg(feature = "_rt-smol")]
+            Self::Smol(handle) => Pin::new(handle).poll(cx),
             #[cfg(feature = "_rt-tokio")]
             Self::Tokio(handle) => Pin::new(handle)
                 .poll(cx)

sqlx-core/src/rt/mod.rs

@@ -11,11 +11,11 @@ use async_io_global_executor::Async;
 
 impl Socket for Async<TcpStream> {
     fn try_read(&mut self, buf: &mut dyn ReadBuf) -> io::Result<usize> {
-        unsafe { self.get_mut().read(buf.init_mut()) }
+        self.get_ref().read(buf.init_mut())
     }
 
     fn try_write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        unsafe { self.get_mut().write(buf) }
+        self.get_ref().write(buf)
     }
 
     fn poll_read_ready(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
@@ -27,18 +27,18 @@ impl Socket for Async<TcpStream> {
     }
 
     fn poll_shutdown(&mut self, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        unsafe { Poll::Ready(self.get_mut().shutdown(Shutdown::Both)) }
+        Poll::Ready(self.get_ref().shutdown(Shutdown::Both))
     }
 }
 
 #[cfg(unix)]
 impl Socket for Async<std::os::unix::net::UnixStream> {
     fn try_read(&mut self, buf: &mut dyn ReadBuf) -> io::Result<usize> {
-        unsafe { self.get_mut().read(buf.init_mut()) }
+        self.get_ref().read(buf.init_mut())
     }
 
     fn try_write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        unsafe { self.get_mut().write(buf) }
+        self.get_ref().write(buf)
     }
 
     fn poll_read_ready(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
@@ -50,6 +50,6 @@ impl Socket for Async<std::os::unix::net::UnixStream> {
     }
 
     fn poll_shutdown(&mut self, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        unsafe { Poll::Ready(self.get_mut().shutdown(Shutdown::Both)) }
+        Poll::Ready(self.get_ref().shutdown(Shutdown::Both))
     }
 }