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: