Initialization fixes

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "pyo3-asyncio"
 description = "PyO3 utilities for Python's Asyncio library"
-version = "0.13.5"
+version = "0.14.0"
 authors = ["Andrew J Westlake <awestlake87@yahoo.com>"]
 readme = "README.md"
 keywords = ["pyo3", "python", "ffi", "async", "asyncio"]
@@ -89,8 +89,11 @@ futures = "0.3"
 inventory = "0.1"
 lazy_static = "1.4"
 once_cell = "1.5"
-pyo3 = "0.13"
-pyo3-asyncio-macros = { path = "pyo3-asyncio-macros", version = "=0.13.5", optional = true }
+pyo3 = "0.14"
+pyo3-asyncio-macros = { path = "pyo3-asyncio-macros", version = "=0.14.0", optional = true }
+
+[dev-dependencies]
+pyo3 = { version = "0.14", features = ["macros"] }
 
 [dependencies.async-std]
 version = "1.9"

examples/async_std.rs

@@ -6,7 +6,7 @@ async fn main() -> PyResult<()> {
         let asyncio = py.import("asyncio")?;
 
         // convert asyncio.sleep into a Rust Future
-        pyo3_asyncio::into_future(asyncio.call_method1("sleep", (1.into_py(py),))?)
+        pyo3_asyncio::async_std::into_future(asyncio.call_method1("sleep", (1.into_py(py),))?)
     })?;
 
     println!("sleeping for 1s");

examples/tokio.rs

@@ -6,7 +6,7 @@ async fn main() -> PyResult<()> {
         let asyncio = py.import("asyncio")?;
 
         // convert asyncio.sleep into a Rust Future
-        pyo3_asyncio::into_future(asyncio.call_method1("sleep", (1.into_py(py),))?)
+        pyo3_asyncio::tokio::into_future(asyncio.call_method1("sleep", (1.into_py(py),))?)
     })?;
 
     println!("sleeping for 1s");

examples/tokio_current_thread.rs

@@ -6,7 +6,7 @@ async fn main() -> PyResult<()> {
         let asyncio = py.import("asyncio")?;
 
         // convert asyncio.sleep into a Rust Future
-        pyo3_asyncio::into_future(asyncio.call_method1("sleep", (1.into_py(py),))?)
+        pyo3_asyncio::tokio::into_future(asyncio.call_method1("sleep", (1.into_py(py),))?)
     })?;
 
     println!("sleeping for 1s");

examples/tokio_multi_thread.rs

@@ -6,7 +6,7 @@ async fn main() -> PyResult<()> {
         let asyncio = py.import("asyncio")?;
 
         // convert asyncio.sleep into a Rust Future
-        pyo3_asyncio::into_future(asyncio.call_method1("sleep", (1.into_py(py),))?)
+        pyo3_asyncio::tokio::into_future(asyncio.call_method1("sleep", (1.into_py(py),))?)
     })?;
 
     println!("sleeping for 1s");

pyo3-asyncio-macros/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "pyo3-asyncio-macros"
 description = "Proc Macro Attributes for PyO3 Asyncio"
-version = "0.13.5"
+version = "0.14.0"
 authors = ["Andrew J Westlake <awestlake87@yahoo.com>"]
 readme = "../README.md"
 keywords = ["pyo3", "python", "ffi", "async", "asyncio"]

pyo3-asyncio-macros/src/lib.rs

@@ -49,16 +49,14 @@ pub fn async_std_main(_attr: TokenStream, item: TokenStream) -> TokenStream {
                 #body
             }
 
-            pyo3::Python::with_gil(|py| {
-                pyo3_asyncio::with_runtime(py, || {
-                    pyo3_asyncio::async_std::run_until_complete(py, main())?;
+            pyo3::prepare_freethreaded_python();
 
-                    Ok(())
-                })
-                .map_err(|e| {
-                    e.print_and_set_sys_last_vars(py);
-                })
-                .unwrap();
+            pyo3::Python::with_gil(|py| {
+                pyo3_asyncio::async_std::run(py, main())
+                    .map_err(|e| {
+                        e.print_and_set_sys_last_vars(py);
+                    })
+                    .unwrap();
             });
         }
     };
@@ -128,6 +126,13 @@ pub fn tokio_main(args: TokenStream, item: TokenStream) -> TokenStream {
 ///     thread::sleep(Duration::from_secs(1));
 ///     Ok(())
 /// }
+///
+/// // blocking test functions can optionally accept an event_loop parameter
+/// #[pyo3_asyncio::async_std::test]
+/// fn test_blocking_sleep_with_event_loop(event_loop: PyObject) -> PyResult<()> {
+///     thread::sleep(Duration::from_secs(1));
+///     Ok(())
+/// }
 /// ```
 #[cfg(not(test))] // NOTE: exporting main breaks tests, we should file an issue.
 #[proc_macro_attribute]
@@ -140,15 +145,31 @@ pub fn async_std_test(_attr: TokenStream, item: TokenStream) -> TokenStream {
     let vis = &input.vis;
 
     let fn_impl = if input.sig.asyncness.is_none() {
+        // Optionally pass an event_loop parameter to blocking tasks
+        let task = if sig.inputs.is_empty() {
+            quote! {
+                Box::pin(pyo3_asyncio::async_std::re_exports::spawn_blocking(move || {
+                    #name()
+                }))
+            }
+        } else {
+            quote! {
+                let event_loop = Python::with_gil(|py| {
+                    pyo3_asyncio::async_std::get_current_loop(py).unwrap().into()
+                });
+                Box::pin(pyo3_asyncio::async_std::re_exports::spawn_blocking(move || {
+                    #name(event_loop)
+                }))
+            }
+        };
+
         quote! {
             #vis fn #name() -> std::pin::Pin<Box<dyn std::future::Future<Output = pyo3::PyResult<()>> + Send>> {
                 #sig {
                     #body
                 }
 
-                Box::pin(pyo3_asyncio::async_std::re_exports::spawn_blocking(move || {
-                    #name()
-                }))
+                #task
             }
         }
     } else {
@@ -204,6 +225,13 @@ pub fn async_std_test(_attr: TokenStream, item: TokenStream) -> TokenStream {
 ///     thread::sleep(Duration::from_secs(1));
 ///     Ok(())
 /// }
+///
+/// // blocking test functions can optionally accept an event_loop parameter
+/// #[pyo3_asyncio::tokio::test]
+/// fn test_blocking_sleep_with_event_loop(event_loop: PyObject) -> PyResult<()> {
+///     thread::sleep(Duration::from_secs(1));
+///     Ok(())
+/// }
 /// ```
 #[cfg(not(test))] // NOTE: exporting main breaks tests, we should file an issue.
 #[proc_macro_attribute]
@@ -216,14 +244,11 @@ pub fn tokio_test(_attr: TokenStream, item: TokenStream) -> TokenStream {
     let vis = &input.vis;
 
     let fn_impl = if input.sig.asyncness.is_none() {
-        quote! {
-            #vis fn #name() -> std::pin::Pin<Box<dyn std::future::Future<Output = pyo3::PyResult<()>> + Send>> {
-                #sig {
-                    #body
-                }
-
-                Box::pin(async {
-                    match pyo3_asyncio::tokio::get_runtime().spawn_blocking(&#name).await {
+        // Optionally pass an event_loop parameter to blocking tasks
+        let task = if sig.inputs.is_empty() {
+            quote! {
+                Box::pin(async move {
+                    match pyo3_asyncio::tokio::get_runtime().spawn_blocking(move || #name()).await {
                         Ok(result) => result,
                         Err(e) => {
                             assert!(e.is_panic());
@@ -232,6 +257,31 @@ pub fn tokio_test(_attr: TokenStream, item: TokenStream) -> TokenStream {
                     }
                 })
             }
+        } else {
+            quote! {
+                let event_loop = Python::with_gil(|py| {
+                    pyo3_asyncio::tokio::get_current_loop(py).unwrap().into()
+                });
+                Box::pin(async move {
+                    match pyo3_asyncio::tokio::get_runtime().spawn_blocking(move || #name(event_loop)).await {
+                        Ok(result) => result,
+                        Err(e) => {
+                            assert!(e.is_panic());
+                            Err(pyo3::exceptions::PyException::new_err("rust future panicked"))
+                        }
+                    }
+                })
+            }
+        };
+
+        quote! {
+            #vis fn #name() -> std::pin::Pin<Box<dyn std::future::Future<Output = pyo3::PyResult<()>> + Send>> {
+                #sig {
+                    #body
+                }
+
+                #task
+            }
         }
     } else {
         quote! {

pyo3-asyncio-macros/src/tokio.rs

@@ -219,16 +219,23 @@ fn parse_knobs(
 
     let config = config.build()?;
 
-    let mut rt = match config.flavor {
+    let builder = match config.flavor {
         RuntimeFlavor::CurrentThread => quote! {
             pyo3_asyncio::tokio::re_exports::runtime::Builder::new_current_thread()
         },
         RuntimeFlavor::Threaded => quote! {
             pyo3_asyncio::tokio::re_exports::runtime::Builder::new_multi_thread()
         },
     };
+
+    let mut builder_init = quote! {
+        builder.enable_all();
+    };
     if let Some(v) = config.worker_threads {
-        rt = quote! { #rt.worker_threads(#v) };
+        builder_init = quote! {
+            builder.worker_threads(#v);
+            #builder_init;
+        };
     }
 
     let rt_init = match config.flavor {
@@ -247,25 +254,21 @@ fn parse_knobs(
                 #body
             }
 
-            pyo3_asyncio::tokio::init(
-                #rt
-                    .enable_all()
-                    .build()
-                    .unwrap()
-            );
+            pyo3::prepare_freethreaded_python();
+
+            let mut builder = #builder;
+            #builder_init;
+
+            pyo3_asyncio::tokio::init(builder);
 
             #rt_init
 
             pyo3::Python::with_gil(|py| {
-                pyo3_asyncio::with_runtime(py, || {
-                    pyo3_asyncio::tokio::run_until_complete(py, main())?;
-
-                    Ok(())
-                })
-                .map_err(|e| {
-                    e.print_and_set_sys_last_vars(py);
-                })
-                .unwrap();
+                pyo3_asyncio::tokio::run(py, main())
+                    .map_err(|e| {
+                        e.print_and_set_sys_last_vars(py);
+                    })
+                    .unwrap();
             });
         }
     };

pytests/common/mod.rs

@@ -12,7 +12,24 @@ async def sleep_for_1s(sleep_for):
     await sleep_for(1)
 "#;
 
-pub(super) async fn test_into_future() -> PyResult<()> {
+pub(super) async fn test_into_future(event_loop: PyObject) -> PyResult<()> {
+    let fut = Python::with_gil(|py| {
+        let test_mod =
+            PyModule::from_code(py, TEST_MOD, "test_rust_coroutine/test_mod.py", "test_mod")?;
+
+        pyo3_asyncio::into_future_with_loop(
+            event_loop.as_ref(py),
+            test_mod.call_method1("py_sleep", (1.into_py(py),))?,
+        )
+    })?;
+
+    fut.await?;
+
+    Ok(())
+}
+
+#[allow(deprecated)]
+pub(super) async fn test_into_future_0_13() -> PyResult<()> {
     let fut = Python::with_gil(|py| {
         let test_mod =
             PyModule::from_code(py, TEST_MOD, "test_rust_coroutine/test_mod.py", "test_mod")?;
@@ -30,32 +47,24 @@ pub(super) fn test_blocking_sleep() -> PyResult<()> {
     Ok(())
 }
 
-pub(super) async fn test_other_awaitables() -> PyResult<()> {
+pub(super) async fn test_other_awaitables(event_loop: PyObject) -> PyResult<()> {
     let fut = Python::with_gil(|py| {
         let functools = py.import("functools")?;
         let time = py.import("time")?;
 
         // spawn a blocking sleep in the threadpool executor - returns a task, not a coroutine
-        let task = pyo3_asyncio::get_event_loop(py).call_method1(
+        let task = event_loop.as_ref(py).call_method1(
             "run_in_executor",
             (
                 py.None(),
                 functools.call_method1("partial", (time.getattr("sleep")?, 1))?,
             ),
         )?;
 
-        pyo3_asyncio::into_future(task)
+        pyo3_asyncio::into_future_with_loop(event_loop.as_ref(py), task)
     })?;
 
     fut.await?;
 
     Ok(())
 }
-
-pub(super) fn test_init_twice() -> PyResult<()> {
-    // try_init has already been called in test main - ensure a second call doesn't mess the other
-    // tests up
-    Python::with_gil(|py| pyo3_asyncio::try_init(py))?;
-
-    Ok(())
-}