backups/grapevine
1
0
Fork 0
mirror of https://gitlab.computer.surgery/matrix/grapevine.git synced 2025-12-17 07:41:23 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Use TokenSet for roomid_mutex_insert

Browse source
This commit is contained in:
Lambda 2024-06-23 20:07:54 +00:00
parent 34ccb2cd06
commit 4893c54f4f
3 changed files with 34 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

52
src/api/client_server/sync.rs
View file

@ -1,6 +1,5 @@
use std::{ use std::{
collections::{BTreeMap, BTreeSet, HashMap, HashSet}, collections::{BTreeMap, BTreeSet, HashMap, HashSet},
sync::Arc,
time::Duration, time::Duration,
}; };
@ -161,17 +160,12 @@ pub(crate) async fn sync_events_route(
{ {
// Get and drop the lock to wait for remaining operations to finish // Get and drop the lock to wait for remaining operations to finish
let mutex_insert = Arc::clone( let room_token = services()
services() .globals
.globals .roomid_mutex_insert
.roomid_mutex_insert .lock_key(room_id.clone())
.write() .await;
.await drop(room_token);
.entry(room_id.clone())
.or_default(),
);
let insert_lock = mutex_insert.lock().await;
drop(insert_lock);
} }
let left_count = services() let left_count = services()
@ -330,17 +324,12 @@ pub(crate) async fn sync_events_route(
{ {
// Get and drop the lock to wait for remaining operations to finish // Get and drop the lock to wait for remaining operations to finish
let mutex_insert = Arc::clone( let room_token = services()
services() .globals
.globals .roomid_mutex_insert
.roomid_mutex_insert .lock_key(room_id.clone())
.write() .await;
.await drop(room_token);
.entry(room_id.clone())
.or_default(),
);
let insert_lock = mutex_insert.lock().await;
drop(insert_lock);
} }
let invite_count = services() let invite_count = services()
@ -481,17 +470,12 @@ async fn load_joined_room(
{ {
// Get and drop the lock to wait for remaining operations to finish // Get and drop the lock to wait for remaining operations to finish
// This will make sure the we have all events until next_batch // This will make sure the we have all events until next_batch
let mutex_insert = Arc::clone( let room_token = services()
services() .globals
.globals .roomid_mutex_insert
.roomid_mutex_insert .lock_key(room_id.to_owned())
.write() .await;
.await drop(room_token);
.entry(room_id.to_owned())
.or_default(),
);
let insert_lock = mutex_insert.lock().await;
drop(insert_lock);
} }
let (timeline_pdus, limited) = let (timeline_pdus, limited) =

7
src/service/globals.rs
View file

@ -79,8 +79,7 @@ pub(crate) struct Service {
Arc<RwLock<HashMap<OwnedServerName, RateLimitState>>>, Arc<RwLock<HashMap<OwnedServerName, RateLimitState>>>,
pub(crate) servername_ratelimiter: pub(crate) servername_ratelimiter:
OnDemandHashMap<OwnedServerName, Semaphore>, OnDemandHashMap<OwnedServerName, Semaphore>,
pub(crate) roomid_mutex_insert: pub(crate) roomid_mutex_insert: TokenSet<OwnedRoomId, marker::Insert>,
RwLock<HashMap<OwnedRoomId, Arc<Mutex<()>>>>,
pub(crate) roomid_mutex_state: TokenSet<OwnedRoomId, marker::State>, pub(crate) roomid_mutex_state: TokenSet<OwnedRoomId, marker::State>,
// this lock will be held longer // this lock will be held longer
@ -279,7 +278,9 @@ impl Service {
"servername_ratelimiter".to_owned(), "servername_ratelimiter".to_owned(),
), ),
roomid_mutex_state: TokenSet::new("roomid_mutex_state".to_owned()), roomid_mutex_state: TokenSet::new("roomid_mutex_state".to_owned()),
roomid_mutex_insert: RwLock::new(HashMap::new()), roomid_mutex_insert: TokenSet::new(
"roomid_mutex_insert".to_owned(),
),
roomid_mutex_federation: RwLock::new(HashMap::new()), roomid_mutex_federation: RwLock::new(HashMap::new()),
roomid_federationhandletime: RwLock::new(HashMap::new()), roomid_federationhandletime: RwLock::new(HashMap::new()),
stateres_mutex: Arc::new(Mutex::new(())), stateres_mutex: Arc::new(Mutex::new(())),

34
src/service/rooms/timeline.rs
View file

@ -259,16 +259,11 @@ impl Service {
.mark_as_referenced(&pdu.room_id, &pdu.prev_events)?; .mark_as_referenced(&pdu.room_id, &pdu.prev_events)?;
services().rooms.state.set_forward_extremities(room_id, leaves)?; services().rooms.state.set_forward_extremities(room_id, leaves)?;
let mutex_insert = Arc::clone( let insert_token = services()
services() .globals
.globals .roomid_mutex_insert
.roomid_mutex_insert .lock_key(pdu.room_id.clone())
.write() .await;
.await
.entry(pdu.room_id.clone())
.or_default(),
);
let insert_lock = mutex_insert.lock().await;
let count1 = services().globals.next_count()?; let count1 = services().globals.next_count()?;
// Mark as read first so the sending client doesn't get a notification // Mark as read first so the sending client doesn't get a notification
@ -290,7 +285,7 @@ impl Service {
// Insert pdu // Insert pdu
self.db.append_pdu(&pdu_id, pdu, &pdu_json, count2)?; self.db.append_pdu(&pdu_id, pdu, &pdu_json, count2)?;
drop(insert_lock); drop(insert_token);
// See if the event matches any known pushers // See if the event matches any known pushers
let power_levels: RoomPowerLevelsEventContent = services() let power_levels: RoomPowerLevelsEventContent = services()
@ -1328,16 +1323,11 @@ impl Service {
.get_shortroomid(&room_id)? .get_shortroomid(&room_id)?
.expect("room exists"); .expect("room exists");
let mutex_insert = Arc::clone( let insert_token = services()
services() .globals
.globals .roomid_mutex_insert
.roomid_mutex_insert .lock_key(room_id.clone())
.write() .await;
.await
.entry(room_id.clone())
.or_default(),
);
let insert_lock = mutex_insert.lock().await;
let count = services().globals.next_count()?; let count = services().globals.next_count()?;
let mut pdu_id = shortroomid.to_be_bytes().to_vec(); let mut pdu_id = shortroomid.to_be_bytes().to_vec();
@ -1347,7 +1337,7 @@ impl Service {
// Insert pdu // Insert pdu
self.db.prepend_backfill_pdu(&pdu_id, &event_id, &value)?; self.db.prepend_backfill_pdu(&pdu_id, &event_id, &value)?;
drop(insert_lock); drop(insert_token);
if pdu.kind == TimelineEventType::RoomMessage { if pdu.kind == TimelineEventType::RoomMessage {
#[derive(Deserialize)] #[derive(Deserialize)]