diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock index dd41cf0e84b..33f6bb5a87e 100644 --- a/litellm-rust/Cargo.lock +++ b/litellm-rust/Cargo.lock @@ -1442,13 +1442,16 @@ name = "litellm-python-bridge" version = "0.1.0" dependencies = [ "criterion", + "futures-util", "litellm-ai-gateway", "litellm-core", "litellm-python-interop", "pyo3", "pyo3-async-runtimes", + "serde", "serde_json", "tokio", + "tokio-tungstenite", ] [[package]] @@ -1669,9 +1672,9 @@ dependencies = [ [[package]] name = "pyo3" -version = "0.29.0" +version = "0.29.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cd274650b21d4bfc26a0a47587962c1edb425f69287324355cd040c3ea66071c" +checksum = "4688ddedf473e32662b9b067670129a8afb8c18e351482c70d62ba4a88171e8b" dependencies = [ "libc", "once_cell", @@ -1697,18 +1700,18 @@ dependencies = [ [[package]] name = "pyo3-build-config" -version = "0.29.0" +version = "0.29.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c5e2a7d2f0d013342f295c048ad19237add5154a55b1c5a254c0ec93d4109078" +checksum = "f41027e41b4bd03f6e60f9f417fe24a6341a6bb744edd62b6f709f2a52ea30e9" dependencies = [ "target-lexicon", ] [[package]] name = "pyo3-ffi" -version = "0.29.0" +version = "0.29.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca85c467da1bbc8d866eea5deff9cf29ea5f7785054a17da36e65bda9c05845b" +checksum = "e591a95526fead067432c3b3a33fc74770b87b1e04e73671090d9c2055a2b327" dependencies = [ "libc", "pyo3-build-config", @@ -1716,9 +1719,9 @@ dependencies = [ [[package]] name = "pyo3-macros" -version = "0.29.0" +version = "0.29.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ac53762fd065daa3194dd09337a38bd793a188100fd1a9304c4ab312d901771" +checksum = "73225868fc1cd84eef2c3c230ddb91273bf1de46aeb8a4248da76d32a0924a1c" dependencies = [ "proc-macro2", "pyo3-macros-backend", @@ -1728,9 +1731,9 @@ dependencies = [ [[package]] name = "pyo3-macros-backend" -version = "0.29.0" +version = "0.29.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4ca3a1557399783172dc5bf39cfca835157732532cba56b71d2292161e53b362" +checksum = "571575aa3749fa6216757dd47d2a3e7ef360f329a40f0666a9fbd14889024952" dependencies = [ "heck", "proc-macro2", diff --git a/litellm-rust/Cargo.toml b/litellm-rust/Cargo.toml index c447d915abe..f99d7b47918 100644 --- a/litellm-rust/Cargo.toml +++ b/litellm-rust/Cargo.toml @@ -18,7 +18,7 @@ litellm-core = { path = "crates/core" } litellm-ai-gateway = { path = "crates/ai-gateway", default-features = false } litellm-python-interop = { path = "crates/python-interop" } axum = "0.7" -pyo3 = "0.29.0" +pyo3 = "0.29.2" pyo3-async-runtimes = { version = "0.29.0", features = ["tokio-runtime"] } pythonize = "0.29.0" rand = "0.8" diff --git a/litellm-rust/crates/python-bridge/Cargo.toml b/litellm-rust/crates/python-bridge/Cargo.toml index 498003de149..e89f9cd9aa9 100644 --- a/litellm-rust/crates/python-bridge/Cargo.toml +++ b/litellm-rust/crates/python-bridge/Cargo.toml @@ -16,16 +16,19 @@ extension-module = ["pyo3/extension-module"] panic-test = [] [dependencies] +futures-util.workspace = true litellm-core = { workspace = true, features = ["bedrock-auth"] } litellm-ai-gateway = { workspace = true, default-features = false } litellm-python-interop.workspace = true pyo3.workspace = true pyo3-async-runtimes.workspace = true +serde.workspace = true serde_json.workspace = true tokio.workspace = true [dev-dependencies] criterion = "0.8.2" +tokio-tungstenite.workspace = true [[bench]] name = "serialization" diff --git a/litellm-rust/crates/python-bridge/src/lib.rs b/litellm-rust/crates/python-bridge/src/lib.rs index f628c987220..47df6b905a8 100644 --- a/litellm-rust/crates/python-bridge/src/lib.rs +++ b/litellm-rust/crates/python-bridge/src/lib.rs @@ -6,6 +6,7 @@ mod routes; use litellm_ai_gateway::io::responses_ws::ResponsesWebSocketConnection as RustResponsesWebSocketConnection; use pyo3::prelude::*; use pyo3::types::PyAny; +use serde_json::Value; use crate::errors::core_error_to_pyerr; use crate::marshal::{marshal_headers, optional_timeout}; @@ -23,16 +24,16 @@ impl ResponsesWebSocketConnection { _cls: &Bound<'py, pyo3::types::PyType>, py: Python<'py>, url: String, - headers: Option>, + #[pyo3(from_py_with = litellm_python_interop::from_py)] headers: Option, timeout_seconds: Option, ) -> PyResult> { - let headers = marshal_headers(py, headers)?; + let headers = marshal_headers(headers)?; let timeout = optional_timeout(timeout_seconds); pyo3_async_runtimes::tokio::future_into_py(py, async move { let inner = RustResponsesWebSocketConnection::connect_url(&url, &headers, timeout) .await .map_err(core_error_to_pyerr)?; - Python::attach(|py| Py::new(py, ResponsesWebSocketConnection { inner })) + Ok(ResponsesWebSocketConnection { inner }) }) } @@ -58,24 +59,36 @@ impl ResponsesWebSocketConnection { } } -#[pymodule] -fn _native(module: &Bound<'_, PyModule>) -> PyResult<()> { - errors::register(module)?; - routes::register(module)?; - module.add_class::()?; - diagnostics::register(module) +#[pymodule(gil_used = false)] +mod _native { + use pyo3::prelude::*; + + #[pymodule_init] + fn init(module: &Bound<'_, PyModule>) -> PyResult<()> { + super::errors::register(module)?; + super::routes::register(module)?; + module.add_class::()?; + super::diagnostics::register(module) + } } #[cfg(test)] mod tests { + use std::ffi::CString; + use std::time::Duration; + + use futures_util::{SinkExt, StreamExt}; + use pyo3::types::PyDict; + use tokio::net::TcpListener; + use tokio_tungstenite::{accept_async, tungstenite::Message}; + use super::*; #[test] fn module_registration_preserves_the_public_surface() { Python::initialize(); Python::attach(|py| { - let module = PyModule::new(py, "_native").expect("module should be created"); - _native(&module).expect("module should register"); + let module = pyo3::wrap_pymodule!(_native)(py).into_bound(py); let expected = [ "RustBridgeDeclined", @@ -93,13 +106,79 @@ mod tests { "gil_stats", ]; - for name in expected { - assert!( - module - .hasattr(name) - .expect("attribute lookup should succeed") - ); - } + let public_names: Vec = module + .dict() + .keys() + .extract::>() + .expect("module names should be strings") + .into_iter() + .filter(|name| !name.starts_with("__")) + .collect(); + assert_eq!(public_names, expected); }); } + + #[test] + fn responses_websocket_connection_round_trips_through_python() { + Python::initialize(); + let runtime = pyo3_async_runtimes::tokio::get_runtime(); + let listener = runtime + .block_on(TcpListener::bind("127.0.0.1:0")) + .expect("listener should bind"); + let address = listener + .local_addr() + .expect("listener should have an address"); + let server = runtime.spawn(async move { + let (stream, _) = listener.accept().await.expect("server should accept"); + let mut socket = accept_async(stream) + .await + .expect("handshake should succeed"); + + let message = socket + .next() + .await + .expect("client should send a frame") + .expect("client frame should be valid"); + assert_eq!(message, Message::Text("from-python".into())); + socket + .send(Message::Text("from-server".into())) + .await + .expect("server should reply"); + assert!(matches!(socket.next().await, Some(Ok(Message::Close(_))))); + }); + + Python::attach(|py| { + let module = pyo3::wrap_pymodule!(_native)(py).into_bound(py); + let locals = PyDict::new(py); + locals + .set_item("native", &module) + .expect("module should enter Python locals"); + locals + .set_item("url", format!("ws://{address}")) + .expect("URL should enter Python locals"); + let code = CString::new( + r#" +import asyncio + +async def exercise(): + connection = await native.ResponsesWebSocketConnection.connect(url) + assert type(connection) is native.ResponsesWebSocketConnection + await connection.send_text("from-python") + assert await connection.recv_text() == "from-server" + await connection.close() + assert await connection.recv_text() is None + +asyncio.run(asyncio.wait_for(exercise(), timeout=5)) +"#, + ) + .expect("Python source should not contain null bytes"); + py.run(&code, Some(&locals), Some(&locals)) + .expect("Python WebSocket methods should round trip"); + }); + + runtime + .block_on(async { tokio::time::timeout(Duration::from_secs(5), server).await }) + .expect("server should finish") + .expect("server task should not panic"); + } } diff --git a/litellm-rust/crates/python-bridge/src/marshal.rs b/litellm-rust/crates/python-bridge/src/marshal.rs index 6c070d2a9ee..724514c6bab 100644 --- a/litellm-rust/crates/python-bridge/src/marshal.rs +++ b/litellm-rust/crates/python-bridge/src/marshal.rs @@ -30,12 +30,9 @@ pub(crate) fn optional_timeout(timeout_seconds: Option) -> Option }) } -pub(crate) fn marshal_headers( - py: Python<'_>, - headers: Option>, -) -> PyResult> { +pub(crate) fn marshal_headers(headers: Option) -> PyResult> { let value = match headers { - Some(headers) => from_py(headers.bind(py))?, + Some(headers) => headers, None => Value::Object(Map::new()), }; let Value::Object(headers) = value else { diff --git a/litellm-rust/crates/python-bridge/src/routes/audio_transcription.rs b/litellm-rust/crates/python-bridge/src/routes/audio_transcription.rs index ec0dfc75df2..d258ab7f1d8 100644 --- a/litellm-rust/crates/python-bridge/src/routes/audio_transcription.rs +++ b/litellm-rust/crates/python-bridge/src/routes/audio_transcription.rs @@ -1,12 +1,14 @@ use litellm_ai_gateway::io::audio_transcription::{ AudioTranscriptionRequest, audio_transcription as run_audio_transcription, }; -use litellm_python_interop::{from_py, release_gil, to_py}; +use litellm_python_interop::from_py; use pyo3::prelude::*; use crate::errors::core_error_to_pyerr; use crate::marshal::{optional_object_to_map, optional_timeout}; +use super::runtime::{run_async, run_sync}; + #[pyfunction] #[pyo3(signature = (model, audio, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, optional_params=None, timeout_seconds=None))] #[allow(clippy::too_many_arguments)] @@ -28,9 +30,10 @@ fn transcription( }; let optional_params = optional_object_to_map(py, "optional_params", optional_params)?; let timeout = optional_timeout(timeout_seconds); - let result = release_gil(py, || { - pyo3_async_runtimes::tokio::get_runtime().block_on(run_audio_transcription( - AudioTranscriptionRequest { + run_sync( + py, + async move { + run_audio_transcription(AudioTranscriptionRequest { model: &model, audio, api_key: api_key.as_deref(), @@ -43,13 +46,11 @@ fn transcription( guardrails: Vec::new(), request_metadata: Default::default(), litellm_call_id: None, - }, - )) - }); - match result { - Ok(value) => to_py(py, &value), - Err(err) => Err(core_error_to_pyerr(err)), - } + }) + .await + }, + core_error_to_pyerr, + ) } #[pyfunction] @@ -73,25 +74,27 @@ fn atranscription( }; let optional_params = optional_object_to_map(py, "optional_params", optional_params)?; let timeout = optional_timeout(timeout_seconds); - pyo3_async_runtimes::tokio::future_into_py(py, async move { - let value = run_audio_transcription(AudioTranscriptionRequest { - model: &model, - audio, - api_key: api_key.as_deref(), - api_base: api_base.as_deref(), - custom_llm_provider: custom_llm_provider.as_deref(), - extra_headers, - optional_params, - timeout, - callbacks: Vec::new(), - guardrails: Vec::new(), - request_metadata: Default::default(), - litellm_call_id: None, - }) - .await - .map_err(core_error_to_pyerr)?; - Python::attach(|py| to_py(py, &value)) - }) + run_async( + py, + async move { + run_audio_transcription(AudioTranscriptionRequest { + model: &model, + audio, + api_key: api_key.as_deref(), + api_base: api_base.as_deref(), + custom_llm_provider: custom_llm_provider.as_deref(), + extra_headers, + optional_params, + timeout, + callbacks: Vec::new(), + guardrails: Vec::new(), + request_metadata: Default::default(), + litellm_call_id: None, + }) + .await + }, + core_error_to_pyerr, + ) } pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { diff --git a/litellm-rust/crates/python-bridge/src/routes/chat_completions.rs b/litellm-rust/crates/python-bridge/src/routes/chat_completions.rs index e1f5445f4bc..094374011ec 100644 --- a/litellm-rust/crates/python-bridge/src/routes/chat_completions.rs +++ b/litellm-rust/crates/python-bridge/src/routes/chat_completions.rs @@ -1,10 +1,10 @@ use std::time::Duration; -use litellm_core::chat_completions::types::{ChatCompletionsRequest, ChatCompletionsResponse}; +use litellm_core::chat_completions::types::ChatCompletionsRequest; use litellm_core::chat_completions::{ chat_completions as run_chat_completions, chat_completions_decline_reason, }; -use litellm_python_interop::{from_py, release_gil, to_py}; +use litellm_python_interop::from_py; use pyo3::exceptions::PyValueError; use pyo3::prelude::*; use serde_json::{Map, Value}; @@ -12,12 +12,7 @@ use serde_json::{Map, Value}; use crate::errors::fallback_route_error_to_pyerr; use crate::marshal::{optional_object_to_map, optional_timeout}; -fn chat_completions_response_to_py( - py: Python<'_>, - response: ChatCompletionsResponse, -) -> PyResult> { - to_py(py, &response) -} +use super::runtime::{run_async, run_sync}; type MarshaledChatCompletionsInputs = ( Value, @@ -95,9 +90,10 @@ fn chat_completions( timeout_seconds, )?; - let result = release_gil(py, || { - pyo3_async_runtimes::tokio::get_runtime().block_on(run_chat_completions( - ChatCompletionsRequest { + run_sync( + py, + async move { + run_chat_completions(ChatCompletionsRequest { model: &model, messages, optional_params, @@ -106,14 +102,11 @@ fn chat_completions( custom_llm_provider: custom_llm_provider.as_deref(), extra_headers, timeout, - }, - )) - }); - - match result { - Ok(response) => chat_completions_response_to_py(py, response), - Err(err) => Err(fallback_route_error_to_pyerr(err)), - } + }) + .await + }, + fallback_route_error_to_pyerr, + ) } #[pyfunction] @@ -138,22 +131,23 @@ fn achat_completions( timeout_seconds, )?; - pyo3_async_runtimes::tokio::future_into_py(py, async move { - let response = run_chat_completions(ChatCompletionsRequest { - model: &model, - messages, - optional_params, - api_key: api_key.as_deref(), - api_base: api_base.as_deref(), - custom_llm_provider: custom_llm_provider.as_deref(), - extra_headers, - timeout, - }) - .await - .map_err(fallback_route_error_to_pyerr)?; - - Python::attach(|py| chat_completions_response_to_py(py, response)) - }) + run_async( + py, + async move { + run_chat_completions(ChatCompletionsRequest { + model: &model, + messages, + optional_params, + api_key: api_key.as_deref(), + api_base: api_base.as_deref(), + custom_llm_provider: custom_llm_provider.as_deref(), + extra_headers, + timeout, + }) + .await + }, + fallback_route_error_to_pyerr, + ) } pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { diff --git a/litellm-rust/crates/python-bridge/src/routes/messages.rs b/litellm-rust/crates/python-bridge/src/routes/messages.rs index c337054e54d..2a4c882aace 100644 --- a/litellm-rust/crates/python-bridge/src/routes/messages.rs +++ b/litellm-rust/crates/python-bridge/src/routes/messages.rs @@ -1,8 +1,8 @@ use std::time::Duration; use litellm_core::messages::messages as run_messages; -use litellm_core::messages::types::{AnthropicMessagesResponse, MessagesRequest}; -use litellm_python_interop::{from_py, release_gil, to_py}; +use litellm_core::messages::types::MessagesRequest; +use litellm_python_interop::from_py; use pyo3::exceptions::PyValueError; use pyo3::prelude::*; use serde_json::{Map, Value}; @@ -10,12 +10,7 @@ use serde_json::{Map, Value}; use crate::errors::fallback_route_error_to_pyerr; use crate::marshal::{optional_object_to_map, optional_timeout}; -fn messages_response_to_py( - py: Python<'_>, - response: AnthropicMessagesResponse, -) -> PyResult> { - to_py(py, &response) -} +use super::runtime::{run_async, run_sync}; type MarshaledMessagesInputs = (Value, Option>, Option); @@ -52,22 +47,22 @@ fn messages( let (body, extra_headers, timeout) = marshal_messages_inputs(py, body, extra_headers, timeout_seconds)?; - let result = release_gil(py, || { - pyo3_async_runtimes::tokio::get_runtime().block_on(run_messages(MessagesRequest { - model: &model, - body, - api_key: api_key.as_deref(), - api_base: api_base.as_deref(), - custom_llm_provider: custom_llm_provider.as_deref(), - extra_headers, - timeout, - })) - }); - - match result { - Ok(response) => messages_response_to_py(py, response), - Err(err) => Err(fallback_route_error_to_pyerr(err)), - } + run_sync( + py, + async move { + run_messages(MessagesRequest { + model: &model, + body, + api_key: api_key.as_deref(), + api_base: api_base.as_deref(), + custom_llm_provider: custom_llm_provider.as_deref(), + extra_headers, + timeout, + }) + .await + }, + fallback_route_error_to_pyerr, + ) } #[pyfunction] @@ -86,21 +81,22 @@ fn amessages( let (body, extra_headers, timeout) = marshal_messages_inputs(py, body, extra_headers, timeout_seconds)?; - pyo3_async_runtimes::tokio::future_into_py(py, async move { - let response = run_messages(MessagesRequest { - model: &model, - body, - api_key: api_key.as_deref(), - api_base: api_base.as_deref(), - custom_llm_provider: custom_llm_provider.as_deref(), - extra_headers, - timeout, - }) - .await - .map_err(fallback_route_error_to_pyerr)?; - - Python::attach(|py| messages_response_to_py(py, response)) - }) + run_async( + py, + async move { + run_messages(MessagesRequest { + model: &model, + body, + api_key: api_key.as_deref(), + api_base: api_base.as_deref(), + custom_llm_provider: custom_llm_provider.as_deref(), + extra_headers, + timeout, + }) + .await + }, + fallback_route_error_to_pyerr, + ) } pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { diff --git a/litellm-rust/crates/python-bridge/src/routes/mod.rs b/litellm-rust/crates/python-bridge/src/routes/mod.rs index a2eb8355767..46724a39bc0 100644 --- a/litellm-rust/crates/python-bridge/src/routes/mod.rs +++ b/litellm-rust/crates/python-bridge/src/routes/mod.rs @@ -4,6 +4,7 @@ mod audio_transcription; mod chat_completions; mod messages; mod ocr; +mod runtime; pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { ocr::register(module)?; diff --git a/litellm-rust/crates/python-bridge/src/routes/ocr.rs b/litellm-rust/crates/python-bridge/src/routes/ocr.rs index 047bf245d7c..0f6c7577722 100644 --- a/litellm-rust/crates/python-bridge/src/routes/ocr.rs +++ b/litellm-rust/crates/python-bridge/src/routes/ocr.rs @@ -1,13 +1,15 @@ use std::time::Duration; use litellm_ai_gateway::io::ocr::{OcrRequest, ocr as run_ocr}; -use litellm_python_interop::{from_py, release_gil, to_py}; +use litellm_python_interop::from_py; use pyo3::prelude::*; use serde_json::{Map, Value}; use crate::errors::core_error_to_pyerr; use crate::marshal::{optional_object_to_map, optional_timeout}; +use super::runtime::{run_async, run_sync}; + type MarshaledOcrInputs = ( Value, Option>, @@ -55,27 +57,27 @@ fn ocr( timeout_seconds, )?; - let result = release_gil(py, || { - pyo3_async_runtimes::tokio::get_runtime().block_on(run_ocr(OcrRequest { - model: &model, - document, - api_key: api_key.as_deref(), - api_base: api_base.as_deref(), - custom_llm_provider: custom_llm_provider.as_deref(), - extra_headers, - optional_params, - timeout, - callbacks: Vec::new(), - guardrails: Vec::new(), - request_metadata: Default::default(), - litellm_call_id: None, - })) - }); - - match result { - Ok(value) => to_py(py, &value), - Err(err) => Err(core_error_to_pyerr(err)), - } + run_sync( + py, + async move { + run_ocr(OcrRequest { + model: &model, + document, + api_key: api_key.as_deref(), + api_base: api_base.as_deref(), + custom_llm_provider: custom_llm_provider.as_deref(), + extra_headers, + optional_params, + timeout, + callbacks: Vec::new(), + guardrails: Vec::new(), + request_metadata: Default::default(), + litellm_call_id: None, + }) + .await + }, + core_error_to_pyerr, + ) } #[pyfunction] @@ -100,26 +102,27 @@ fn aocr( timeout_seconds, )?; - pyo3_async_runtimes::tokio::future_into_py(py, async move { - let value = run_ocr(OcrRequest { - model: &model, - document, - api_key: api_key.as_deref(), - api_base: api_base.as_deref(), - custom_llm_provider: custom_llm_provider.as_deref(), - extra_headers, - optional_params, - timeout, - callbacks: Vec::new(), - guardrails: Vec::new(), - request_metadata: Default::default(), - litellm_call_id: None, - }) - .await - .map_err(core_error_to_pyerr)?; - - Python::attach(|py| to_py(py, &value)) - }) + run_async( + py, + async move { + run_ocr(OcrRequest { + model: &model, + document, + api_key: api_key.as_deref(), + api_base: api_base.as_deref(), + custom_llm_provider: custom_llm_provider.as_deref(), + extra_headers, + optional_params, + timeout, + callbacks: Vec::new(), + guardrails: Vec::new(), + request_metadata: Default::default(), + litellm_call_id: None, + }) + .await + }, + core_error_to_pyerr, + ) } pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { diff --git a/litellm-rust/crates/python-bridge/src/routes/runtime.rs b/litellm-rust/crates/python-bridge/src/routes/runtime.rs new file mode 100644 index 00000000000..491d3e5df4b --- /dev/null +++ b/litellm-rust/crates/python-bridge/src/routes/runtime.rs @@ -0,0 +1,423 @@ +use std::future::Future; +use std::panic::AssertUnwindSafe; +use std::time::Duration; + +use futures_util::FutureExt; +use litellm_core::error::{CoreError, CoreResult}; +use litellm_python_interop::{Pythonized, panic_to_pyerr, release_gil}; +use pyo3::exceptions::PyRuntimeError; +use pyo3::prelude::*; +use serde::Serialize; +use tokio::runtime::{Handle, Runtime}; +use tokio::time::{self, MissedTickBehavior}; + +pub(super) fn run_sync( + py: Python<'_>, + future: F, + map_error: fn(CoreError) -> PyErr, +) -> PyResult> +where + T: Serialize + Send + 'static, + F: Future> + Send + 'static, +{ + run_sync_on( + py, + pyo3_async_runtimes::tokio::get_runtime(), + future, + map_error, + ) +} + +fn run_sync_on( + py: Python<'_>, + runtime: &Runtime, + future: F, + map_error: fn(CoreError) -> PyErr, +) -> PyResult> +where + T: Serialize + Send + 'static, + F: Future> + Send + 'static, +{ + if Handle::try_current().is_ok() { + return Err(PyRuntimeError::new_err( + "synchronous native routes cannot run from a Tokio context; use the async route", + )); + } + + let result = release_gil(py, move || runtime.block_on(wait_for_sync_result(future)))?; + let result = map_core_result(result, map_error)?; + Pythonized(result).into_pyobject(py).map(Bound::unbind) +} + +pub(super) fn run_async( + py: Python<'_>, + future: F, + map_error: fn(CoreError) -> PyErr, +) -> PyResult> +where + T: Serialize + Send + 'static, + F: Future> + Send + 'static, +{ + pyo3_async_runtimes::tokio::future_into_py(py, async move { + let result = catch_route_panic(future).await?; + let result = map_core_result(result, map_error)?; + Ok(Pythonized(result)) + }) +} + +fn map_core_result(result: CoreResult, map_error: fn(CoreError) -> PyErr) -> PyResult { + match result { + Ok(value) => Ok(value), + Err(error) => Err( + std::panic::catch_unwind(AssertUnwindSafe(|| map_error(error))) + .map_err(panic_to_pyerr)?, + ), + } +} + +async fn catch_route_panic(future: F) -> PyResult> +where + F: Future>, +{ + AssertUnwindSafe(future) + .catch_unwind() + .await + .map_err(panic_to_pyerr) +} + +async fn wait_for_sync_result(future: F) -> PyResult> +where + F: Future>, +{ + let future = catch_route_panic(future); + tokio::pin!(future); + + let signal_interval = Duration::from_millis(50); + let mut signal_checks = + time::interval_at(time::Instant::now() + signal_interval, signal_interval); + signal_checks.set_missed_tick_behavior(MissedTickBehavior::Delay); + loop { + tokio::select! { + result = &mut future => return result, + _ = signal_checks.tick() => Python::attach(|py| py.check_signals())?, + } + } +} + +#[cfg(test)] +mod tests { + use std::ffi::CString; + use std::future::poll_fn; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::{Arc, mpsc}; + use std::task::Poll; + use std::thread; + use std::time::Instant; + + use pyo3::panic::PanicException; + use pyo3::types::{PyDict, PyModule}; + use serde::Serializer; + use tokio::runtime::Builder; + + use super::*; + + fn runtime_error(error: CoreError) -> PyErr { + PyRuntimeError::new_err(error.to_string()) + } + + fn panicking_error_mapper(_error: CoreError) -> PyErr { + panic!("error mapper panicked") + } + + struct PanickingOutput; + + static ASYNC_PROBE_COMPLETED: AtomicUsize = AtomicUsize::new(0); + + impl Serialize for PanickingOutput { + fn serialize(&self, _serializer: S) -> Result + where + S: Serializer, + { + panic!("serializer panicked") + } + } + + #[pyfunction] + fn async_serialization_panic(py: Python<'_>) -> PyResult> { + run_async(py, async { Ok(PanickingOutput) }, runtime_error) + } + + #[pyfunction] + fn async_runtime_probe(py: Python<'_>) -> PyResult> { + run_async( + py, + async { + ASYNC_PROBE_COMPLETED.fetch_add(1, Ordering::SeqCst); + Ok(true) + }, + runtime_error, + ) + } + + #[pyfunction] + fn runtime_worker_count() -> usize { + pyo3_async_runtimes::tokio::get_runtime() + .metrics() + .num_workers() + } + + #[pyfunction] + fn runtime_is_responsive(_py: Python<'_>, expected_completions: usize) -> bool { + let completion_deadline = Instant::now() + Duration::from_secs(2); + while ASYNC_PROBE_COMPLETED.load(Ordering::SeqCst) < expected_completions { + if Instant::now() >= completion_deadline { + return false; + } + thread::sleep(Duration::from_millis(1)); + } + + let (heartbeat_tx, heartbeat_rx) = mpsc::sync_channel(1); + pyo3_async_runtimes::tokio::get_runtime().spawn(async move { + let _ = heartbeat_tx.send(()); + }); + heartbeat_rx.recv_timeout(Duration::from_secs(2)).is_ok() + } + + fn extract_bool(py: Python<'_>, result: PyResult>) -> bool { + result + .expect("route should complete") + .bind(py) + .extract() + .expect("result should convert") + } + + #[test] + fn sync_runner_polls_future_on_the_caller_thread() { + Python::initialize(); + Python::attach(|py| { + let caller_thread = std::thread::current().id(); + let result = run_sync( + py, + async move { Ok(std::thread::current().id() == caller_thread) }, + runtime_error, + ); + + assert!(extract_bool(py, result)); + }); + } + + #[test] + fn sync_runner_releases_gil_while_waiting() { + Python::initialize(); + Python::attach(|py| { + let result = run_sync( + py, + async { + let gil_acquired = tokio::time::timeout( + Duration::from_secs(2), + tokio::task::spawn_blocking(|| Python::attach(|_| true)), + ) + .await; + Ok(matches!(gil_acquired, Ok(Ok(true)))) + }, + runtime_error, + ); + + assert!(extract_bool(py, result)); + }); + } + + #[test] + fn sync_runner_rejects_calls_from_a_tokio_context() { + Python::initialize(); + let runtime = Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime should build"); + + let error = runtime.block_on(async { + Python::attach(|py| { + run_sync::(py, async { Ok(true) }, runtime_error) + .expect_err("sync route should reject a nested Tokio runtime") + }) + }); + + assert_eq!( + error.to_string(), + "RuntimeError: synchronous native routes cannot run from a Tokio context; use the async route" + ); + } + + #[test] + fn sync_runner_can_drive_a_current_thread_runtime() { + Python::initialize(); + let runtime = Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime should build"); + Python::attach(|py| { + let result = run_sync_on( + py, + &runtime, + async { + tokio::task::yield_now().await; + Ok(true) + }, + runtime_error, + ); + assert!(extract_bool(py, result)); + }); + } + + #[test] + fn sync_runner_maps_a_panicked_future() { + Python::initialize(); + Python::attach(|py| { + let error = run_sync::( + py, + poll_fn(|_| -> Poll> { panic!("route future panicked") }), + runtime_error, + ) + .expect_err("panicked route should become a Python exception"); + + assert!(error.is_instance_of::(py)); + assert_eq!(error.to_string(), "PanicException: route future panicked"); + }); + } + + #[test] + fn sync_runner_maps_a_panicked_error_mapper() { + Python::initialize(); + Python::attach(|py| { + let error = run_sync::( + py, + async { Err(CoreError::InvalidRequest("invalid".to_string())) }, + panicking_error_mapper, + ) + .expect_err("panicked mapper should become a Python exception"); + + assert!(error.is_instance_of::(py)); + assert_eq!(error.to_string(), "PanicException: error mapper panicked"); + }); + } + + #[test] + fn sync_runner_surfaces_serializer_panics() { + Python::initialize(); + Python::attach(|py| { + let error = run_sync(py, async { Ok(PanickingOutput) }, runtime_error) + .expect_err("serializer panic should become a Python exception"); + + assert!(error.is_instance_of::(py)); + assert_eq!(error.to_string(), "PanicException: serializer panicked"); + }); + } + + #[test] + fn sync_runner_supports_concurrent_callers_on_the_shared_runtime() { + Python::initialize(); + let barrier = Arc::new(tokio::sync::Barrier::new(2)); + let callers: Vec<_> = (0..2) + .map(|_| { + let barrier = Arc::clone(&barrier); + thread::spawn(move || { + Python::attach(|py| { + extract_bool( + py, + run_sync( + py, + async move { + Ok(tokio::time::timeout(Duration::from_secs(2), barrier.wait()) + .await + .is_ok()) + }, + runtime_error, + ), + ) + }) + }) + }) + .collect(); + let results: Vec<_> = callers + .into_iter() + .map(|caller| caller.join().expect("caller should not panic")) + .collect(); + + assert_eq!(results, vec![true, true]); + } + + #[test] + fn async_runner_surfaces_serializer_panics() { + Python::initialize(); + Python::attach(|py| { + let module = PyModule::new(py, "runtime").expect("module should be created"); + module + .add_function( + wrap_pyfunction!(async_serialization_panic, &module) + .expect("function should wrap"), + ) + .expect("function should register"); + let locals = PyDict::new(py); + locals + .set_item("runtime", &module) + .expect("module should enter Python locals"); + let code = CString::new( + r#" +import asyncio + +async def exercise(): + try: + await runtime.async_serialization_panic() + except BaseException as error: + assert type(error).__name__ == "PanicException" + assert str(error) == "serializer panicked" + else: + raise AssertionError("serializer panic was not raised") + +asyncio.run(exercise()) +"#, + ) + .expect("Python source should not contain null bytes"); + py.run(&code, Some(&locals), Some(&locals)) + .expect("serializer panic should reach the Python awaiter"); + }); + } + + #[test] + fn async_result_delivery_does_not_stall_tokio_workers() { + Python::initialize(); + ASYNC_PROBE_COMPLETED.store(0, Ordering::SeqCst); + Python::attach(|py| { + let module = PyModule::new(py, "runtime").expect("module should be created"); + for function in [ + wrap_pyfunction!(async_runtime_probe, &module).expect("function should wrap"), + wrap_pyfunction!(runtime_worker_count, &module).expect("function should wrap"), + wrap_pyfunction!(runtime_is_responsive, &module).expect("function should wrap"), + ] { + module + .add_function(function) + .expect("function should register"); + } + let locals = PyDict::new(py); + locals + .set_item("runtime", &module) + .expect("module should enter Python locals"); + let code = CString::new( + r#" +import asyncio + +async def exercise(): + worker_count = runtime.runtime_worker_count() + awaitables = [runtime.async_runtime_probe() for _ in range(worker_count)] + assert runtime.runtime_is_responsive(worker_count) + assert await asyncio.gather(*awaitables) == [True] * worker_count + +asyncio.run(exercise()) +"#, + ) + .expect("Python source should not contain null bytes"); + py.run(&code, Some(&locals), Some(&locals)) + .expect("result delivery should leave Tokio workers responsive"); + }); + } +} diff --git a/litellm-rust/crates/python-interop/src/lib.rs b/litellm-rust/crates/python-interop/src/lib.rs index df2bd260fdb..2e562bdae70 100644 --- a/litellm-rust/crates/python-interop/src/lib.rs +++ b/litellm-rust/crates/python-interop/src/lib.rs @@ -2,4 +2,4 @@ mod gil; mod marshal; pub use gil::{release_count, release_gil}; -pub use marshal::{from_py, to_py}; +pub use marshal::{Pythonized, from_py, panic_to_pyerr, to_py}; diff --git a/litellm-rust/crates/python-interop/src/marshal.rs b/litellm-rust/crates/python-interop/src/marshal.rs index c3d0638427c..a16d1e0ae13 100644 --- a/litellm-rust/crates/python-interop/src/marshal.rs +++ b/litellm-rust/crates/python-interop/src/marshal.rs @@ -1,4 +1,8 @@ +use std::any::Any; +use std::panic::{AssertUnwindSafe, catch_unwind}; + use pyo3::exceptions::PyValueError; +use pyo3::panic::PanicException; use pyo3::prelude::*; use serde::Serialize; use serde::de::DeserializeOwned; @@ -18,3 +22,71 @@ where .map(Bound::unbind) .map_err(|error| PyValueError::new_err(error.to_string())) } + +pub struct Pythonized(pub T); + +impl<'py, T> IntoPyObject<'py> for Pythonized +where + T: Serialize, +{ + type Target = PyAny; + type Output = Bound<'py, PyAny>; + type Error = PyErr; + + fn into_pyobject(self, py: Python<'py>) -> PyResult { + catch_unwind(AssertUnwindSafe(|| pythonize::pythonize(py, &self.0))) + .map_err(panic_to_pyerr)? + .map_err(|error| PyValueError::new_err(error.to_string())) + } +} + +pub fn panic_to_pyerr(payload: Box) -> PyErr { + let message = payload + .downcast_ref::() + .map(String::as_str) + .or_else(|| payload.downcast_ref::<&str>().copied()) + .unwrap_or("panic from Rust code"); + PanicException::new_err(message.to_string()) +} + +#[cfg(test)] +mod tests { + use serde::Serializer; + + use super::*; + + struct PanickingSerializer; + + impl Serialize for PanickingSerializer { + fn serialize(&self, _serializer: S) -> Result + where + S: Serializer, + { + panic!("serializer panicked") + } + } + + #[test] + fn pythonized_converts_on_the_attached_thread() { + Python::initialize(); + Python::attach(|py| { + let value: Vec = Pythonized(vec![1, 2, 3]) + .into_pyobject(py) + .and_then(|value| value.extract()) + .expect("value should convert"); + assert_eq!(value, vec![1, 2, 3]); + }); + } + + #[test] + fn pythonized_maps_serializer_panics_to_a_base_exception() { + Python::initialize(); + Python::attach(|py| { + let error = Pythonized(PanickingSerializer) + .into_pyobject(py) + .expect_err("serializer panic should become a Python exception"); + assert!(error.is_instance_of::(py)); + assert_eq!(error.to_string(), "PanicException: serializer panicked"); + }); + } +}