matrix-rust-sdk

public

Star0

Watch0

26 commits 4 branches 0 tags 0 releases16.6 MiB

use std::{pin::Pin, sync::Arc};

use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
use matrix_sdk::{
    RoomInfo, RoomState, SqliteEventCacheStore, StateStore,
    cross_process_lock::CrossProcessLockConfig,
    store::StoreConfig,
    sync::{JoinedRoomUpdate, RoomUpdates},
    test_utils::client::MockClientBuilder,
};
use matrix_sdk_base::event_cache::store::{DynEventCacheStore, IntoEventCacheStore, MemoryStore};
use matrix_sdk_test::{ALICE, base64_sha256_hash, event_factory::EventFactory};
use ruma::{
    OwnedRoomId, RoomId,
    events::{relation::RelationType, room::message::RoomMessageEventContentWithoutRelation},
    room_id,
};
use tempfile::tempdir;
use tokio::runtime::Builder;

type StoreBuilder = Box<dyn Fn() -> Pin<Box<dyn Future<Output = Arc<DynEventCacheStore>>>>>;

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("Event cache room updates");
    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();

        let mut changes = matrix_sdk::StateChanges::default();

        for i in 0..num_rooms {
            // Synapse's room IDs for rooms v1 to v11 have an 18 characters localpart.
            let raw_room_id = format!("!firstbatchroom{i:04}:example.com");

            let room_id = if i % 10 == 9 {
                // Make 1 in 10 rooms use a room v12 ID, which is a base64 hash similar to an
                // event ID.
                RoomId::new_v2(&base64_sha256_hash(raw_room_id.as_bytes())).unwrap()
            } else {
                OwnedRoomId::try_from(raw_room_id).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 = event_factory.text_msg(format!("Message {j}")).into();
                joined_room_update.timeline.events.push(event);
            }
            room_updates.joined.insert(room_id.clone(), 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 store_builders: Vec<(_, StoreBuilder)> = vec![
            (
                "memory",
                Box::new(|| Box::pin(async { MemoryStore::default().into_event_cache_store() })),
            ),
            (
                "SQLite",
                Box::new(move || {
                    let temp_dir = temp_dir.clone();
                    Box::pin(async move {
                        // Remove all the files in the temp_dir, to reset the event cache state.
                        for entry in temp_dir.path().read_dir().unwrap() {
                            let entry = entry.unwrap();
                            let path = entry.path();
                            if path.is_dir() {
                                // If it's a directory, remove it recursively.
                                std::fs::remove_dir_all(path).unwrap();
                            } else {
                                std::fs::remove_file(path).unwrap();
                            }
                        }

                        // Recreate a new store.
                        SqliteEventCacheStore::open(temp_dir.path().join("bench"), None)
                            .await
                            .unwrap()
                            .into_event_cache_store()
                    })
                }),
            ),
        ];

        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 (store_name, store_builder) in &store_builders {
            let client = runtime.block_on(async {
                let event_cache_store = store_builder().await;

                let client = MockClientBuilder::new(None)
                    .on_builder(|builder| {
                        builder.store_config(
                            StoreConfig::new(CrossProcessLockConfig::multi_process(
                                "cross-process-store-locks-holder-name",
                            ))
                            .state_store(state_store.clone())
                            .event_cache_store(event_cache_store.clone()),
                        )
                    })
                    .build()
                    .await;

                client.event_cache().subscribe().unwrap();

                client
            });

            // Define a state store with all rooms known in it.
            // Define the throughput.
            group.throughput(Throughput::Elements(num_rooms));

            // Bench the handling of room updates.
            group.bench_function(
                BenchmarkId::new(
                    format!("Event cache room updates[{store_name}]"),
                    format!("room count: {num_rooms}"),
                ),
                |bencher| {
                    bencher.to_async(&runtime).iter(
                        // The routine itself.
                        || {
                            let room_updates = room_updates.clone();
                            let client = client.clone();

                            async move {
                                client.event_cache().clear_all_rooms().await.unwrap();

                                client
                                    .event_cache()
                                    .handle_room_updates(room_updates.clone())
                                    .await
                                    .unwrap();
                            }
                        },
                    )
                },
            );
        }
    }

    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);

    // Room v1-v11 ID.
    let room_id = room_id!("!initialtestingroom:ben.ch");
    // Room v12 ID.
    let other_room_id = room_id!("!ICMDdumUm6RRX_eWYY2wMb2w0CY0Z_5OvlY2gBR6ELc");

    // 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 = event_factory.text_msg("hello world").into_event();
        let target_event_id =
            { &target_event.event_id().expect("generated event has an event ID") };
        joined_room_update.timeline.events.push(target_event);

        // Add the numerous edits.
        for i in 0..num_related_events {
            let event = event_factory
                .text_msg(format!("* edit {i}"))
                .edit(
                    target_event_id,
                    RoomMessageEventContentWithoutRelation::text_plain(format!("edit {i}")),
                )
                .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 = event_factory.text_msg(format!("unrelated message {i}")).into();
            joined_room_update.timeline.events.push(event);
        }

        // Add other events, in the same room, related to other events.
        let other_target_event = event_factory.text_msg("hello world").into_event();
        let other_target_event_id =
            other_target_event.event_id().expect("generated event has an event ID");
        joined_room_update.timeline.events.push(other_target_event);

        for _i in 0..NUM_OTHER_EVENTS {
            let event = event_factory.reaction(&other_target_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 = event_factory.text_msg(format!("hi {i}")).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(CrossProcessLockConfig::multi_process(
                                "cross-process-store-locks-holder-name",
                            ))
                            .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()
                                    .unwrap();
                                assert_eq!(target.event_id().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, find_event_relations,
}

criterion_main!(event_cache);

1	use std::{pin::Pin, sync::Arc};
2
3	use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
4	use matrix_sdk::{
5	RoomInfo, RoomState, SqliteEventCacheStore, StateStore,
6	cross_process_lock::CrossProcessLockConfig,
7	store::StoreConfig,
8	sync::{JoinedRoomUpdate, RoomUpdates},
9	test_utils::client::MockClientBuilder,
10	};
11	use matrix_sdk_base::event_cache::store::{DynEventCacheStore, IntoEventCacheStore, MemoryStore};
12	use matrix_sdk_test::{ALICE, base64_sha256_hash, event_factory::EventFactory};
13	use ruma::{
14	OwnedRoomId, RoomId,
15	events::{relation::RelationType, room::message::RoomMessageEventContentWithoutRelation},
16	room_id,
17	};
18	use tempfile::tempdir;
19	use tokio::runtime::Builder;
20
21	type StoreBuilder = Box<dyn Fn() -> Pin<Box<dyn Future<Output = Arc<DynEventCacheStore>>>>>;
22
23	fn handle_room_updates(c: &mut Criterion) {
24	// Create a new asynchronous runtime.
25	let runtime = Builder::new_multi_thread()
26	.enable_time()
27	.enable_io()
28	.build()
29	.expect("Failed to create an asynchronous runtime");
30
31	let mut group = c.benchmark_group("Event cache room updates");
32	group.sample_size(10);
33
34	const NUM_EVENTS: usize = 1000;
35
36	for num_rooms in [1, 10, 100] {
37	// Add some joined rooms, each with NUM_EVENTS in it, to the sync response.
38	let mut room_updates = RoomUpdates::default();
39
40	let mut changes = matrix_sdk::StateChanges::default();
41
42	for i in 0..num_rooms {
43	// Synapse's room IDs for rooms v1 to v11 have an 18 characters localpart.
44	let raw_room_id = format!("!firstbatchroom{i:04}:example.com");
45
46	let room_id = if i % 10 == 9 {
47	// Make 1 in 10 rooms use a room v12 ID, which is a base64 hash similar to an
48	// event ID.
49	RoomId::new_v2(&base64_sha256_hash(raw_room_id.as_bytes())).unwrap()
50	} else {
51	OwnedRoomId::try_from(raw_room_id).unwrap()
52	};
53	let event_factory = EventFactory::new().room(&room_id).sender(&ALICE);
54
55	let mut joined_room_update = JoinedRoomUpdate::default();
56	for j in 0..NUM_EVENTS {
57	let event = event_factory.text_msg(format!("Message {j}")).into();
58	joined_room_update.timeline.events.push(event);
59	}
60	room_updates.joined.insert(room_id.clone(), joined_room_update);
61
62	changes.add_room(RoomInfo::new(&room_id, RoomState::Joined));
63	}
64
65	// Declare new stores for this set of events.
66	let temp_dir = Arc::new(tempdir().unwrap());
67
68	let store_builders: Vec<(_, StoreBuilder)> = vec![
69	(
70	"memory",
71	Box::new(\|\| Box::pin(async { MemoryStore::default().into_event_cache_store() })),
72	),
73	(
74	"SQLite",
75	Box::new(move \|\| {
76	let temp_dir = temp_dir.clone();
77	Box::pin(async move {
78	// Remove all the files in the temp_dir, to reset the event cache state.
79	for entry in temp_dir.path().read_dir().unwrap() {
80	let entry = entry.unwrap();
81	let path = entry.path();
82	if path.is_dir() {
83	// If it's a directory, remove it recursively.
84	std::fs::remove_dir_all(path).unwrap();
85	} else {
86	std::fs::remove_file(path).unwrap();
87	}
88	}
89
90	// Recreate a new store.
91	SqliteEventCacheStore::open(temp_dir.path().join("bench"), None)
92	.await
93	.unwrap()
94	.into_event_cache_store()
95	})
96	}),
97	),
98	];
99
100	let state_store = runtime.block_on(async {
101	let state_store = matrix_sdk::MemoryStore::new();
102	state_store.save_changes(&changes).await.unwrap();
103	Arc::new(state_store)
104	});
105
106	for (store_name, store_builder) in &store_builders {
107	let client = runtime.block_on(async {
108	let event_cache_store = store_builder().await;
109
110	let client = MockClientBuilder::new(None)
111	.on_builder(\|builder\| {
112	builder.store_config(
113	StoreConfig::new(CrossProcessLockConfig::multi_process(
114	"cross-process-store-locks-holder-name",
115	))
116	.state_store(state_store.clone())
117	.event_cache_store(event_cache_store.clone()),
118	)
119	})
120	.build()
121	.await;
122
123	client.event_cache().subscribe().unwrap();
124
125	client
126	});
127
128	// Define a state store with all rooms known in it.
129	// Define the throughput.
130	group.throughput(Throughput::Elements(num_rooms));
131
132	// Bench the handling of room updates.
133	group.bench_function(
134	BenchmarkId::new(
135	format!("Event cache room updates[{store_name}]"),
136	format!("room count: {num_rooms}"),
137	),
138	\|bencher\| {
139	bencher.to_async(&runtime).iter(
140	// The routine itself.
141	\|\| {
142	let room_updates = room_updates.clone();
143	let client = client.clone();
144
145	async move {
146	client.event_cache().clear_all_rooms().await.unwrap();
147
148	client
149	.event_cache()
150	.handle_room_updates(room_updates.clone())
151	.await
152	.unwrap();
153	}
154	},
155	)
156	},
157	);
158	}
159	}
160
161	group.finish()
162	}
163
164	fn find_event_relations(c: &mut Criterion) {
165	// Number of other events to saturate the DB, but that will not be affected by
166	// the benchmark. A small multiple of this number will be added.
167	// When running locally, run with more events than in Codespeed CI.
168	#[cfg(feature = "codspeed")]
169	const NUM_OTHER_EVENTS: usize = 100;
170	#[cfg(not(feature = "codspeed"))]
171	const NUM_OTHER_EVENTS: usize = 1000;
172
173	// Create a new asynchronous runtime.
174	let runtime = Builder::new_multi_thread()
175	.enable_time()
176	.enable_io()
177	.build()
178	.expect("Failed to create an asynchronous runtime");
179
180	let mut group = c.benchmark_group("Event cache room updates");
181	group.sample_size(10);
182
183	// Room v1-v11 ID.
184	let room_id = room_id!("!initialtestingroom:ben.ch");
185	// Room v12 ID.
186	let other_room_id = room_id!("!ICMDdumUm6RRX_eWYY2wMb2w0CY0Z_5OvlY2gBR6ELc");
187
188	// Make the state store aware of the room, so that `client.get_room()` works
189	// with it.
190	let mut changes = matrix_sdk::StateChanges::default();
191	changes.add_room(RoomInfo::new(room_id, RoomState::Joined));
192	changes.add_room(RoomInfo::new(other_room_id, RoomState::Joined));
193	let state_store = runtime.block_on(async {
194	let state_store = matrix_sdk::MemoryStore::new();
195	state_store.save_changes(&changes).await.unwrap();
196	Arc::new(state_store)
197	});
198
199	for num_related_events in [10, 100, 1000] {
200	// Prefill the event cache store with one event and N related events.
201	let mut room_updates = RoomUpdates::default();
202
203	let event_factory = EventFactory::new().room(room_id).sender(&ALICE);
204
205	let mut joined_room_update = JoinedRoomUpdate::default();
206
207	// Add the target event.
208	let target_event = event_factory.text_msg("hello world").into_event();
209	let target_event_id =
210	{ &target_event.event_id().expect("generated event has an event ID") };
211	joined_room_update.timeline.events.push(target_event);
212
213	// Add the numerous edits.
214	for i in 0..num_related_events {
215	let event = event_factory
216	.text_msg(format!("* edit {i}"))
217	.edit(
218	target_event_id,
219	RoomMessageEventContentWithoutRelation::text_plain(format!("edit {i}")),
220	)
221	.into();
222	joined_room_update.timeline.events.push(event);
223	}
224
225	// Add other events, in the same room, without a relation.
226	for i in 0..NUM_OTHER_EVENTS {
227	let event = event_factory.text_msg(format!("unrelated message {i}")).into();
228	joined_room_update.timeline.events.push(event);
229	}
230
231	// Add other events, in the same room, related to other events.
232	let other_target_event = event_factory.text_msg("hello world").into_event();
233	let other_target_event_id =
234	other_target_event.event_id().expect("generated event has an event ID");
235	joined_room_update.timeline.events.push(other_target_event);
236
237	for _i in 0..NUM_OTHER_EVENTS {
238	let event = event_factory.reaction(&other_target_event_id, "👍").into();
239	joined_room_update.timeline.events.push(event);
240	}
241
242	room_updates.joined.insert(room_id.to_owned(), joined_room_update);
243
244	// Add other events, in another room.
245	let mut other_joined_room_update = JoinedRoomUpdate::default();
246	let event_factory = event_factory.room(other_room_id);
247	for i in 0..NUM_OTHER_EVENTS {
248	let event = event_factory.text_msg(format!("hi {i}")).into();
249	other_joined_room_update.timeline.events.push(event);
250	}
251	room_updates.joined.insert(other_room_id.to_owned(), other_joined_room_update);
252
253	changes.add_room(RoomInfo::new(room_id, RoomState::Joined));
254
255	// Declare new stores for this set of events.
256	let temp_dir = Arc::new(tempdir().unwrap());
257
258	let stores = vec![
259	("memory", MemoryStore::default().into_event_cache_store()),
260	(
261	"SQLite",
262	runtime.block_on(async {
263	SqliteEventCacheStore::open(temp_dir.path().join("bench"), None)
264	.await
265	.unwrap()
266	.into_event_cache_store()
267	}),
268	),
269	];
270
271	for (store_name, event_cache_store) in stores {
272	let (client, room_event_cache, _drop_handles) = runtime.block_on(async {
273	let client = MockClientBuilder::new(None)
274	.on_builder(\|builder\| {
275	builder.store_config(
276	StoreConfig::new(CrossProcessLockConfig::multi_process(
277	"cross-process-store-locks-holder-name",
278	))
279	.state_store(state_store.clone())
280	.event_cache_store(event_cache_store),
281	)
282	})
283	.build()
284	.await;
285
286	client.event_cache().subscribe().unwrap();
287
288	// Sync the updates before starting the benchmark.
289	let mut update_recv = client.event_cache().subscribe_to_room_generic_updates();
290
291	client.event_cache().handle_room_updates(room_updates.clone()).await.unwrap();
292
293	// Wait for the event cache to notify us of the room updates.
294	let update = update_recv.recv().await.unwrap();
295	assert!(update.room_id == room_id \|\| update.room_id == other_room_id);
296
297	let update = update_recv.recv().await.unwrap();
298	assert!(update.room_id == room_id \|\| update.room_id == other_room_id);
299
300	let room = client.get_room(room_id).unwrap();
301	let room_event_cache = room.event_cache().await.unwrap();
302
303	(client, room_event_cache.0, room_event_cache.1)
304	});
305
306	// Define the throughput.
307	group.throughput(Throughput::Elements(num_related_events));
308
309	for filter in [None, Some(vec![RelationType::Replacement])] {
310	group.bench_function(
311	BenchmarkId::new(
312	format!("Event cache find_event_relations[{store_name}]"),
313	format!(
314	"{num_related_events} events, {} filter",
315	if filter.is_some() { "edits" } else { "#no" },
316	),
317	),
318	\|bencher\| {
319	bencher.to_async(&runtime).iter_batched(
320	// The setup.
321	\|\| (room_event_cache.clone(), filter.clone()),
322	// The routine itself.
323	\|(room_event_cache, filter)\| async move {
324	let (target, relations) = room_event_cache
325	.find_event_with_relations(target_event_id, filter)
326	.await
327	.unwrap()
328	.unwrap();
329	assert_eq!(target.event_id().unwrap(), *target_event_id);
330	assert_eq!(relations.len(), num_related_events as usize);
331	},
332	criterion::BatchSize::PerIteration,
333	)
334	},
335	);
336	}
337
338	{
339	let _guard = runtime.enter();
340	drop(room_event_cache);
341	drop(client);
342	drop(_drop_handles);
343	}
344	}
345	}
346
347	{
348	let _guard = runtime.enter();
349	drop(state_store);
350	}
351
352	group.finish()
353	}
354
355	criterion_group! {
356	name = event_cache;
357	config = Criterion::default();
358	targets = handle_room_updates, find_event_relations,
359	}
360
361	criterion_main!(event_cache);