Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Remove random pauses to prevent the database from deadlocking

Browse source
This commit is contained in:
Antonio Scandurra 2022-12-05 12:03:45 +01:00
parent d97a8364ad
commit d3c411677a
2 changed files with 62 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

108
crates/collab/src/db.rs
View file

@ -2131,31 +2131,35 @@ impl Database {
F: Send + Fn(TransactionHandle) -> Fut, F: Send + Fn(TransactionHandle) -> Fut,
Fut: Send + Future<Output = Result<T>>, Fut: Send + Future<Output = Result<T>>,
{ {
loop { let body = async {
let (tx, result) = self.run(self.with_transaction(&f)).await?; loop {
match result { let (tx, result) = self.with_transaction(&f).await?;
Ok(result) => { match result {
match self.run(async move { Ok(tx.commit().await?) }).await { Ok(result) => {
Ok(()) => return Ok(result), match tx.commit().await.map_err(Into::into) {
Err(error) => { Ok(()) => return Ok(result),
if is_serialization_error(&error) { Err(error) => {
// Retry (don't break the loop) if is_serialization_error(&error) {
} else { // Retry (don't break the loop)
return Err(error); } else {
return Err(error);
}
} }
} }
} }
} Err(error) => {
Err(error) => { tx.rollback().await?;
self.run(tx.rollback()).await?; if is_serialization_error(&error) {
if is_serialization_error(&error) { // Retry (don't break the loop)
// Retry (don't break the loop) } else {
} else { return Err(error);
return Err(error); }
} }
} }
} }
} };
self.run(body).await
} }
async fn room_transaction<F, Fut, T>(&self, f: F) -> Result<RoomGuard<T>> async fn room_transaction<F, Fut, T>(&self, f: F) -> Result<RoomGuard<T>>
@ -2163,39 +2167,43 @@ impl Database {
F: Send + Fn(TransactionHandle) -> Fut, F: Send + Fn(TransactionHandle) -> Fut,
Fut: Send + Future<Output = Result<(RoomId, T)>>, Fut: Send + Future<Output = Result<(RoomId, T)>>,
{ {
loop { let body = async {
let (tx, result) = self.run(self.with_transaction(&f)).await?; loop {
match result { let (tx, result) = self.with_transaction(&f).await?;
Ok((room_id, data)) => { match result {
let lock = self.rooms.entry(room_id).or_default().clone(); Ok((room_id, data)) => {
let _guard = lock.lock_owned().await; let lock = self.rooms.entry(room_id).or_default().clone();
match self.run(async move { Ok(tx.commit().await?) }).await { let _guard = lock.lock_owned().await;
Ok(()) => { match tx.commit().await.map_err(Into::into) {
return Ok(RoomGuard { Ok(()) => {
data, return Ok(RoomGuard {
_guard, data,
_not_send: PhantomData, _guard,
}); _not_send: PhantomData,
} });
Err(error) => { }
if is_serialization_error(&error) { Err(error) => {
// Retry (don't break the loop) if is_serialization_error(&error) {
} else { // Retry (don't break the loop)
return Err(error); } else {
return Err(error);
}
} }
} }
} }
} Err(error) => {
Err(error) => { tx.rollback().await?;
self.run(tx.rollback()).await?; if is_serialization_error(&error) {
if is_serialization_error(&error) { // Retry (don't break the loop)
// Retry (don't break the loop) } else {
} else { return Err(error);
return Err(error); }
} }
} }
} }
} };
self.run(body).await
} }
async fn with_transaction<F, Fut, T>(&self, f: &F) -> Result<(DatabaseTransaction, Result<T>)> async fn with_transaction<F, Fut, T>(&self, f: &F) -> Result<(DatabaseTransaction, Result<T>)>
@ -2233,13 +2241,7 @@ impl Database {
background.simulate_random_delay().await; background.simulate_random_delay().await;
} }
let result = self.runtime.as_ref().unwrap().block_on(future); self.runtime.as_ref().unwrap().block_on(future)
if let Some(background) = self.background.as_ref() {
background.simulate_random_delay().await;
}
result
} }
#[cfg(not(test))] #[cfg(not(test))]

8
crates/collab/src/integration_tests.rs
View file

@ -5672,7 +5672,13 @@ impl TestServer {
async fn start(background: Arc<executor::Background>) -> Self { async fn start(background: Arc<executor::Background>) -> Self {
static NEXT_LIVE_KIT_SERVER_ID: AtomicUsize = AtomicUsize::new(0); static NEXT_LIVE_KIT_SERVER_ID: AtomicUsize = AtomicUsize::new(0);
let test_db = TestDb::sqlite(background.clone()); let use_postgres = env::var("USE_POSTGRES").ok();
let use_postgres = use_postgres.as_deref();
let test_db = if use_postgres == Some("true") || use_postgres == Some("1") {
TestDb::postgres(background.clone())
} else {
TestDb::sqlite(background.clone())
};
let live_kit_server_id = NEXT_LIVE_KIT_SERVER_ID.fetch_add(1, SeqCst); let live_kit_server_id = NEXT_LIVE_KIT_SERVER_ID.fetch_add(1, SeqCst);
let live_kit_server = live_kit_client::TestServer::create( let live_kit_server = live_kit_client::TestServer::create(
format!("http://livekit.{}.test", live_kit_server_id), format!("http://livekit.{}.test", live_kit_server_id),