Skip to content

Workspace store deadlock protection #9132

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Dec 13, 2024
Merged

Workspace store deadlock protection #9132

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5cf90d8
Add `libparsec_platform_async::try_task_id()`
touilleMan Dec 6, 2024
3233fd5
Improve `WorkspaceStore` to prevent deadlocks (including runtime deal…
touilleMan Dec 6, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 37 additions & 30 deletions libparsec/crates/client/src/workspace/store/cache.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,17 +193,20 @@ pub(super) async fn populate_cache_from_local_storage(
) -> Result<ArcLocalChildManifest, PopulateCacheFromLocalStorageError> {
// 1) Local storage lookup

let maybe_found = {
let mut maybe_storage = store.storage.lock().await;
let storage = match &mut *maybe_storage {
None => return Err(PopulateCacheFromLocalStorageError::Stopped),
Some(storage) => storage,
};

storage.get_manifest(entry_id).await.map_err(|err| {
PopulateCacheFromLocalStorageError::Internal(err.context("cannot access local storage"))
})?
};
let maybe_found = store
.data
.with_storage(|maybe_storage| async move {
let storage = maybe_storage
.as_mut()
.ok_or_else(|| PopulateCacheFromLocalStorageError::Stopped)?;

storage.get_manifest(entry_id).await.map_err(|err| {
PopulateCacheFromLocalStorageError::Internal(
err.context("cannot access local storage"),
)
})
})
.await?;

let manifest = match maybe_found {
Some(encrypted) => {
Expand All @@ -230,8 +233,9 @@ pub(super) async fn populate_cache_from_local_storage(

// 2) We got our manifest, don't forget to update the cache before returning it

let mut cache = store.current_view_cache.lock().expect("Mutex is poisoned");
let manifest = cache.manifests.insert_if_missing(manifest);
let manifest = store
.data
.with_current_view_cache(|cache| cache.manifests.insert_if_missing(manifest));

Ok(manifest)
}
Expand Down Expand Up @@ -368,21 +372,23 @@ pub(super) async fn populate_cache_from_local_storage_or_server(
encrypted: manifest.dump_and_encrypt(&store.device.local_symkey),
},
};
let outcome = {
let mut maybe_storage = store.storage.lock().await;
let storage = match &mut *maybe_storage {
None => return Err(PopulateCacheFromLocalStorageOrServerError::Stopped),
Some(storage) => storage,
};
storage
.populate_manifest(&update_data)
.await
.map_err(|err| {
PopulateCacheFromLocalStorageOrServerError::Internal(
err.context("cannot populate local storage with manifest"),
)
})?
};
let outcome = store
.data
.with_storage(|maybe_storage| async move {
let storage = maybe_storage
.as_mut()
.ok_or_else(|| PopulateCacheFromLocalStorageOrServerError::Stopped)?;

storage
.populate_manifest(&update_data)
.await
.map_err(|err| {
PopulateCacheFromLocalStorageOrServerError::Internal(
err.context("cannot populate local storage with manifest"),
)
})
})
.await?;
match outcome {
PopulateManifestOutcome::Stored => break manifest,
// A concurrent operation has populated the local storage !
Expand Down Expand Up @@ -424,8 +430,9 @@ pub(super) async fn populate_cache_from_local_storage_or_server(

// 4) Also update the cache

let mut cache = store.current_view_cache.lock().expect("Mutex is poisoned");
let manifest = cache.manifests.insert_if_missing(manifest);
let manifest = store
.data
.with_current_view_cache(|cache| cache.manifests.insert_if_missing(manifest));

Ok(manifest)
}
97 changes: 57 additions & 40 deletions libparsec/crates/client/src/workspace/store/file_updater.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

use std::sync::Arc;

use libparsec_platform_async::event::EventListener;
use libparsec_types::prelude::*;

use crate::certif::{InvalidCertificateError, InvalidKeysBundleError, InvalidManifestError};
Expand Down Expand Up @@ -49,8 +50,9 @@ pub(super) async fn for_update_file(
// Guard's drop will panic if the lock is not released
macro_rules! release_guard_on_error {
($update_guard:expr) => {
let mut cache_guard = store.current_view_cache.lock().expect("Mutex is poisoned");
cache_guard.lock_update_manifests.release($update_guard);
store.data.with_current_view_cache(|cache| {
cache.lock_update_manifests.release($update_guard);
});
};
}

Expand All @@ -62,36 +64,49 @@ pub(super) async fn for_update_file(
listener.await;
}

let update_guard = {
let mut cache_guard = store.current_view_cache.lock().expect("Mutex is poisoned");

// 1) Lock for update
// 1) Lock for update

let outcome = cache_guard.lock_update_manifests.take(entry_id);
match outcome {
enum LockForUpdateOutcome {
WaitAndRetryStep1(EventListener),
// There is no GoToStep2 because it's the dispatching step
GoToStep3(ManifestUpdateLockGuard),
GoToStep4((ManifestUpdateLockGuard, ArcLocalChildManifest)),
}
let outcome = store.data.with_current_view_cache(|cache| {
match cache.lock_update_manifests.take(entry_id) {
ManifestUpdateLockTakeOutcome::Taken(update_guard) => {
// 2) Cache lookup for entry...

let found = cache_guard.manifests.get(&entry_id);
let found = cache.manifests.get(&entry_id);
if let Some(manifest) = found {
// Cache hit ! We go to step 4.
break (update_guard, manifest.clone());
LockForUpdateOutcome::GoToStep4((update_guard, manifest.clone()))
} else {
// The entry is not in cache, go to step 3 for a lookup in the local storage.
// Note we keep the update lock: this has no impact on read operation, and
// any other write operation taking the lock will have no choice but to try
// to populate the cache just like we are going to do.
LockForUpdateOutcome::GoToStep3(update_guard)
}
// The entry is not in cache, go to step 3 for a lookup in the local storage.
// Note we keep the update lock: this has no impact on read operation, and
// any other write operation taking the lock will have no choice but to try
// to populate the cache just like we are going to do.
update_guard
}

ManifestUpdateLockTakeOutcome::NeedWait(listener) => {
if !wait {
return Err(ForUpdateFileError::WouldBlock);
}
maybe_need_wait = Some(listener);
continue;
LockForUpdateOutcome::WaitAndRetryStep1(listener)
}
}
});
let update_guard = match outcome {
LockForUpdateOutcome::GoToStep3(update_guard) => update_guard,
LockForUpdateOutcome::GoToStep4((update_guard, manifest)) => {
break (update_guard, manifest)
}
LockForUpdateOutcome::WaitAndRetryStep1(listener) => {
if !wait {
return Err(ForUpdateFileError::WouldBlock);
}
maybe_need_wait = Some(listener);
continue;
}
};

// Be careful here: `update_guard` must be manually released in case of error !
Expand Down Expand Up @@ -143,7 +158,7 @@ pub(super) async fn for_update_file(
};

let updater = FileUpdater {
_update_guard: update_guard,
update_guard,
#[cfg(debug_assertions)]
entry_id: manifest.base.id,
};
Expand All @@ -159,7 +174,7 @@ pub(super) async fn for_update_file(
/// keep hold on the store.
#[derive(Debug)]
pub(crate) struct FileUpdater {
_update_guard: ManifestUpdateLockGuard,
update_guard: ManifestUpdateLockGuard,
#[cfg(debug_assertions)]
entry_id: VlobID,
}
Expand All @@ -176,39 +191,41 @@ impl FileUpdater {
#[cfg(debug_assertions)]
assert_eq!(manifest.base.id, self.entry_id);

let mut storage_guard = store.storage.lock().await;
let storage = storage_guard
.as_mut()
.ok_or_else(|| UpdateFileManifestAndContinueError::Stopped)?;

let update_data = UpdateManifestData {
entry_id: manifest.base.id,
base_version: manifest.base.version,
need_sync: manifest.need_sync,
encrypted: manifest.dump_and_encrypt(&store.device.local_symkey),
};

let new_chunks = new_chunks
.map(|(chunk_id, cleartext)| (chunk_id, store.device.local_symkey.encrypt(cleartext)));
storage
.update_manifest_and_chunks(&update_data, new_chunks, removed_chunks)
store
.data
.with_storage(|maybe_storage| async move {
let storage = maybe_storage
.as_mut()
.ok_or_else(|| UpdateFileManifestAndContinueError::Stopped)?;

storage
.update_manifest_and_chunks(&update_data, new_chunks, removed_chunks)
.await
.map_err(UpdateFileManifestAndContinueError::Internal)
})
.await?;

// Finally update cache
let mut cache = store.current_view_cache.lock().expect("Mutex is poisoned");
cache
.manifests
.insert(ArcLocalChildManifest::File(manifest));
store.data.with_current_view_cache(|cache| {
cache
.manifests
.insert(ArcLocalChildManifest::File(manifest));
});

Ok(())
}

pub fn close(self, store: &super::WorkspaceStore) {
store
.current_view_cache
.lock()
.expect("Mutex is poisoned")
.lock_update_manifests
.release(self._update_guard);
store.data.with_current_view_cache(|cache| {
cache.lock_update_manifests.release(self.update_guard);
});
}
}
Loading
Loading