bench(event cache): add a benchmark for measuring how long it takes to handle a sync update

benchmarks/Cargo.toml

@@ -47,3 +47,7 @@ harness = false
 [[bench]]
 name = "timeline"
 harness = false
+
+[[bench]]
+name = "event_cache"
+harness = false

benchmarks/benches/event_cache.rs

@@ -0,0 +1,154 @@
+use std::{
+    sync::Arc,
+    time::{Duration, Instant},
+};
+
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use matrix_sdk::{
+    SqliteEventCacheStore,
+    store::StoreConfig,
+    sync::{JoinedRoomUpdate, RoomUpdates},
+    test_utils::mocks::MatrixMockServer,
+};
+use matrix_sdk_base::event_cache::store::{IntoEventCacheStore, MemoryStore};
+use matrix_sdk_test::{ALICE, JoinedRoomBuilder, event_factory::EventFactory};
+use ruma::{EventId, RoomId};
+use tempfile::tempdir;
+use tokio::runtime::Builder;
+
+fn handle_room_updates(c: &mut Criterion) {
+    // 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("reading");
+    group.sample_size(10);
+
+    const NUM_EVENTS: usize = 1000;
+
+    for num_rooms in [1, 10, 100] {
+        // Add some joined rooms, each with NUM_EVENTS in it, to the sync response.
+        let mut room_updates = RoomUpdates::default();
+
+        for i in 0..num_rooms {
+            let room_id = RoomId::parse(format!("!room{i}:example.com")).unwrap();
+            let event_factory = EventFactory::new().room(&room_id).sender(&ALICE);
+
+            let mut joined_room_update = JoinedRoomUpdate::default();
+            for j in 0..NUM_EVENTS {
+                let event_id = EventId::parse(format!("$ev{i}_{j}")).unwrap();
+                let event =
+                    event_factory.text_msg(format!("Message {j}")).event_id(&event_id).into();
+                joined_room_update.timeline.events.push(event);
+            }
+            room_updates.joined.insert(room_id.clone(), joined_room_update);
+        }
+
+        // Declare new stores for this set of events.
+        let sqlite_temp_dir = tempdir().unwrap();
+        let stores = vec![
+            ("memory store", MemoryStore::default().into_event_cache_store()),
+            (
+                "sqlite store",
+                runtime.block_on(async {
+                    SqliteEventCacheStore::open(sqlite_temp_dir.path().join("bench"), None)
+                        .await
+                        .unwrap()
+                        .into_event_cache_store()
+                }),
+            ),
+        ];
+
+        let server = Arc::new(runtime.block_on(MatrixMockServer::new()));
+
+        for (store_name, store) in &stores {
+            // Define the throughput.
+            group.throughput(Throughput::Elements(num_rooms));
+
+            // Bench the handling of room updates.
+            group.bench_function(
+                BenchmarkId::new(format!("handle_room_updates/{store_name}"), num_rooms),
+                |bencher| {
+                    // Ideally we'd use `iter_with_setup` here, but it doesn't allow an async setup
+                    // (which we need to setup the client), see also
+                    // https://github.com/bheisler/criterion.rs/issues/751.
+                    bencher.to_async(&runtime).iter_custom(|num_iters| {
+                        let room_updates = room_updates.clone();
+                        let server = server.clone();
+
+                        async move {
+                            let mut total_time = Duration::new(0, 0);
+
+                            for _ in 0..num_iters {
+                                // Setup code.
+                                let client = server
+                                    .client_builder()
+                                    .store_config(
+                                        StoreConfig::new(
+                                            "cross-process-store-locks-holder-name".to_owned(),
+                                        )
+                                        .event_cache_store(store.clone()),
+                                    )
+                                    .build()
+                                    .await;
+
+                                client.event_cache().subscribe().unwrap();
+
+                                // Make sure the client knows about all the to-be joined rooms.
+                                server
+                                    .mock_sync()
+                                    .ok_and_run(&client, |builder| {
+                                        for room_id in room_updates.joined.keys() {
+                                            builder
+                                                .add_joined_room(JoinedRoomBuilder::new(room_id));
+                                        }
+                                    })
+                                    .await;
+
+                                let time_before = Instant::now();
+                                client
+                                    .event_cache()
+                                    .handle_room_updates(room_updates.clone())
+                                    .await
+                                    .unwrap();
+                                total_time += time_before.elapsed();
+                            }
+
+                            total_time
+                        }
+                    })
+                },
+            );
+        }
+
+        for (_store_name, store) in stores.into_iter() {
+            let _guard = runtime.enter();
+            drop(store);
+        }
+    }
+
+    group.finish()
+}
+
+fn criterion() -> Criterion {
+    #[cfg(target_os = "linux")]
+    let criterion = Criterion::default().with_profiler(pprof::criterion::PProfProfiler::new(
+        100,
+        pprof::criterion::Output::Flamegraph(None),
+    ));
+    #[cfg(not(target_os = "linux"))]
+    let criterion = Criterion::default();
+
+    criterion
+}
+
+criterion_group! {
+    name = event_cache;
+    config = criterion();
+    targets = handle_room_updates,
+}
+
+criterion_main!(event_cache);

crates/matrix-sdk-base/src/event_cache/store/traits.rs

@@ -464,6 +464,12 @@ pub trait IntoEventCacheStore {
     fn into_event_cache_store(self) -> Arc<DynEventCacheStore>;
 }
 
+impl IntoEventCacheStore for Arc<DynEventCacheStore> {
+    fn into_event_cache_store(self) -> Arc<DynEventCacheStore> {
+        self
+    }
+}
+
 impl<T> IntoEventCacheStore for T
 where
     T: EventCacheStore + Sized + 'static,

crates/matrix-sdk/src/event_cache/mod.rs

@@ -244,6 +244,12 @@ impl EventCache {
         .await;
     }
 
+    /// For benchmarking purposes only.
+    #[doc(hidden)]
+    pub async fn handle_room_updates(&self, updates: RoomUpdates) -> Result<()> {
+        self.inner.handle_room_updates(updates).await
+    }
+
     #[instrument(skip_all)]
     async fn listen_task(
         inner: Arc<EventCacheInner>,
@@ -519,6 +525,11 @@ impl EventCacheInner {
             self.multiple_room_updates_lock.lock().await
         };
 
+        // Note: bnjbvr tried to make this concurrent at some point, but it turned out
+        // to be a performance regression, even for large sync updates. Lacking
+        // time to investigate, this code remains sequential for now. See also
+        // https://github.com/matrix-org/matrix-rust-sdk/pull/5426.
+
         // Left rooms.
         for (room_id, left_room_update) in updates.left {
             let room = self.for_room(&room_id).await?;