mirror of
https://github.com/element-hq/synapse.git
synced 2026-11-04 06:57:20 +00:00
Cut sliding sync's per-room Python calls from Rust
The benchmark showed two per-room costs under the GIL on large accounts: - Sorting called the events stream cache's `get_max_pos_of_last_change` once per joined room, 2000 Python calls per request on a 2000-room account, even for a [0,19] list. That method is a lookup in the cache's `_entity_to_key` dict, so Rust now reads the dict directly. - `RoomStatusMapView` imported `HaveSentRoom` and built a new object for each room the extensions asked about, about 13 us a call. The class is frozen, so one `LIVE` and one `NEVER` value now serve every room. Rust's reactor time on an incremental request on a 2000-room account fell from 0.55 of Python's to 0.35, and with a [0,499] list and extensions from 0.84 to 0.49. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01L51DoxB8VnH8UjXJFDgpNG
This commit is contained in:
1 parent
98c8ed9f9e
commit
c3ee410acc
2 files changed
+39
-13
No files matched your search
@@ -22,7 +22,7 @@ use std::sync::Arc;
|
||||
|
||||
use anyhow::bail;
|
||||
use indexmap::IndexMap;
|
||||
use pyo3::{exceptions::PyKeyError, prelude::*};
|
||||
use pyo3::{exceptions::PyKeyError, prelude::*, types::PyDict};
|
||||
use regex::Regex;
|
||||
|
||||
use crate::push::utils::{glob_to_regex, GlobMatchType};
|
||||
@@ -389,6 +389,11 @@ async fn rewind_changes(
|
||||
/// The positions the events stream change cache has for `room_ids`, and
|
||||
/// the earliest it knows: `get_max_pos_of_last_change` and
|
||||
/// `get_earliest_known_position`, read under the GIL.
|
||||
///
|
||||
/// `get_max_pos_of_last_change` is a lookup in the cache's `_entity_to_key`
|
||||
/// dict, so this looks each room up in that dict directly. A method call per
|
||||
/// joined room was most of the GIL time of an incremental request on a
|
||||
/// large account.
|
||||
fn stream_cache_positions(
|
||||
cache: &Py<PyAny>,
|
||||
room_ids: &[Arc<str>],
|
||||
@@ -398,12 +403,12 @@ fn stream_cache_positions(
|
||||
let earliest: i64 = cache
|
||||
.call_method0(pyo3::intern!(py, "get_earliest_known_position"))?
|
||||
.extract()?;
|
||||
let method = cache.getattr(pyo3::intern!(py, "get_max_pos_of_last_change"))?;
|
||||
let entity_to_key = cache.getattr(pyo3::intern!(py, "_entity_to_key"))?;
|
||||
let entity_to_key = entity_to_key.cast::<PyDict>()?;
|
||||
let mut positions = HashMap::with_capacity(room_ids.len());
|
||||
for room_id in room_ids {
|
||||
let position: Option<i64> = method.call1((&**room_id,))?.extract()?;
|
||||
if let Some(position) = position {
|
||||
positions.insert(Arc::clone(room_id), position);
|
||||
if let Some(position) = entity_to_key.get_item(&**room_id)? {
|
||||
positions.insert(Arc::clone(room_id), position.extract()?);
|
||||
}
|
||||
}
|
||||
Ok((earliest, positions))
|
||||
|
||||
@@ -32,6 +32,7 @@ use pyo3::{
|
||||
exceptions::{PyRuntimeError, PyValueError},
|
||||
intern,
|
||||
prelude::*,
|
||||
sync::PyOnceLock,
|
||||
types::{PyDict, PyTuple},
|
||||
};
|
||||
|
||||
@@ -270,21 +271,41 @@ impl ConnectionState {
|
||||
}
|
||||
}
|
||||
|
||||
/// Python's `HaveSentRoom` class, with one `LIVE` and one `NEVER` value. The
|
||||
/// class is frozen, so those two serve every room. The extensions ask for
|
||||
/// each room in their lists, so building them per call showed in requests
|
||||
/// with hundreds of rooms.
|
||||
struct PyHaveSentRoom {
|
||||
class: Py<PyAny>,
|
||||
live: Py<PyAny>,
|
||||
never: Py<PyAny>,
|
||||
}
|
||||
|
||||
static PY_HAVE_SENT_ROOM: PyOnceLock<PyHaveSentRoom> = PyOnceLock::new();
|
||||
|
||||
/// A Python `HaveSentRoom`.
|
||||
fn have_sent_room_to_py<'py, T>(
|
||||
py: Python<'py>,
|
||||
status: &HaveSentRoom<T>,
|
||||
token_to_py: impl FnOnce(&T) -> PyResult<Bound<'py, PyAny>>,
|
||||
) -> PyResult<Bound<'py, PyAny>> {
|
||||
let class = py
|
||||
.import(intern!(py, "synapse.types.handlers.sliding_sync"))?
|
||||
.getattr(intern!(py, "HaveSentRoom"))?;
|
||||
let cached = PY_HAVE_SENT_ROOM.get_or_try_init(py, || -> PyResult<_> {
|
||||
let class = py
|
||||
.import(intern!(py, "synapse.types.handlers.sliding_sync"))?
|
||||
.getattr(intern!(py, "HaveSentRoom"))?;
|
||||
Ok(PyHaveSentRoom {
|
||||
live: class.call_method0(intern!(py, "live"))?.unbind(),
|
||||
never: class.call_method0(intern!(py, "never"))?.unbind(),
|
||||
class: class.unbind(),
|
||||
})
|
||||
})?;
|
||||
match status {
|
||||
HaveSentRoom::Never => class.call_method0(intern!(py, "never")),
|
||||
HaveSentRoom::Live => class.call_method0(intern!(py, "live")),
|
||||
HaveSentRoom::Previously(token) => {
|
||||
class.call_method1(intern!(py, "previously"), (token_to_py(token)?,))
|
||||
}
|
||||
HaveSentRoom::Never => Ok(cached.never.bind(py).clone()),
|
||||
HaveSentRoom::Live => Ok(cached.live.bind(py).clone()),
|
||||
HaveSentRoom::Previously(token) => cached
|
||||
.class
|
||||
.bind(py)
|
||||
.call_method1(intern!(py, "previously"), (token_to_py(token)?,)),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user