feat: Pause IO for BackupProvider (#4182)

This makes the BackupProvider automatically invoke pause-io while it
is needed.

It needed to make the guard independent from the Context lifetime to
make this work.  Which is a bit sad.
This commit is contained in:
Floris Bruynooghe
2023-03-20 19:57:17 +01:00
committed by GitHub
parent e04efdbd94
commit 35f50a8965
4 changed files with 21 additions and 23 deletions

View File

@@ -91,7 +91,7 @@ pub async fn imex(
let cancel = context.alloc_ongoing().await?;
let res = {
let mut guard = context.scheduler.pause(context).await;
let mut guard = context.scheduler.pause(context.clone()).await;
let res = imex_inner(context, what, path, passphrase)
.race(async {
cancel.recv().await.ok();

View File

@@ -91,6 +91,7 @@ impl BackupProvider {
// Acquire global "ongoing" mutex.
let cancel_token = context.alloc_ongoing().await?;
let mut paused_guard = context.scheduler.pause(context.clone()).await;
let context_dir = context
.get_blobdir()
.parent()
@@ -118,15 +119,19 @@ impl BackupProvider {
Ok((provider, ticket)) => (provider, ticket),
Err(err) => {
context.free_ongoing().await;
paused_guard.resume().await;
return Err(err);
}
};
let handle = tokio::spawn(Self::watch_provider(
context.clone(),
provider,
cancel_token,
dbfile,
));
let handle = {
let context = context.clone();
tokio::spawn(async move {
let res = Self::watch_provider(&context, provider, cancel_token, dbfile).await;
context.free_ongoing().await;
paused_guard.resume().await;
res
})
};
let slf = Self { handle, ticket };
let qr = slf.qr();
*context.export_provider.lock().expect("poisoned lock") = Some(qr);
@@ -181,7 +186,7 @@ impl BackupProvider {
/// The *cancel_token* is the handle for the ongoing process mutex, when this completes
/// we must cancel this operation.
async fn watch_provider(
context: Context,
context: &Context,
mut provider: Provider,
cancel_token: Receiver<()>,
_dbfile: TempPathGuard,
@@ -262,7 +267,6 @@ impl BackupProvider {
context.emit_event(SendProgress::Failed.into())
}
}
context.free_ongoing().await;
res
}
@@ -373,7 +377,7 @@ pub async fn get_backup(context: &Context, qr: Qr) -> Result<()> {
!context.is_configured().await?,
"Cannot import backups to accounts in use."
);
let mut guard = context.scheduler.pause(context).await;
let mut guard = context.scheduler.pause(context.clone()).await;
// Acquire global "ongoing" mutex.
let cancel_token = context.alloc_ongoing().await?;

View File

@@ -95,10 +95,10 @@ impl SchedulerState {
/// If in the meantime [`SchedulerState::start`] or [`SchedulerState::stop`] is called
/// resume will do the right thing and restore the scheduler to the state requested by
/// the last call.
pub(crate) async fn pause<'a>(&'_ self, context: &'a Context) -> IoPausedGuard<'a> {
pub(crate) async fn pause<'a>(&'_ self, context: Context) -> IoPausedGuard {
let mut inner = self.inner.write().await;
inner.paused = true;
Self::do_stop(inner, context).await;
Self::do_stop(inner, &context).await;
IoPausedGuard {
context,
done: false,
@@ -195,12 +195,12 @@ struct InnerSchedulerState {
}
#[derive(Debug)]
pub(crate) struct IoPausedGuard<'a> {
context: &'a Context,
pub(crate) struct IoPausedGuard {
context: Context,
done: bool,
}
impl<'a> IoPausedGuard<'a> {
impl IoPausedGuard {
pub(crate) async fn resume(&mut self) {
self.done = true;
let mut inner = self.context.scheduler.inner.write().await;
@@ -211,7 +211,7 @@ impl<'a> IoPausedGuard<'a> {
}
}
impl<'a> Drop for IoPausedGuard<'a> {
impl Drop for IoPausedGuard {
fn drop(&mut self) {
if self.done {
return;