perf: Don't hold state lock when unnecessary

This commit is contained in:
timedout committed 2026-10-16 19:26:27 +01:00
1 parent b0a5caf576
commit 3ebd068013
29 files changed
+92 -139

No files matched your search

+9 -9
View File
@@ -75,7 +75,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
PartialPdu::state(String::new(), &create_content),
server_user,
room_id.as_deref(),
&state_lock,
Some(&state_lock),
)
.boxed()
.await?;
@@ -100,7 +100,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
),
server_user,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.boxed()
.await?;
@@ -135,7 +135,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
PartialPdu::state(String::new(), &power_levels_content),
server_user,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.boxed()
.await?;
@@ -148,7 +148,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
PartialPdu::state(String::new(), &RoomJoinRulesEventContent::new(JoinRule::Invite)),
server_user,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.boxed()
.await?;
@@ -164,7 +164,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
),
server_user,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.boxed()
.await?;
@@ -180,7 +180,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
),
server_user,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.boxed()
.await?;
@@ -194,7 +194,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
PartialPdu::state(String::new(), &RoomNameEventContent::new(room_name)),
server_user,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.boxed()
.await?;
@@ -207,7 +207,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
PartialPdu::state(String::new(), &RoomTopicEventContent::markdown(room_topic)),
server_user,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.boxed()
.await?;
@@ -223,7 +223,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
PartialPdu::state(String::new(), &alias_content),
server_user,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.boxed()
.await?;
+6 -6
View File
@@ -56,7 +56,7 @@ impl super::Service {
),
server_user,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.await?;
@@ -70,7 +70,7 @@ impl super::Service {
),
user_id,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.await?;
} else {
@@ -84,7 +84,7 @@ impl super::Service {
),
server_user,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.await?;
}
@@ -116,7 +116,7 @@ impl super::Service {
PartialPdu::state(String::new(), &room_power_levels),
server_user,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.await?;
@@ -196,7 +196,7 @@ impl super::Service {
PartialPdu::state(String::new(), &room_power_levels),
self.services.globals.server_user.as_ref(),
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.await
.inspect_err(|e| {
@@ -243,7 +243,7 @@ impl super::Service {
PartialPdu::state(user_id.to_string(), &member_content),
self.services.globals.server_user.as_ref(),
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.await
.map(|_| ())
+4 -11
View File
@@ -32,7 +32,7 @@ use tokio::sync::RwLock;
use crate::{
Dep, account_data, config, globals,
media::{MXC_LENGTH, mxc::Mxc},
rooms::{self, state::RoomMutexGuard},
rooms,
};
pub struct Service {
@@ -492,7 +492,6 @@ impl Service {
) -> Result<()> {
assert!(self.user_is_admin(user_id).await, "sender is not admin");
let state_lock = self.services.state.mutex.lock(room_id).await;
if let Err(e) = self
.services
.timeline
@@ -500,11 +499,11 @@ impl Service {
PartialPdu::timeline(&self.text_or_file(content).await),
user_id,
Some(room_id),
&state_lock,
None,
)
.await
{
self.handle_response_error(e, room_id, user_id, &state_lock)
self.handle_response_error(e, room_id, user_id)
.boxed()
.await
.unwrap_or_else(default_log);
@@ -518,7 +517,6 @@ impl Service {
e: Error,
room_id: &RoomId,
user_id: &UserId,
state_lock: &RoomMutexGuard,
) -> Result<()> {
error!("Failed to build and append admin room response PDU: \"{e}\"");
let content = RoomMessageEventContent::text_plain(format!(
@@ -528,12 +526,7 @@ impl Service {
self.services
.timeline
.build_and_append_pdu(
PartialPdu::timeline(&content),
user_id,
Some(room_id),
state_lock,
)
.build_and_append_pdu(PartialPdu::timeline(&content), user_id, Some(room_id), None)
.await?;
Ok(())
@@ -325,7 +325,6 @@ impl super::Service {
trace!(%user_id, "user does not have power level to send dummy event, skipping");
continue;
}
let state_lock = self.services.state.mutex.lock(room_id).await;
if self
.services
.timeline
@@ -336,7 +335,7 @@ impl super::Service {
},
&user_id,
Some(room_id),
&state_lock,
None,
)
.await
.inspect(|_| debug!(sender=%user_id, "Successfully sent a dummy event"))
@@ -60,6 +60,14 @@ impl super::Service {
let timer = Instant::now();
let min_depth = self.services.metadata.get_mindepth(room_id).await;
let room_version_rules = get_room_version_rules(create_event)?;
// We lock the room here to prevent the current state from changing
// beneath us mid-check.
let state_lock = if incoming_pdu.state_key.is_some() || incoming_pdu.prev_events.len() > 1
{
Some(self.services.state.mutex.lock(room_id).await)
} else {
None
};
// We now need to resolve the state before the event so that we can
// perform PDU check 5 (event auth passes based on state before the
@@ -81,14 +89,6 @@ impl super::Service {
// that it passes based on the *current* room state (state across all
// forward extremities). If it doesn't, we accept it, but soft-fail
// it, and this prevents it being promoted.
// We lock the room here to prevent the current state from changing
// beneath us mid-check.
trace!(
room_id = %room_id,
"Locking the room"
);
let state_lock = self.services.state.mutex.lock(room_id).await;
let passes_current_state = self
.current_state_check_6(&incoming_pdu, &room_version_rules, create_event)
.await
@@ -175,9 +175,9 @@ impl super::Service {
&room_version_rules,
state_before,
should_soft_fail,
&state_lock,
)
.await?;
drop(state_lock);
if should_soft_fail {
debug_info!(
@@ -193,7 +193,6 @@ impl super::Service {
}
// Event has passed all auth/stateres checks
drop(state_lock);
if incoming_pdu.depth > min_depth && incoming_pdu.state_key().is_some() {
self.services
.metadata
+3 -3
View File
@@ -291,7 +291,7 @@ impl Service {
PartialPdu::state(sender_user.to_string(), &content),
sender_user,
Some(room_id),
state_lock,
Some(&state_lock),
)
.await?;
@@ -664,7 +664,7 @@ impl Service {
debug!("Forcing state for new room");
self.services
.state
.force_state(room_id, statehash_before_join, added, removed, state_lock)
.force_state(room_id, statehash_before_join, added, removed)
.await?;
debug!("Updating joined counts for new room");
@@ -701,7 +701,7 @@ impl Service {
// exist
self.services
.state
.set_room_state(room_id, statehash_after_join, state_lock);
.set_room_state(room_id, statehash_after_join);
if !resident_before {
// NOTE: We replace local extremities for this room if we were not a
// resident before. We might be doing a remote join to satisfy
+3 -10
View File
@@ -102,7 +102,6 @@ impl Service {
shortstatehash: u64,
statediffnew: Arc<CompressedState>,
_statediffremoved: Arc<CompressedState>,
state_lock: &RoomMutexGuard,
) -> Result {
let event_ids = statediffnew
.iter()
@@ -139,7 +138,7 @@ impl Service {
self.services.state_cache.update_joined_count(room_id).await;
self.set_room_state(room_id, shortstatehash, state_lock);
self.set_room_state(room_id, shortstatehash);
self.services.sync.wake_all_joined(room_id).await;
@@ -363,14 +362,8 @@ impl Service {
}
/// Set the state hash to a new version, but does not update state_cache.
#[tracing::instrument(skip(self, _mutex_lock), level = "debug")]
pub fn set_room_state(
&self,
room_id: &RoomId,
shortstatehash: u64,
// Take mutex guard to make sure users get the room state mutex
_mutex_lock: &RoomMutexGuard,
) {
#[tracing::instrument(skip(self), level = "debug")]
pub fn set_room_state(&self, room_id: &RoomId, shortstatehash: u64) {
const BUFSIZE: usize = size_of::<u64>();
self.db
+1 -1
View File
@@ -146,7 +146,7 @@ impl super::Service {
),
sender,
Some(room_id),
state_lock,
Some(state_lock),
)
.await
.is_ok()
+5 -12
View File
@@ -14,7 +14,6 @@ use conduwuit::{
trace,
utils::{
IterStream, TryFutureExtExt,
mutex_map::Guard,
stream::{BroadbandExt, ReadyExt},
},
warn,
@@ -27,7 +26,7 @@ use conduwuit_core::{
};
use futures::{FutureExt, StreamExt, TryFutureExt, pin_mut};
use ruma::{
CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, OwnedServerName,
CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedServerName,
RoomId, RoomVersionId, ServerName, UserId,
events::{
AnySyncTimelineEvent, GlobalAccountDataEventType, TimelineEventType,
@@ -43,7 +42,7 @@ use ruma::{
serde::Raw,
};
use super::{ExtractBody, ExtractRelatesTo, ExtractRelatesToEventId, RoomMutexGuard};
use super::{ExtractBody, ExtractRelatesTo, ExtractRelatesToEventId};
use crate::{
appservice::RegistrationInfo,
rooms::state_compressor::{CompressedState, HashSetCompressStateEvent},
@@ -59,7 +58,6 @@ impl super::Service {
room_version_rules: &RoomVersionRules,
state_before: HashMap<u64, OwnedEventId>,
should_soft_fail: bool,
state_lock: &'a RoomMutexGuard,
) -> Result<Option<RawPduId>> {
let room_id = pdu.room_id_or_hash();
@@ -104,7 +102,6 @@ impl super::Service {
.get_forward_extremities(&room_id)
.collect()
.await,
state_lock,
)
.await?;
@@ -113,7 +110,6 @@ impl super::Service {
pdu,
pdu_json,
extremities.iter().map(Borrow::borrow),
state_lock,
&room_id,
)
.await?;
@@ -166,11 +162,10 @@ impl super::Service {
room_version_rules: &RoomVersionRules,
state_before: HashMap<u64, OwnedEventId>,
current_extremities: Vec<OwnedEventId>,
state_lock: &Guard<OwnedRoomId, ()>,
) -> Result<Vec<OwnedEventId>> {
if incoming_pdu.state_key().is_some() {
debug!("Event is a state-event. Deriving new room state");
self.derive_new_state(incoming_pdu, room_version_rules, state_before, state_lock)
self.derive_new_state(incoming_pdu, room_version_rules, state_before)
.await?;
}
@@ -269,7 +264,6 @@ impl super::Service {
incoming_pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
state_before: HashMap<u64, OwnedEventId>,
state_lock: &Guard<OwnedRoomId, ()>,
) -> Result {
let room_id = incoming_pdu.room_id_or_hash();
// We also add state after incoming event to the fork states
@@ -302,7 +296,7 @@ impl super::Service {
self.services
.state
.force_state(&room_id, shortstatehash, added, removed, state_lock)
.force_state(&room_id, shortstatehash, added, removed)
.await
}
@@ -352,7 +346,6 @@ impl super::Service {
pdu: &'a PduEvent,
mut pdu_json: CanonicalJsonObject,
leaves: Leaves,
state_lock: &'a RoomMutexGuard,
room_id: &'a RoomId,
) -> Result<RawPduId>
where
@@ -387,7 +380,7 @@ impl super::Service {
trace!("setting forward extremities");
self.services
.state
.set_forward_extremities(room_id, leaves, state_lock)
.set_forward_extremities(room_id, leaves)
.await;
let insert_lock = self.mutex_insert.lock(room_id).await;
+2 -4
View File
@@ -30,7 +30,7 @@ impl super::Service {
partial_pdu: PartialPdu,
sender: &UserId,
room_id: Option<&RoomId>,
state_lock: &RoomMutexGuard,
state_lock: Option<&RoomMutexGuard>,
) -> Result<OwnedEventId> {
let (pdu, pdu_json) = self
.create_hash_and_sign_event(partial_pdu, sender, room_id, state_lock)
@@ -164,9 +164,7 @@ impl super::Service {
// a moment in time where events in the current room state do not
// exist
trace!("Setting room state for room {room_id}");
self.services
.state
.set_room_state(&room_id, statehashid, state_lock);
self.services.state.set_room_state(&room_id, statehashid);
self.services.sync.wake_all_joined(&room_id).await;
+5 -2
View File
@@ -84,7 +84,7 @@ impl super::Service {
partial_pdu: PartialPdu,
sender: &UserId,
room_id: Option<&RoomId>,
_mutex_lock: &RoomMutexGuard,
mutex_lock: Option<&RoomMutexGuard>,
) -> Result<(PduEvent, RoomVersionRules)> {
let PartialPdu {
event_type,
@@ -95,6 +95,9 @@ impl super::Service {
redacts,
timestamp,
} = partial_pdu;
if state_key.is_some() {
debug_assert!(mutex_lock.is_some(), "No state mutex held while creating state event");
}
trace!(
"Creating event of type {} in room {}",
@@ -287,7 +290,7 @@ impl super::Service {
partial_pdu: PartialPdu,
sender: &UserId,
room_id: Option<&RoomId>,
mutex_lock: &RoomMutexGuard,
mutex_lock: Option<&RoomMutexGuard>,
) -> Result<(PduEvent, CanonicalJsonObject)> {
if !self.services.globals.user_is_local(sender) {
return Err!(Request(Forbidden("Sender must be a local user")));
+1 -1
View File
@@ -61,7 +61,7 @@ impl super::Service {
},
sender,
Some(room_id),
state_lock,
Some(state_lock),
)
.await?;
+1 -1
View File
@@ -191,7 +191,7 @@ impl super::Service {
PartialPdu::state(user_id.to_string(), &current_membership),
user_id,
Some(&room_id),
&state_lock,
Some(&state_lock),
)
.await;
}