git-mirrors/continuwuity
1
0
Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity.git synced 2025-09-10 13:12:49 +02:00
Code Issues Projects Releases Packages Wiki Activity

pass and use transaction id to collect timing info

Browse source
This commit is contained in:
Jacob Taylor 2025-08-31 09:16:52 -07:00
commit 9723753b5c
1 changed files with 22 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

31
src/api/server/send.rs
View file

@ -79,10 +79,11 @@ pub(crate) async fn send_transaction_message_route(
} }
let txn_start_time = Instant::now(); let txn_start_time = Instant::now();
let transaction_id = body.transaction_id.to_string();
trace!( trace!(
pdus = body.pdus.len(), pdus = body.pdus.len(),
edus = body.edus.len(), edus = body.edus.len(),
id = ?body.transaction_id, id = transaction_id,
"Processing transaction", "Processing transaction",
); );
@ -106,17 +107,17 @@ pub(crate) async fn send_transaction_message_route(
pdus = body.pdus.len(), pdus = body.pdus.len(),
edus = body.edus.len(), edus = body.edus.len(),
elapsed = ?txn_start_time.elapsed(), elapsed = ?txn_start_time.elapsed(),
id = ?body.transaction_id, id = transaction_id,
"Validated transaction", "Validated transaction",
); );
let results = handle(&services, &client, body.origin(), txn_start_time, pdus, edus).await?; let results = handle(&services, &client, body.origin(), txn_start_time, pdus, edus, &*transaction_id).await?;
info!( info!(
pdus = body.pdus.len(), pdus = body.pdus.len(),
edus = body.edus.len(), edus = body.edus.len(),
elapsed = ?txn_start_time.elapsed(), elapsed = ?txn_start_time.elapsed(),
id = ?body.transaction_id, id = transaction_id,
"Finished txn", "Finished txn",
); );
for (id, result) in &results { for (id, result) in &results {
@ -142,11 +143,13 @@ async fn handle(
started: Instant, started: Instant,
pdus: impl Stream<Item = Pdu> + Send, pdus: impl Stream<Item = Pdu> + Send,
edus: impl Stream<Item = Edu> + Send, edus: impl Stream<Item = Edu> + Send,
transaction_id: &str,
) -> Result<ResolvedMap> { ) -> Result<ResolvedMap> {
let handle_start = Instant::now();
edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu)) edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu))
.boxed() .boxed()
.await; .await;
let pdu_start = Instant::now();
// group pdus by room // group pdus by room
let pdus = pdus let pdus = pdus
.collect() .collect()
@ -157,14 +160,14 @@ async fn handle(
.collect() .collect()
}) })
.await; .await;
let results_start = Instant::now();
// we can evaluate rooms concurrently // we can evaluate rooms concurrently
let results: ResolvedMap = pdus let results: ResolvedMap = pdus
.into_iter() .into_iter()
.try_stream() .try_stream()
.broad_and_then(|(room_id, pdus): (_, Vec<_>)| { .broad_and_then(|(room_id, pdus): (_, Vec<_>)| {
let count = pdus.len(); let count = pdus.len();
handle_room(services, client, origin, started, room_id, pdus.into_iter(), count) handle_room(services, client, origin, started, room_id, pdus.into_iter(), count, transaction_id)
.map_ok(Vec::into_iter) .map_ok(Vec::into_iter)
.map_ok(IterStream::try_stream) .map_ok(IterStream::try_stream)
}) })
@ -172,7 +175,14 @@ async fn handle(
.try_collect() .try_collect()
.boxed() .boxed()
.await?; .await?;
let handle_stop = Instant::now();
info!(
edus = pdu_start.saturating_duration_since(handle_start).as_micros(),
pdus = results_start.saturating_duration_since(pdu_start).as_micros(),
handle_room = handle_stop.saturating_duration_since(results_start).as_micros(),
transaction_id = ?transaction_id,
"handled incoming transaction",
);
Ok(results) Ok(results)
} }
@ -184,6 +194,7 @@ async fn handle_room(
room_id: OwnedRoomId, room_id: OwnedRoomId,
pdus: impl Iterator<Item = Pdu> + Send, pdus: impl Iterator<Item = Pdu> + Send,
count: usize, count: usize,
transaction_id: &str,
) -> Result<Vec<(OwnedEventId, Result)>> { ) -> Result<Vec<(OwnedEventId, Result)>> {
let _room_lock = services let _room_lock = services
.rooms .rooms
@ -201,6 +212,7 @@ async fn handle_room(
trace!( trace!(
%room_id, %room_id,
%event_id, %event_id,
transaction_id = ?transaction_id,
pdu = n + 1, pdu = n + 1,
total = count, total = count,
pdu_elapsed = ?pdu_start_time.elapsed(), pdu_elapsed = ?pdu_start_time.elapsed(),
@ -217,11 +229,12 @@ async fn handle_room(
info!( info!(
%room_id, %room_id,
%event_id, %event_id,
transaction_id = ?transaction_id,
pdu = n + 1, pdu = n + 1,
total = count, total = count,
pdu_elapsed = ?pdu_start_time.elapsed(), pdu_elapsed = ?pdu_start_time.elapsed(),
txn_elapsed = ?txn_start_time.elapsed(), txn_elapsed = ?txn_start_time.elapsed(),
"Finished handling PDU {event_id}", "Finished handling PDU",
); );
n += 1; n += 1;