WIP

Author: SemMulder

libs/opsqueue_python/src/common.rs

@@ -443,13 +443,12 @@ pub async fn check_signals_in_background() -> FatalPythonException {
 
 /// Sets up a Tokio runtime to use for a client.
 ///
-/// Rather than the current-thread scheduler,
-/// we use a (single extra!) background thread,
-/// allowing us to keep (GIL-less) tasks alive in the background
-/// even when returning back to Python
+/// We use a current_thread runtime since the Python GIL
+/// together with a multithreaded runtime and pyo3_log is a dangerous combination:
+///  - https://docs.rs/pyo3/latest/pyo3/marker/struct.Python.html#deadlocks
+///  - https://docs.rs/pyo3-log/latest/pyo3_log/#interaction-with-python-gil
 pub fn start_runtime() -> Arc<tokio::runtime::Runtime> {
-    let runtime = tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(1)
+    let runtime = tokio::runtime::Builder::new_current_thread()
         .enable_all()
         .build()
         .expect("Failed to create Tokio runtime in opsqueue client");

libs/opsqueue_python/src/producer.rs

@@ -217,17 +217,19 @@ impl ProducerClient {
         py.allow_threads(|| {
             let prefix = uuid::Uuid::now_v7().to_string();
             tracing::debug!("Uploading submission chunks to object store subfolder {prefix}...");
-            let chunk_count = Python::with_gil(|py| {
-                self.block_unless_interrupted(async {
-                    let chunk_contents = chunk_contents.bind(py);
-                    let stream = futures::stream::iter(chunk_contents)
-                        .map(|item| item.and_then(|item| item.extract()).map_err(Into::into));
+            let chunk_count = self.block_unless_interrupted(async {
+                    let chunk_contents = std::iter::from_fn(move || {
+                        Python::with_gil(|py|
+                            chunk_contents.bind(py).clone().next()
+                                .map(|item| item.and_then(
+                                    |item| item.extract()).map_err(Into::into)))
+                    });
+                    let stream = futures::stream::iter(chunk_contents);
                     self.object_store_client
                         .store_chunks(&prefix, ChunkType::Input, stream)
                         .await
                         .map_err(|e| CError(R(L(e))))
-                })
-            })?;
+                })?;
             let chunk_count = chunk::ChunkIndex::from(chunk_count);
             tracing::debug!("Finished uploading to object store. {prefix} contains {chunk_count} chunks");
 

opsqueue/src/consumer/client.rs

@@ -2,6 +2,7 @@ use std::{
     collections::HashMap,
     str::FromStr,
     sync::{atomic::AtomicBool, Arc},
+    thread,
     time::Duration,
 };
 
@@ -200,14 +201,25 @@ impl Client {
         }
 
         let healthy = Arc::new(AtomicBool::new(true));
-        tokio::spawn(Self::background_task(
-            cancellation_token.clone(),
-            healthy.clone(),
-            in_flight_requests.clone(),
-            ws_stream,
-            ws_sink.clone(),
-            config,
-        ));
+
+        let bg_cancellation_token = cancellation_token.clone();
+        let bg_healthy = healthy.clone();
+        let bg_in_flight_requests = in_flight_requests.clone();
+        let bg_ws_sink = ws_sink.clone();
+        thread::spawn(move || {
+            tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("Failed to create Tokio runtime for background client")
+                .block_on(Self::background_task(
+                    bg_cancellation_token,
+                    bg_healthy,
+                    bg_in_flight_requests,
+                    ws_stream,
+                    bg_ws_sink,
+                    config,
+                ))
+        });
 
         let me = Self {
             in_flight_requests,