Improve Linux builds, support venv and uvloop, and use pyo3_async_runtimes

Author: Qard

Cargo.toml

@@ -27,7 +27,7 @@ http-rewriter = { git = "ssh://git@github.com/platformatic/http-rewriter.git" }
 # Default enable napi4 feature, see https://nodejs.org/api/n-api.html#node-api-version-matrix
 napi = { version = "3.0.0-beta.8", default-features = false, features = ["napi4"], optional = true }
 napi-derive = { version = "3.0.0-beta.8", optional = true }
-pyo3 = { version = "0.25.1", features = ["auto-initialize", "experimental-async"] }
+pyo3 = { version = "0.25.1", features = ["experimental-async"] }
 pyo3-async-runtimes = { version = "0.25.0", features = ["tokio-runtime"] }
 thiserror = "2.0.12"
 tokio = { version = "1.45.1", features = ["full"] }

package.json

@@ -1,5 +1,5 @@
 {
-  "name": "python-node",
+  "name": "@platformatic/python-node",
   "description": "Run ASGI-compatible Python apps in Node.js",
   "private": true,
   "version": "0.0.0",

src/asgi/mod.rs

@@ -1,7 +1,7 @@
 use std::{
-  env::current_dir,
+  env::{current_dir, var},
   ffi::CString,
-  fs::read_to_string,
+  fs::{read_dir, read_to_string},
   path::{Path, PathBuf},
 };
 
@@ -13,11 +13,6 @@ unsafe extern "C" {
   fn dlopen(filename: *const i8, flag: i32) -> *mut c_void;
 }
 
-#[cfg(target_os = "linux")]
-const RTLD_GLOBAL: i32 = 0x100;
-#[cfg(target_os = "linux")]
-const RTLD_NOW: i32 = 0x2;
-
 use bytes::BytesMut;
 use http_handler::{Handler, Request, RequestExt, Response, extensions::DocumentRoot};
 use pyo3::exceptions::PyRuntimeError;
@@ -48,6 +43,34 @@ pub use websocket::{
   WebSocketConnectionScope, WebSocketReceiveMessage, WebSocketSendException, WebSocketSendMessage,
 };
 
+/// Find all Python site-packages directories in a virtual environment
+fn find_python_site_packages(venv_path: &Path) -> Vec<PathBuf> {
+  let mut site_packages_paths = Vec::new();
+
+  // Check both lib and lib64 directories
+  for lib_dir in &["lib", "lib64"] {
+    let lib_path = venv_path.join(lib_dir);
+    if let Ok(entries) = read_dir(lib_path) {
+      for entry in entries.flatten() {
+        let entry_path = entry.path();
+        if entry_path.is_dir() {
+          if let Some(dir_name) = entry_path.file_name().and_then(|n| n.to_str()) {
+            // Look for directories matching python3.* pattern
+            if dir_name.starts_with("python3.") {
+              let site_packages = entry_path.join("site-packages");
+              if site_packages.exists() {
+                site_packages_paths.push(site_packages);
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+
+  site_packages_paths
+}
+
 /// Load Python library with RTLD_GLOBAL on Linux to make symbols available
 #[cfg(target_os = "linux")]
 fn ensure_python_symbols_global() {
@@ -80,28 +103,22 @@ fn ensure_python_symbols_global() {
       }
     }
 
-    // Fallback to common library names if find command fails
-    let python_libs = [
-      "libpython3.12.so.1.0\0",
-      "libpython3.11.so.1.0\0",
-      "libpython3.10.so.1.0\0",
-      "libpython3.9.so.1.0\0",
-      "libpython3.8.so.1.0\0",
-    ];
+    const RTLD_GLOBAL: i32 = 0x100;
+    const RTLD_NOW: i32 = 0x2;
 
-    for lib_name in &python_libs {
+    // Fallback to trying common library names if find command fails
+    // Try a range of Python versions (3.9 to 3.100 should cover future versions)
+    for minor in 9..=100 {
+      let lib_name = format!("libpython3.{}.so.1.0\0", minor);
       let handle = dlopen(lib_name.as_ptr() as *const i8, RTLD_NOW | RTLD_GLOBAL);
       if !handle.is_null() {
         // Successfully loaded Python library with RTLD_GLOBAL
-        break;
+        return;
       }
     }
-  }
-}
 
-#[cfg(not(target_os = "linux"))]
-fn ensure_python_symbols_global() {
-  // On non-Linux platforms, this is typically not needed
+    eprintln!("Failed to locate system Python library");
+  }
 }
 
 /// Core ASGI handler that loads and manages a Python ASGI application
@@ -116,7 +133,10 @@ impl Asgi {
     docroot: Option<String>,
     app_target: Option<PythonHandlerTarget>,
   ) -> Result<Self, HandlerError> {
+    pyo3::prepare_freethreaded_python();
+
     // Ensure Python symbols are globally available before initializing
+    #[cfg(target_os = "linux")]
     ensure_python_symbols_global();
 
     // Determine document root
@@ -155,6 +175,29 @@ impl Asgi {
     let module_name = CString::new(target.file.clone()).map_err(HandlerError::StringCovertError)?;
 
     Python::with_gil(|py| -> PyResult<PyObject> {
+      // Set up sys.path with docroot and virtual environment paths
+      let sys = py.import("sys")?;
+      let path = sys.getattr("path")?;
+
+      // Add docroot to sys.path
+      path.call_method1("insert", (0, docroot.to_string_lossy()))?;
+
+      // Check for VIRTUAL_ENV and add virtual environment paths
+      if let Ok(virtual_env) = var("VIRTUAL_ENV") {
+        let venv_path = PathBuf::from(&virtual_env);
+
+        // Dynamically find all Python site-packages directories
+        let site_packages_paths = find_python_site_packages(&venv_path);
+
+        // Add all found site-packages paths to sys.path
+        for site_packages in &site_packages_paths {
+          path.call_method1("insert", (0, site_packages.to_string_lossy()))?;
+        }
+
+        // Also add the virtual environment root
+        path.call_method1("insert", (0, virtual_env))?;
+      }
+
       let module = PyModule::from_code(py, &code, &file_name, &module_name)?;
       Ok(module.getattr(&target.function)?.unbind())
     })
@@ -165,13 +208,51 @@ impl Asgi {
   pub fn docroot(&self) -> &Path {
     &self.docroot
   }
+
+  /// Handle a request synchronously using the pyo3_async_runtimes managed runtime
+  pub fn handle_sync(&self, request: Request) -> Result<Response, HandlerError> {
+    pyo3_async_runtimes::tokio::get_runtime().block_on(self.handle(request))
+  }
+
+  /// Install an event loop for this thread, using uvloop if available
+  pub fn install_loop(&self) -> Result<(), HandlerError> {
+    Python::with_gil(|py| -> PyResult<()> {
+      let asyncio = py.import("asyncio")?;
+
+      // Check if there's already an event loop on this thread
+      let needs_new_loop = match asyncio.call_method0("get_event_loop") {
+        Ok(existing_loop) => {
+          // Check if the existing loop is closed
+          existing_loop.call_method0("is_closed")?.extract::<bool>()?
+        }
+        Err(_) => true, // No event loop exists
+      };
+
+      if needs_new_loop {
+        // Set up event loop for this thread, using uvloop if available
+        let loop_ = if let Ok(uvloop) = py.import("uvloop") {
+          // Install uvloop policy if not already installed
+          let _ = uvloop.call_method0("install");
+          uvloop.call_method0("new_event_loop")?
+        } else {
+          asyncio.call_method0("new_event_loop")?
+        };
+        asyncio.call_method1("set_event_loop", (&loop_,))?;
+      }
+
+      Ok(())
+    }).map_err(HandlerError::PythonError)
+  }
 }
 
 #[async_trait::async_trait]
 impl Handler for Asgi {
   type Error = HandlerError;
 
   async fn handle(&self, request: Request) -> Result<Response, Self::Error> {
+    // Ensure the event loop is installed
+    self.install_loop()?;
+
     // Set document root extension
     let mut request = request;
     request.set_document_root(DocumentRoot {
@@ -237,40 +318,40 @@ impl Handler for Asgi {
 
       // Channel closed without complete response
       if response_started {
-        let _ = response_tx.send(Err("Response started but not completed"));
+        let _ = response_tx.send(Err(HandlerError::ResponseInterrupted));
       } else {
-        let _ = response_tx.send(Err("No response received"));
+        let _ = response_tx.send(Err(HandlerError::NoResponse));
       }
     });
 
     // Execute Python
     let py_func = Python::with_gil(|py| self.app_function.clone_ref(py));
-    let python_result = tokio::task::spawn_blocking(move || {
-      execute_coroutine(py_func, scope, rx_receiver, tx_sender)
-    })
-    .await
-    .map_err(|_| HandlerError::PythonError(PyRuntimeError::new_err("Python task panicked")))?;
-
-    // Check if Python execution had errors
-    if let Err(e) = python_result {
-      return Err(HandlerError::PythonError(e));
-    }
 
-    // If the channel is closed without a response, return an error
-    let response = match response_rx.await {
-      Err(_) => {
-        return Err(HandlerError::PythonError(PyRuntimeError::new_err(
-          "Response channel error",
-        )));
-      }
-      Ok(response) => response,
-    };
+    // Now create the coroutine and convert it to a future
+    let coroutine = Python::with_gil(|py| {
+      let scope_py = scope.into_pyobject(py)?;
+      py_func.call1(py, (scope_py, rx_receiver, tx_sender))
+    })?;
+
+    // TODO: This will block the current thread until the coroutine completes.
+    // We should see if there's a way to execute coroutines concurrently.
+    // Blocking in an async function is not great as tokio will assume the
+    // function should yield control when it's not busy, so we're wasting a
+    // thread here. Likely we should implement `Stream` around a coroutine
+    // wrapper to poll it instead. The `run` is internally running the
+    // `run_until_complete` method, which blocks the current thread until
+    // the coroutine completes.
+    Python::with_gil(|py| {
+      pyo3_async_runtimes::tokio::run(py, async move {
+        Python::with_gil(|py| {
+          pyo3_async_runtimes::tokio::into_future(coroutine.into_bound(py))
+        })?.await
+      })
+    })?;
 
     // If an error was sent through the channel, return it
-    let (status, headers, body) = match response {
-      Ok(res) => res,
-      Err(e) => return Err(HandlerError::PythonError(PyRuntimeError::new_err(e))),
-    };
+    let maybe_response = response_rx.await?;
+    let (status, headers, body) = maybe_response?;
 
     // If we reach here, we have a valid response
     let mut builder = http_handler::response::Builder::new().status(status);
@@ -284,25 +365,3 @@ impl Handler for Asgi {
       .map_err(HandlerError::HttpHandlerError)
   }
 }
-
-// Execute the Python coroutine for the ASGI app via asyncio
-fn execute_coroutine(
-  py_func: PyObject,
-  scope: HttpConnectionScope,
-  rx_receiver: Receiver,
-  tx_sender: Sender,
-) -> PyResult<()> {
-  Python::with_gil(|py| {
-    let scope_py = scope.into_pyobject(py)?;
-    let coroutine = py_func.call1(py, (scope_py, rx_receiver, tx_sender))?;
-
-    let asyncio = py.import("asyncio")?;
-    let loop_ = asyncio.call_method0("new_event_loop")?;
-    asyncio.call_method1("set_event_loop", (&loop_,))?;
-
-    loop_.call_method1("run_until_complete", (coroutine,))?;
-    loop_.call_method0("close")?;
-
-    Ok(())
-  })
-}

src/lib.rs

@@ -14,7 +14,7 @@ use std::sync::Arc;
 #[cfg(feature = "napi-support")]
 use http_handler::napi::{Request as NapiRequest, Response as NapiResponse};
 #[cfg(feature = "napi-support")]
-use http_handler::{Handler, Request, Response};
+use http_handler::{Request, Response};
 #[cfg(feature = "napi-support")]
 #[allow(unused_imports)]
 use http_rewriter::napi::Rewriter;
@@ -23,11 +23,10 @@ use http_rewriter::napi::Rewriter;
 extern crate napi_derive;
 #[cfg(feature = "napi-support")]
 use napi::bindgen_prelude::*;
-#[cfg(feature = "napi-support")]
-use tokio::runtime::Runtime;
 
 mod asgi;
 pub use asgi::Asgi;
+use tokio::sync::oneshot::error::RecvError;
 
 /// The Python module and function for handling requests.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -306,6 +305,18 @@ pub enum HandlerError {
   #[error("Python error: {0}")]
   PythonError(#[from] pyo3::prelude::PyErr),
 
+  /// Error when response channel is closed before sending a response.
+  #[error("No response sent")]
+  NoResponse,
+
+  /// Error when response is interrupted.
+  #[error("Response interrupted")]
+  ResponseInterrupted,
+
+  /// Error when response channel is closed.
+  #[error("Response channel closed")]
+  ResponseChannelClosed(#[from] RecvError),
+
   /// Error when creating an HTTP response fails.
   #[error("Failed to create response: {0}")]
   HttpHandlerError(#[from] http_handler::Error),
@@ -317,18 +328,12 @@ impl Task for PythonRequestTask {
   type Output = Response;
   type JsValue = NapiResponse;
 
-  // Handle the PHP request in the worker thread.
+  // Handle the Python request in the worker thread.
   fn compute(&mut self) -> Result<Self::Output> {
-    // Can't use Handle::current() as this thread won't have a runtime configured.
-    let runtime = Runtime::new().map_err(|err| Error::from_reason(err.to_string()))?;
-
-    runtime.block_on(async {
-      self
-        .asgi
-        .handle(self.request.clone())
-        .await
-        .map_err(|err| Error::from_reason(err.to_string()))
-    })
+    self
+      .asgi
+      .handle_sync(self.request.clone())
+      .map_err(|err| Error::from_reason(err.to_string()))
   }
 
   // Handle converting the PHP response to a JavaScript response in the main thread.