From 00c35955ee7de41f7c87e18b0d8ef185c9026015 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 27 Jul 2026 16:09:03 +0000 Subject: [PATCH] Move per-homeserver Rust state into a RustRuntime object on the HomeServer Previously the tokio runtime was stashed in a hidden attribute on the reactor object, installed lazily by whichever Rust code first needed it, and started via callWhenRunning (which never fires under trial's MemoryReactorClock, hence reactor.run() workarounds in several tests). Instead, introduce: * `Reactor`, a typed Rust wrapper around the Twisted reactor. The duck type is validated once at the FFI boundary, and the wrapper is the single place naming the Twisted API surface Rust relies on. * `RustRuntime`, a frozen pyclass wrapping `Arc`, constructed once per homeserver via `hs.get_rust_runtime()`. The tokio runtime starts lazily on first use; shutdown is driven by a reactor shutdown trigger holding only a `Weak` reference (so there is no uncollectable reference cycle through the Rust struct), with a `Drop` backstop for reactors whose triggers never fire. Rust consumers (`HttpClient`, `VersionsHandler`, the Python DB pool wrapper) now receive the runtime or reactor handle explicitly instead of fishing state out of reactor attributes, and the reactor.run() / manual-startup workarounds in tests are no longer needed. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01HyuiV3m44ZbK9o24EnMMES --- changelog.d/19969.misc | 1 + rust/src/deferred.rs | 32 ++- rust/src/handlers/mod.rs | 14 +- rust/src/handlers/versions.rs | 9 +- rust/src/http_client.rs | 16 +- rust/src/lib.rs | 4 +- rust/src/reactor.rs | 88 ++++++++ rust/src/runtime.rs | 191 ++++++++++++++++++ rust/src/storage/db/python_db_pool.rs | 5 +- rust/src/tokio_runtime.rs | 107 ---------- synapse/api/auth/mas.py | 2 +- synapse/server.py | 13 ++ synapse/synapse_rust/__init__.pyi | 17 ++ synapse/synapse_rust/http_client.pyi | 6 +- tests/handlers/test_oauth_delegation.py | 2 - tests/push/test_http.py | 11 +- tests/rest/client/test_delayed_events.py | 11 +- tests/rest/client/test_login_token_request.py | 11 +- tests/rest/client/test_matrixrtc.py | 11 +- tests/rest/client/test_versions.py | 11 +- tests/synapse_rust/test_http_client.py | 11 +- 21 files changed, 374 insertions(+), 199 deletions(-) create mode 100644 changelog.d/19969.misc create mode 100644 rust/src/reactor.rs create mode 100644 rust/src/runtime.rs delete mode 100644 rust/src/tokio_runtime.rs diff --git a/changelog.d/19969.misc b/changelog.d/19969.misc new file mode 100644 index 0000000000..b225128489 --- /dev/null +++ b/changelog.d/19969.misc @@ -0,0 +1 @@ +Refactor how the Rust side of Synapse is tied to the reactor: per-homeserver Rust state (the tokio runtime and reactor handle) now lives in a `RustRuntime` object owned by the `HomeServer`, instead of being stashed in an attribute on the reactor. diff --git a/rust/src/deferred.rs b/rust/src/deferred.rs index 62a1ce6b90..be0d536797 100644 --- a/rust/src/deferred.rs +++ b/rust/src/deferred.rs @@ -25,7 +25,8 @@ use pyo3::{ }; use tokio::sync::oneshot; -use crate::tokio_runtime::runtime; +use crate::reactor::Reactor; +use crate::runtime::RustRuntimeInner; create_exception!( synapse.synapse_rust.http_client, @@ -74,7 +75,7 @@ fn logging_context_module(py: Python<'_>) -> PyResult<&Bound<'_, PyAny>> { /// Does not handle deferred cancellation or contextvars. pub fn create_deferred<'py, F, O>( py: Python<'py>, - reactor: &Bound<'py, PyAny>, + runtime: &Arc, fut: F, ) -> PyResult> where @@ -85,12 +86,12 @@ where let deferred_callback = deferred.getattr("callback")?.unbind(); let deferred_errback = deferred.getattr("errback")?.unbind(); - let rt = runtime(reactor)?; - let handle = rt.handle()?; + let handle = runtime.tokio_handle()?; let task = handle.spawn(fut); - // Unbind the reactor so that we can pass it to the task - let reactor = reactor.clone().unbind(); + // Keep the runtime state (and, through it, the reactor) alive while the + // task is in flight. + let runtime = Arc::clone(runtime); handle.spawn(async move { let res = task.await; @@ -104,19 +105,18 @@ where }, }; - // Re-bind the reactor - let reactor = reactor.bind(py); - // Send the result to the deferred, via `.callback(..)` or `.errback(..)` match res { Ok(obj) => { - reactor - .call_method("callFromThread", (deferred_callback, obj), None) + runtime + .reactor() + .call_from_thread(py, (deferred_callback, obj)) .expect("callFromThread should not fail"); // There's nothing we can really do with errors here } Err(err) => { - reactor - .call_method("callFromThread", (deferred_errback, err), None) + runtime + .reactor() + .call_from_thread(py, (deferred_errback, err)) .expect("callFromThread should not fail"); // There's nothing we can really do with errors here } } @@ -137,7 +137,7 @@ where /// the Twisted reactor and runs to completion regardless of whether the returned Rust /// future is ever polled; awaiting it only observes the result. pub(crate) async fn run_python_awaitable( - reactor: Py, + reactor: Reactor, make_awaitable: F, ) -> PyResult> where @@ -235,9 +235,7 @@ where }, )?; - reactor - .bind(py) - .call_method1(intern!(py, "callFromThread"), (starter,))?; + reactor.call_from_thread(py, (starter,))?; Ok(()) })?; diff --git a/rust/src/handlers/mod.rs b/rust/src/handlers/mod.rs index 5df019adff..856ce7f6e4 100644 --- a/rust/src/handlers/mod.rs +++ b/rust/src/handlers/mod.rs @@ -22,6 +22,7 @@ use pyo3::{ }; use crate::config::SynapseHomeServerConfig; +use crate::runtime::RustRuntime; use crate::storage::db::python_db_pool::PythonDatabasePoolWrapper; use crate::storage::store::Store; @@ -39,16 +40,19 @@ impl RustHandlers { pub fn py_new(py: Python<'_>, homeserver: &Bound<'_, PyAny>) -> PyResult { let config: SynapseHomeServerConfig = homeserver.getattr("config")?.extract()?; - // The Twisted reactor, used both to drive our Tokio runtime and to - // marshal database work back onto the reactor thread. - let reactor = homeserver.call_method0("get_reactor")?.unbind(); + // The per-homeserver Rust state, which gives us the tokio runtime + // and the Twisted reactor. + let runtime: Bound<'_, RustRuntime> = + homeserver.call_method0("get_rust_runtime")?.extract()?; + let runtime = Arc::clone(runtime.get().inner()); // hs.get_datastores().main.db_pool let db_pool_py = homeserver .call_method0("get_datastores")? .getattr("main")? .getattr("db_pool")?; - let db_pool = PythonDatabasePoolWrapper::new(&db_pool_py, reactor.clone_ref(py))?; + let db_pool = + PythonDatabasePoolWrapper::new(&db_pool_py, runtime.reactor().clone_ref(py))?; // Store is shared across all of the handlers so let's use an `Arc` let store = Arc::new(Store { @@ -64,7 +68,7 @@ impl RustHandlers { versions::VersionsHandler { global_unstable_feature_map: Arc::clone(&global_unstable_feature_map), store: Arc::clone(&store), - reactor: reactor.clone_ref(py), + runtime: Arc::clone(&runtime), }, )?; diff --git a/rust/src/handlers/versions.rs b/rust/src/handlers/versions.rs index 25da9d23fc..01289291ce 100644 --- a/rust/src/handlers/versions.rs +++ b/rust/src/handlers/versions.rs @@ -21,6 +21,7 @@ use serde::Serialize; use crate::config::{types::RoomCreationPreset, SynapseHomeServerConfig}; use crate::deferred::create_deferred; +use crate::runtime::RustRuntimeInner; use crate::storage::store::{PerUserExperimentalFeature, Store}; /// `GET /_matrix/client/versions` response @@ -45,9 +46,9 @@ impl<'py> IntoPyObject<'py> for VersionsResponse { pub struct VersionsHandler { pub global_unstable_feature_map: Arc, pub store: Arc, - /// The Twisted reactor, used to bridge our `async` response back into a - /// Twisted deferred that Python can `await`. - pub reactor: Py, + /// The per-homeserver Rust state, used to bridge our `async` response + /// back into a Twisted deferred that Python can `await`. + pub runtime: Arc, } #[pymethods] @@ -63,7 +64,7 @@ impl VersionsHandler { let store = Arc::clone(&self.store); let global_unstable_feature_map = Arc::clone(&self.global_unstable_feature_map); - create_deferred(py, self.reactor.bind(py), async move { + create_deferred(py, &self.runtime, async move { build_versions_response(&store, &global_unstable_feature_map, user_id.as_deref()) .await .map_err(|err| { diff --git a/rust/src/http_client.rs b/rust/src/http_client.rs index aaa1066a67..1a88cf47d0 100644 --- a/rust/src/http_client.rs +++ b/rust/src/http_client.rs @@ -13,6 +13,7 @@ */ use std::collections::HashMap; +use std::sync::Arc; use anyhow::Context; use http_body_util::BodyExt; @@ -21,7 +22,7 @@ use reqwest::RequestBuilder; use crate::deferred::create_deferred; use crate::errors::HttpResponseException; -use crate::tokio_runtime::runtime; +use crate::runtime::{RustRuntime, RustRuntimeInner}; /// Called when registering modules with python. pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { @@ -42,21 +43,18 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> #[pyclass] struct HttpClient { client: reqwest::Client, - reactor: Py, + runtime: Arc, } #[pymethods] impl HttpClient { #[new] - #[pyo3(signature = (reactor, user_agent, http2_only = false))] + #[pyo3(signature = (runtime, user_agent, http2_only = false))] pub fn py_new( - reactor: Bound, + runtime: &Bound<'_, RustRuntime>, user_agent: &str, http2_only: bool, ) -> PyResult { - // Make sure the runtime gets installed - let _ = runtime(&reactor)?; - let mut builder = reqwest::Client::builder().user_agent(user_agent); if http2_only { @@ -69,7 +67,7 @@ impl HttpClient { Ok(HttpClient { client, - reactor: reactor.unbind(), + runtime: Arc::clone(runtime.get().inner()), }) } @@ -107,7 +105,7 @@ impl HttpClient { builder: RequestBuilder, response_limit: usize, ) -> PyResult> { - create_deferred(py, self.reactor.bind(py), async move { + create_deferred(py, &self.runtime, async move { let response = builder.send().await.context("sending request")?; let status = response.status(); diff --git a/rust/src/lib.rs b/rust/src/lib.rs index 28783afbba..90c2d85f6c 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -19,11 +19,12 @@ pub mod json; pub mod matrix_const; pub mod msc4388_rendezvous; pub mod push; +pub mod reactor; pub mod rendezvous; pub mod room_versions; +pub mod runtime; pub mod segmenter; pub mod storage; -pub mod tokio_runtime; pub mod types; lazy_static! { @@ -76,6 +77,7 @@ fn synapse_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { handlers::register_module(py, m)?; http_client::register_module(py, m)?; rendezvous::register_module(py, m)?; + runtime::register_module(py, m)?; msc4388_rendezvous::register_module(py, m)?; segmenter::register_module(py, m)?; room_versions::register_module(py, m)?; diff --git a/rust/src/reactor.rs b/rust/src/reactor.rs new file mode 100644 index 0000000000..999103d51e --- /dev/null +++ b/rust/src/reactor.rs @@ -0,0 +1,88 @@ +/* + * This file is licensed under the Affero General Public License (AGPL) version 3. + * + * Copyright (C) 2026 Element Creations Ltd + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * See the GNU Affero General Public License for more details: + * . + * + */ + +//! A typed wrapper around the Twisted reactor. + +use pyo3::{call::PyCallArgs, exceptions::PyTypeError, intern, prelude::*}; + +/// The reactor methods Rust code relies on. +/// +/// Extracting a [`Reactor`] from a Python object fails if any of these are +/// missing, so mistakes surface as a `TypeError` at the FFI boundary rather +/// than as an `AttributeError` on a tokio worker thread later. +const REQUIRED_METHODS: &[&str] = &["callFromThread", "addSystemEventTrigger"]; + +/// The Twisted reactor, as seen from Rust. +/// +/// This is not a static guarantee that the object behaves like a reactor (it +/// is a foreign Python object), but the duck type is checked once at +/// extraction, and this module is the single place that names the Twisted +/// API surface that Rust code depends on. +pub struct Reactor(Py); + +impl<'a, 'py> FromPyObject<'a, 'py> for Reactor { + type Error = PyErr; + + fn extract(obj: Borrowed<'a, 'py, PyAny>) -> PyResult { + for name in REQUIRED_METHODS { + if !obj.hasattr(*name)? { + return Err(PyTypeError::new_err(format!( + "expected a Twisted reactor, but {} has no `{name}` method", + obj.get_type() + ))); + } + } + + Ok(Reactor(obj.to_owned().unbind())) + } +} + +impl Reactor { + /// `reactor.callFromThread(f, *args)`: schedule a call on the reactor + /// thread. This is the only reactor method that is safe to call from + /// other threads (e.g. tokio workers). + /// + /// `args` is the full argument tuple, starting with the callable itself. + pub fn call_from_thread<'py>( + &self, + py: Python<'py>, + args: impl PyCallArgs<'py>, + ) -> PyResult<()> { + self.0 + .bind(py) + .call_method1(intern!(py, "callFromThread"), args)?; + + Ok(()) + } + + /// Register `callable` to run after the reactor has shut down, via + /// `reactor.addSystemEventTrigger("after", "shutdown", callable)`. + pub fn add_shutdown_trigger( + &self, + py: Python<'_>, + callable: &Bound<'_, PyAny>, + ) -> PyResult<()> { + self.0.bind(py).call_method1( + intern!(py, "addSystemEventTrigger"), + (intern!(py, "after"), intern!(py, "shutdown"), callable), + )?; + + Ok(()) + } + + pub fn clone_ref(&self, py: Python<'_>) -> Reactor { + Reactor(self.0.clone_ref(py)) + } +} diff --git a/rust/src/runtime.rs b/rust/src/runtime.rs new file mode 100644 index 0000000000..8675423d0d --- /dev/null +++ b/rust/src/runtime.rs @@ -0,0 +1,191 @@ +/* + * This file is licensed under the Affero General Public License (AGPL) version 3. + * + * Copyright (C) 2026 Element Creations Ltd + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * See the GNU Affero General Public License for more details: + * . + * + */ + +//! The per-homeserver state for the Rust side of Synapse. +//! +//! A [`RustRuntime`] is created once per homeserver (`hs.get_rust_runtime()`) +//! and holds everything the Rust side keeps for the lifetime of that +//! homeserver: currently the tokio thread pool and a handle to the Twisted +//! reactor. Rust consumers (e.g. the HTTP client) clone the inner +//! [`Arc`] at construction time and don't need the GIL (or +//! the Python-facing object) to reach it afterwards. + +use std::sync::{Arc, Mutex, Weak}; +use std::time::Duration; + +use anyhow::Context; +use pyo3::{exceptions::PyRuntimeError, prelude::*}; +use tokio::runtime::{Handle, Runtime}; + +use crate::reactor::Reactor; + +/// How long to wait for in-flight tokio tasks when shutting down with the +/// reactor. +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); + +/// State of the lazily-started tokio runtime. +enum TokioState { + /// Not started yet; the runtime is built on first use. + NotStarted, + Running(Runtime), + /// Shut down after the reactor stopped. Cannot be restarted. + Shutdown, +} + +/// The state shared between the Python-facing [`RustRuntime`] handle and any +/// Rust-side consumers holding an `Arc` of this. +pub struct RustRuntimeInner { + reactor: Reactor, + tokio: Mutex, + worker_threads: usize, +} + +impl RustRuntimeInner { + /// The Twisted reactor this homeserver runs on. + pub fn reactor(&self) -> &Reactor { + &self.reactor + } + + /// Get a handle to the tokio runtime, starting the runtime if it hasn't + /// been started yet. + pub fn tokio_handle(&self) -> PyResult { + let mut state = self + .tokio + .lock() + .map_err(|_| PyRuntimeError::new_err("tokio runtime lock poisoned"))?; + + match &*state { + TokioState::Running(runtime) => Ok(runtime.handle().clone()), + TokioState::NotStarted => { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(self.worker_threads) + .enable_all() + .build() + .context("building tokio runtime")?; + let handle = runtime.handle().clone(); + *state = TokioState::Running(runtime); + Ok(handle) + } + TokioState::Shutdown => Err(PyRuntimeError::new_err( + "the tokio runtime has been shut down", + )), + } + } + + /// Shut the tokio runtime down, waiting (with the GIL released) for + /// in-flight tasks to finish. Called via [`ShutdownHook`] when the + /// reactor shuts down. + fn shutdown(&self, py: Python<'_>) -> PyResult<()> { + let mut state = self + .tokio + .lock() + .map_err(|_| PyRuntimeError::new_err("tokio runtime lock poisoned"))?; + let previous = std::mem::replace(&mut *state, TokioState::Shutdown); + // Don't hold the lock while blocking on the shutdown below. + drop(state); + + if let TokioState::Running(runtime) = previous { + py.detach(|| runtime.shutdown_timeout(SHUTDOWN_TIMEOUT)); + } + + Ok(()) + } +} + +impl Drop for RustRuntimeInner { + fn drop(&mut self) { + // Backstop for reactors whose shutdown trigger never fires (e.g. + // `MemoryReactorClock` in tests, which is never actually run). + // `shutdown_background` rather than a blocking shutdown, because the + // last `Arc` may be dropped from a task running on this very + // runtime, where blocking would panic. + if let Ok(state) = self.tokio.get_mut() { + if let TokioState::Running(runtime) = std::mem::replace(state, TokioState::Shutdown) { + runtime.shutdown_background(); + } + } + } +} + +/// The Python-facing handle to the per-homeserver Rust state. +/// +/// Constructed by `HomeServer.get_rust_runtime()`, and passed to the Rust +/// classes that need it (which take a clone of the inner [`Arc`] and drop +/// this handle). +#[pyclass(frozen, name = "RustRuntime", module = "synapse.synapse_rust")] +pub struct RustRuntime { + inner: Arc, +} + +impl RustRuntime { + pub fn inner(&self) -> &Arc { + &self.inner + } +} + +#[pymethods] +impl RustRuntime { + #[new] + #[pyo3(signature = (reactor, worker_threads = 4))] + fn py_new(py: Python<'_>, reactor: Reactor, worker_threads: usize) -> PyResult { + let inner = Arc::new(RustRuntimeInner { + reactor, + tokio: Mutex::new(TokioState::NotStarted), + worker_threads, + }); + + // Shut the tokio runtime down when the reactor does. The trigger + // holds only a `Weak` reference: Twisted keeping the hook alive must + // not keep the runtime (nor, via it, the reactor) alive, as that + // would be a reference cycle passing through a Rust field that + // Python's GC cannot see into. + let hook = Py::new( + py, + ShutdownHook { + inner: Arc::downgrade(&inner), + }, + )?; + inner + .reactor + .add_shutdown_trigger(py, hook.bind(py).as_any())?; + + Ok(RustRuntime { inner }) + } +} + +/// The callable registered with +/// `reactor.addSystemEventTrigger("after", "shutdown", ...)`. +#[pyclass(frozen)] +struct ShutdownHook { + inner: Weak, +} + +#[pymethods] +impl ShutdownHook { + fn __call__(&self, py: Python<'_>) -> PyResult<()> { + if let Some(inner) = self.inner.upgrade() { + inner.shutdown(py)?; + } + + Ok(()) + } +} + +/// Called when registering modules with python. +pub fn register_module(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { + m.add_class::()?; + + Ok(()) +} diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs index 89003d574b..b683a9c00e 100644 --- a/rust/src/storage/db/python_db_pool.rs +++ b/rust/src/storage/db/python_db_pool.rs @@ -40,6 +40,7 @@ use pyo3::{ }; use crate::deferred::run_python_awaitable; +use crate::reactor::Reactor; use crate::storage::db::{ DatabasePool, DbRow, DbValue, ErasedInteraction, ErasedResult, Transaction, }; @@ -112,13 +113,13 @@ pub struct PythonDatabasePoolWrapper { /// never gets garbage collected and never points back at the homeserver, so it is /// not part of any reference cycle. Ideally, we could worry about it but /// practically probably doesn't matter. - reactor: Py, + reactor: Reactor, } impl PythonDatabasePoolWrapper { /// Build a wrapper around the Python `DatabasePool` (e.g. /// `hs.get_datastores().main.db_pool`) and the Twisted `reactor`. - pub fn new(database_pool: &Bound<'_, PyAny>, reactor: Py) -> PyResult { + pub fn new(database_pool: &Bound<'_, PyAny>, reactor: Reactor) -> PyResult { Ok(Self { database_pool_py_ref: PyWeakrefReference::new(database_pool)?.unbind(), reactor, diff --git a/rust/src/tokio_runtime.rs b/rust/src/tokio_runtime.rs deleted file mode 100644 index f239040f30..0000000000 --- a/rust/src/tokio_runtime.rs +++ /dev/null @@ -1,107 +0,0 @@ -/* - * This file is licensed under the Affero General Public License (AGPL) version 3. - * - * Copyright (C) 2026 Element Creations Ltd - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU Affero General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * See the GNU Affero General Public License for more details: - * . - * - */ - -use anyhow::Context; -use pyo3::prelude::*; -use tokio::runtime::Runtime; - -/// This is the name of the attribute where we store the runtime on the reactor -static TOKIO_RUNTIME_ATTR: &str = "__synapse_rust_tokio_runtime"; - -/// A Python wrapper around a Tokio runtime. -/// -/// This allows us to 'store' the runtime on the reactor instance, starting it -/// when the reactor starts, and stopping it when the reactor shuts down. -#[pyclass] -pub struct PyTokioRuntime { - runtime: Option, -} - -#[pymethods] -impl PyTokioRuntime { - fn start(&mut self) -> PyResult<()> { - // TODO: allow customization of the runtime like the number of threads - let runtime = tokio::runtime::Builder::new_multi_thread() - .worker_threads(4) - .enable_all() - .build()?; - - self.runtime = Some(runtime); - - Ok(()) - } - - fn shutdown(&mut self) -> PyResult<()> { - let runtime = self - .runtime - .take() - .context("Runtime was already shutdown")?; - - // Dropping the runtime will shut it down - drop(runtime); - - Ok(()) - } -} - -impl PyTokioRuntime { - /// Get the handle to the Tokio runtime, if it is running. - pub fn handle(&self) -> PyResult<&tokio::runtime::Handle> { - let handle = self - .runtime - .as_ref() - .context("Tokio runtime is not running")? - .handle(); - - Ok(handle) - } -} - -/// Get a handle to the Tokio runtime stored on the reactor instance, or create -/// a new one. -pub fn runtime<'a>(reactor: &Bound<'a, PyAny>) -> PyResult> { - if !reactor.hasattr(TOKIO_RUNTIME_ATTR)? { - install_runtime(reactor)?; - } - - get_runtime(reactor) -} - -/// Install a new Tokio runtime on the reactor instance. -fn install_runtime(reactor: &Bound) -> PyResult<()> { - let py = reactor.py(); - let runtime = PyTokioRuntime { runtime: None }; - let runtime = runtime.into_pyobject(py)?; - - // Attach the runtime to the reactor, starting it when the reactor is - // running, stopping it when the reactor is shutting down - reactor.call_method1("callWhenRunning", (runtime.getattr("start")?,))?; - reactor.call_method1( - "addSystemEventTrigger", - ("after", "shutdown", runtime.getattr("shutdown")?), - )?; - reactor.setattr(TOKIO_RUNTIME_ATTR, runtime)?; - - Ok(()) -} - -/// Get a reference to a Tokio runtime handle stored on the reactor instance. -fn get_runtime<'a>(reactor: &Bound<'a, PyAny>) -> PyResult> { - // This will raise if `TOKIO_RUNTIME_ATTR` is not set or if it is - // not a `Runtime`. Careful that this could happen if the user sets it - // manually, or if multiple versions of `pyo3-twisted` are used! - let runtime: Bound = reactor.getattr(TOKIO_RUNTIME_ATTR)?.extract()?; - Ok(runtime.borrow()) -} diff --git a/synapse/api/auth/mas.py b/synapse/api/auth/mas.py index ed0427d6f3..d094410ff9 100644 --- a/synapse/api/auth/mas.py +++ b/synapse/api/auth/mas.py @@ -109,7 +109,7 @@ class MasDelegatedAuth(BaseAuth): self._http_client = hs.get_proxied_http_client() self._rust_http_client = HttpClient( - reactor=hs.get_reactor(), + runtime=hs.get_rust_runtime(), user_agent=self._http_client.user_agent.decode("utf8"), http2_only=self._config.force_http2, ) diff --git a/synapse/server.py b/synapse/server.py index b756223e54..346a239d9b 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -174,6 +174,7 @@ from synapse.state import StateHandler, StateResolutionHandler from synapse.storage import Databases from synapse.storage.controllers import StorageControllers from synapse.streams.events import EventSources +from synapse.synapse_rust import RustRuntime from synapse.synapse_rust.handlers import RustHandlers from synapse.synapse_rust.msc4388_rendezvous import MSC4388RendezvousHandler from synapse.synapse_rust.rendezvous import RendezvousHandler @@ -964,6 +965,18 @@ class HomeServer(metaclass=abc.ABCMeta): def get_rust_handlers(self) -> RustHandlers: return RustHandlers(self) + @cache_in_self + def get_rust_runtime(self) -> RustRuntime: + """The per-homeserver state for the Rust side of Synapse: the tokio + thread pool, plus anything else Rust code keeps for the lifetime of + the homeserver. + + The tokio runtime is started lazily on first use, and shut down by a + reactor shutdown trigger. + """ + # TODO: make the number of worker threads configurable + return RustRuntime(reactor=self.get_reactor(), worker_threads=4) + @cache_in_self def get_event_sources(self) -> EventSources: return EventSources(self) diff --git a/synapse/synapse_rust/__init__.pyi b/synapse/synapse_rust/__init__.pyi index cb3eb7df07..ac254c871d 100644 --- a/synapse/synapse_rust/__init__.pyi +++ b/synapse/synapse_rust/__init__.pyi @@ -1,4 +1,21 @@ +from synapse.types import ISynapseReactor + def sum_as_string(a: int, b: int) -> str: ... def get_rust_file_digest() -> str: ... def reset_logging_config() -> None: ... def get_rustc_version() -> str: ... + +class RustRuntime: + """The per-homeserver state for the Rust side of Synapse. + + Holds the tokio thread pool (started lazily on first use, shut down by a + reactor shutdown trigger) and a handle to the reactor. Rust classes that + need either take this as a constructor argument; get it from + `hs.get_rust_runtime()`. + """ + + def __init__( + self, + reactor: ISynapseReactor, + worker_threads: int = 4, + ) -> None: ... diff --git a/synapse/synapse_rust/http_client.pyi b/synapse/synapse_rust/http_client.pyi index 1814daec73..8a193d98b1 100644 --- a/synapse/synapse_rust/http_client.pyi +++ b/synapse/synapse_rust/http_client.pyi @@ -14,7 +14,7 @@ from typing import Mapping from twisted.internet.defer import Deferred -from synapse.types import ISynapseReactor +from synapse.synapse_rust import RustRuntime class HttpClient: """ @@ -23,7 +23,7 @@ class HttpClient: def __init__( self, - reactor: ISynapseReactor, + runtime: RustRuntime, user_agent: str, http2_only: bool = False, ) -> None: @@ -31,7 +31,7 @@ class HttpClient: Create a new HTTP client backed by reqwest. Args: - reactor: The Twisted reactor to coordinate with + runtime: The per-homeserver Rust state (`hs.get_rust_runtime()`) user_agent: The user agent to use for requests http2_only: Whether to use HTTP/2 only, even on unencrypted connections. By default, it will always use HTTP/1.1 over unencrypted connections, and diff --git a/tests/handlers/test_oauth_delegation.py b/tests/handlers/test_oauth_delegation.py index 995a1134b2..fd3188d232 100644 --- a/tests/handlers/test_oauth_delegation.py +++ b/tests/handlers/test_oauth_delegation.py @@ -216,8 +216,6 @@ class MasAuthDelegation(HomeserverTestCase): def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: self.server = FakeMasServer() hs = self.setup_test_homeserver() - # This triggers the server startup hooks, which starts the Tokio thread pool - reactor.run() self._auth = checked_cast(MasDelegatedAuth, hs.get_auth()) return hs diff --git a/tests/push/test_http.py b/tests/push/test_http.py index 47521a773f..9d748b727e 100644 --- a/tests/push/test_http.py +++ b/tests/push/test_http.py @@ -1236,19 +1236,14 @@ class MSC3881VersionsTestCase(HomeserverTestCase): def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: hs = self.setup_test_homeserver() - # XXX: We must create the Rust HTTP client before we call `reactor.run()` below. - # Twisted's `MemoryReactor` doesn't invoke `callWhenRunning` callbacks if it's - # already running and we rely on that to start the Tokio thread pool in Rust. In - # the future, this may not matter, see https://github.com/twisted/twisted/pull/12514 self._http_client = hs.get_proxied_http_client() + # The tokio thread pool is started lazily on first use, so no + # reactor startup hooks need to run here. _ = HttpClient( - reactor=hs.get_reactor(), + runtime=hs.get_rust_runtime(), user_agent=self._http_client.user_agent.decode("utf8"), ) - # This triggers the server startup hooks, which starts the Tokio thread pool - reactor.run() - return hs def tearDown(self) -> None: diff --git a/tests/rest/client/test_delayed_events.py b/tests/rest/client/test_delayed_events.py index 75d716244a..c97ac1ec49 100644 --- a/tests/rest/client/test_delayed_events.py +++ b/tests/rest/client/test_delayed_events.py @@ -44,19 +44,14 @@ class DelayedEventsUnstableSupportTestCase(HomeserverTestCase): def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: hs = self.setup_test_homeserver() - # XXX: We must create the Rust HTTP client before we call `reactor.run()` below. - # Twisted's `MemoryReactor` doesn't invoke `callWhenRunning` callbacks if it's - # already running and we rely on that to start the Tokio thread pool in Rust. In - # the future, this may not matter, see https://github.com/twisted/twisted/pull/12514 self._http_client = hs.get_proxied_http_client() + # The tokio thread pool is started lazily on first use, so no + # reactor startup hooks need to run here. _ = HttpClient( - reactor=hs.get_reactor(), + runtime=hs.get_rust_runtime(), user_agent=self._http_client.user_agent.decode("utf8"), ) - # This triggers the server startup hooks, which starts the Tokio thread pool - reactor.run() - return hs def tearDown(self) -> None: diff --git a/tests/rest/client/test_login_token_request.py b/tests/rest/client/test_login_token_request.py index d6b2cf054e..a0b3ef7995 100644 --- a/tests/rest/client/test_login_token_request.py +++ b/tests/rest/client/test_login_token_request.py @@ -48,19 +48,14 @@ class LoginTokenRequestServletTestCase(unittest.HomeserverTestCase): self.hs.config.registration.auto_join_rooms = [] self.hs.config.captcha.enable_registration_captcha = False - # XXX: We must create the Rust HTTP client before we call `reactor.run()` below. - # Twisted's `MemoryReactor` doesn't invoke `callWhenRunning` callbacks if it's - # already running and we rely on that to start the Tokio thread pool in Rust. In - # the future, this may not matter, see https://github.com/twisted/twisted/pull/12514 self._http_client = self.hs.get_proxied_http_client() + # The tokio thread pool is started lazily on first use, so no + # reactor startup hooks need to run here. _ = HttpClient( - reactor=self.hs.get_reactor(), + runtime=self.hs.get_rust_runtime(), user_agent=self._http_client.user_agent.decode("utf8"), ) - # This triggers the server startup hooks, which starts the Tokio thread pool - reactor.run() - return self.hs def tearDown(self) -> None: diff --git a/tests/rest/client/test_matrixrtc.py b/tests/rest/client/test_matrixrtc.py index f2bf1596be..3c402ff725 100644 --- a/tests/rest/client/test_matrixrtc.py +++ b/tests/rest/client/test_matrixrtc.py @@ -115,19 +115,14 @@ class MatrixRtcVersionsTestCase(HomeserverTestCase): def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: hs = self.setup_test_homeserver() - # XXX: We must create the Rust HTTP client before we call `reactor.run()` below. - # Twisted's `MemoryReactor` doesn't invoke `callWhenRunning` callbacks if it's - # already running and we rely on that to start the Tokio thread pool in Rust. In - # the future, this may not matter, see https://github.com/twisted/twisted/pull/12514 self._http_client = hs.get_proxied_http_client() + # The tokio thread pool is started lazily on first use, so no + # reactor startup hooks need to run here. _ = HttpClient( - reactor=hs.get_reactor(), + runtime=hs.get_rust_runtime(), user_agent=self._http_client.user_agent.decode("utf8"), ) - # This triggers the server startup hooks, which starts the Tokio thread pool - reactor.run() - return hs def tearDown(self) -> None: diff --git a/tests/rest/client/test_versions.py b/tests/rest/client/test_versions.py index bbdbe38e07..171e2189db 100644 --- a/tests/rest/client/test_versions.py +++ b/tests/rest/client/test_versions.py @@ -40,19 +40,14 @@ class VersionsTestCase(unittest.HomeserverTestCase): def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: hs = self.setup_test_homeserver() - # XXX: We must create the Rust HTTP client before we call `reactor.run()` below. - # Twisted's `MemoryReactor` doesn't invoke `callWhenRunning` callbacks if it's - # already running and we rely on that to start the Tokio thread pool in Rust. In - # the future, this may not matter, see https://github.com/twisted/twisted/pull/12514 self._http_client = hs.get_proxied_http_client() + # The tokio thread pool is started lazily on first use, so no + # reactor startup hooks need to run here. _ = HttpClient( - reactor=hs.get_reactor(), + runtime=hs.get_rust_runtime(), user_agent=self._http_client.user_agent.decode("utf8"), ) - # This triggers the server startup hooks, which starts the Tokio thread pool - reactor.run() - return hs def tearDown(self) -> None: diff --git a/tests/synapse_rust/test_http_client.py b/tests/synapse_rust/test_http_client.py index 845fe2b503..9b5e79809e 100644 --- a/tests/synapse_rust/test_http_client.py +++ b/tests/synapse_rust/test_http_client.py @@ -89,19 +89,14 @@ class HttpClientTestCase(HomeserverTestCase): def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: hs = self.setup_test_homeserver() - # XXX: We must create the Rust HTTP client before we call `reactor.run()` below. - # Twisted's `MemoryReactor` doesn't invoke `callWhenRunning` callbacks if it's - # already running and we rely on that to start the Tokio thread pool in Rust. In - # the future, this may not matter, see https://github.com/twisted/twisted/pull/12514 self._http_client = hs.get_proxied_http_client() + # The tokio thread pool is started lazily on first use, so no + # reactor startup hooks need to run here. self._rust_http_client = HttpClient( - reactor=hs.get_reactor(), + runtime=hs.get_rust_runtime(), user_agent=self._http_client.user_agent.decode("utf8"), ) - # This triggers the server startup hooks, which starts the Tokio thread pool - reactor.run() - return hs def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: