From 001dadffe13df6c45d515c91670818a7a47b997e Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Tue, 26 Aug 2025 15:57:06 +0200 Subject: [PATCH] bench: add a benchmark for finding related events (#5578) Does what it says on the tin. Split from the performance fix, so we can get some initial numbers on the CI bench runs too. Part of investigating https://github.com/matrix-org/matrix-rust-sdk/issues/5572 --- benchmarks/benches/event_cache.rs | 201 +++++++++++++++++++++++++++++- 1 file changed, 199 insertions(+), 2 deletions(-) diff --git a/benchmarks/benches/event_cache.rs b/benchmarks/benches/event_cache.rs index 9afb46ff3..10f7627ed 100644 --- a/benchmarks/benches/event_cache.rs +++ b/benchmarks/benches/event_cache.rs @@ -9,7 +9,11 @@ use matrix_sdk::{ }; use matrix_sdk_base::event_cache::store::{DynEventCacheStore, IntoEventCacheStore, MemoryStore}; use matrix_sdk_test::{ALICE, event_factory::EventFactory}; -use ruma::{EventId, RoomId}; +use ruma::{ + EventId, RoomId, event_id, + events::{relation::RelationType, room::message::RoomMessageEventContentWithoutRelation}, + room_id, +}; use tempfile::tempdir; use tokio::runtime::Builder; @@ -147,10 +151,203 @@ fn handle_room_updates(c: &mut Criterion) { group.finish() } +fn find_event_relations(c: &mut Criterion) { + // Number of other events to saturate the DB, but that will not be affected by + // the benchmark. A small multiple of this number will be added. + // When running locally, run with more events than in Codespeed CI. + #[cfg(feature = "codspeed")] + const NUM_OTHER_EVENTS: usize = 100; + #[cfg(not(feature = "codspeed"))] + const NUM_OTHER_EVENTS: usize = 1000; + + // Create a new asynchronous runtime. + let runtime = Builder::new_multi_thread() + .enable_time() + .enable_io() + .build() + .expect("Failed to create an asynchronous runtime"); + + let mut group = c.benchmark_group("Event cache room updates"); + group.sample_size(10); + + let room_id = room_id!("!room:ben.ch"); + let other_room_id = room_id!("!other-room:ben.ch"); + + // Make the state store aware of the room, so that `client.get_room()` works + // with it. + let mut changes = matrix_sdk::StateChanges::default(); + changes.add_room(RoomInfo::new(room_id, RoomState::Joined)); + changes.add_room(RoomInfo::new(other_room_id, RoomState::Joined)); + let state_store = runtime.block_on(async { + let state_store = matrix_sdk::MemoryStore::new(); + state_store.save_changes(&changes).await.unwrap(); + Arc::new(state_store) + }); + + for num_related_events in [10, 100, 1000] { + // Prefill the event cache store with one event and N related events. + let mut room_updates = RoomUpdates::default(); + + let event_factory = EventFactory::new().room(room_id).sender(&ALICE); + + let mut joined_room_update = JoinedRoomUpdate::default(); + + // Add the target event. + let target_event_id = event_id!("$target"); + let target_event = + event_factory.text_msg("hello world").event_id(target_event_id).into_event(); + joined_room_update.timeline.events.push(target_event); + + // Add the numerous edits. + for i in 0..num_related_events { + let event_id = EventId::parse(format!("$edit{i}")).unwrap(); + let event = event_factory + .text_msg(format!("* edit {i}")) + .edit( + target_event_id, + RoomMessageEventContentWithoutRelation::text_plain(format!("edit {i}")), + ) + .event_id(&event_id) + .into(); + joined_room_update.timeline.events.push(event); + } + + // Add other events, in the same room, without a relation. + for i in 0..NUM_OTHER_EVENTS { + let event_id = EventId::parse(format!("$msg{i}")).unwrap(); + let event = + event_factory.text_msg(format!("unrelated message {i}")).event_id(&event_id).into(); + joined_room_update.timeline.events.push(event); + } + + // Add other events, in the same room, related to other events. + let other_target_event_id = event_id!("$other_target"); + let other_target_event = + event_factory.text_msg("hello world").event_id(other_target_event_id).into_event(); + joined_room_update.timeline.events.push(other_target_event); + + for i in 0..NUM_OTHER_EVENTS { + let event_id = EventId::parse(format!("$unrelated{i}")).unwrap(); + let event = + event_factory.reaction(other_target_event_id, "👍").event_id(&event_id).into(); + joined_room_update.timeline.events.push(event); + } + + room_updates.joined.insert(room_id.to_owned(), joined_room_update); + + // Add other events, in another room. + let mut other_joined_room_update = JoinedRoomUpdate::default(); + let event_factory = event_factory.room(other_room_id); + for i in 0..NUM_OTHER_EVENTS { + let event_id = EventId::parse(format!("$other_room{i}")).unwrap(); + let event = event_factory.text_msg(format!("hi {i}")).event_id(&event_id).into(); + other_joined_room_update.timeline.events.push(event); + } + room_updates.joined.insert(other_room_id.to_owned(), other_joined_room_update); + + changes.add_room(RoomInfo::new(room_id, RoomState::Joined)); + + // Declare new stores for this set of events. + let temp_dir = Arc::new(tempdir().unwrap()); + + let stores = vec![ + ("memory", MemoryStore::default().into_event_cache_store()), + ( + "SQLite", + runtime.block_on(async { + SqliteEventCacheStore::open(temp_dir.path().join("bench"), None) + .await + .unwrap() + .into_event_cache_store() + }), + ), + ]; + + for (store_name, event_cache_store) in stores { + let (client, room_event_cache, _drop_handles) = runtime.block_on(async { + let client = MockClientBuilder::new(None) + .on_builder(|builder| { + builder.store_config( + StoreConfig::new("cross-process-store-locks-holder-name".to_owned()) + .state_store(state_store.clone()) + .event_cache_store(event_cache_store), + ) + }) + .build() + .await; + + client.event_cache().subscribe().unwrap(); + + // Sync the updates before starting the benchmark. + let mut update_recv = client.event_cache().subscribe_to_room_generic_updates(); + + client.event_cache().handle_room_updates(room_updates.clone()).await.unwrap(); + + // Wait for the event cache to notify us of the room updates. + let update = update_recv.recv().await.unwrap(); + assert!(update.room_id == room_id || update.room_id == other_room_id); + + let update = update_recv.recv().await.unwrap(); + assert!(update.room_id == room_id || update.room_id == other_room_id); + + let room = client.get_room(room_id).unwrap(); + let room_event_cache = room.event_cache().await.unwrap(); + + (client, room_event_cache.0, room_event_cache.1) + }); + + // Define the throughput. + group.throughput(Throughput::Elements(num_related_events)); + + for filter in [None, Some(vec![RelationType::Replacement])] { + group.bench_function( + BenchmarkId::new( + format!("Event cache find_event_relations[{store_name}]"), + format!( + "{num_related_events} events, {} filter", + if filter.is_some() { "edits" } else { "#no" }, + ), + ), + |bencher| { + bencher.to_async(&runtime).iter_batched( + // The setup. + || (room_event_cache.clone(), filter.clone()), + // The routine itself. + |(room_event_cache, filter)| async move { + let (target, relations) = room_event_cache + .find_event_with_relations(target_event_id, filter) + .await + .unwrap(); + assert_eq!(target.event_id().as_deref().unwrap(), target_event_id); + assert_eq!(relations.len(), num_related_events as usize); + }, + criterion::BatchSize::PerIteration, + ) + }, + ); + } + + { + let _guard = runtime.enter(); + drop(room_event_cache); + drop(client); + drop(_drop_handles); + } + } + } + + { + let _guard = runtime.enter(); + drop(state_store); + } + + group.finish() +} + criterion_group! { name = event_cache; config = Criterion::default(); - targets = handle_room_updates, + targets = handle_room_updates, find_event_relations, } criterion_main!(event_cache);