idling works!!!

This commit is contained in:
Quentin 2024-01-18 17:33:57 +01:00
parent e1161cab0e
commit 185033c462
Signed by: quentin
GPG key ID: E9602264D639FF68
6 changed files with 72 additions and 62 deletions

View file

@ -2,7 +2,7 @@ use std::sync::{Arc, Weak};
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use anyhow::{anyhow, bail, Result}; use anyhow::{anyhow, bail, Result};
use log::{debug, error, info}; use log::error;
use rand::prelude::*; use rand::prelude::*;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use tokio::sync::{watch, Notify}; use tokio::sync::{watch, Notify};
@ -84,21 +84,21 @@ impl<S: BayouState> Bayou<S> {
// 1. List checkpoints // 1. List checkpoints
let checkpoints = self.list_checkpoints().await?; let checkpoints = self.list_checkpoints().await?;
debug!("(sync) listed checkpoints: {:?}", checkpoints); tracing::debug!("(sync) listed checkpoints: {:?}", checkpoints);
// 2. Load last checkpoint if different from currently used one // 2. Load last checkpoint if different from currently used one
let checkpoint = if let Some((ts, key)) = checkpoints.last() { let checkpoint = if let Some((ts, key)) = checkpoints.last() {
if *ts == self.checkpoint.0 { if *ts == self.checkpoint.0 {
(*ts, None) (*ts, None)
} else { } else {
debug!("(sync) loading checkpoint: {}", key); tracing::debug!("(sync) loading checkpoint: {}", key);
let buf = self let buf = self
.storage .storage
.blob_fetch(&storage::BlobRef(key.to_string())) .blob_fetch(&storage::BlobRef(key.to_string()))
.await? .await?
.value; .value;
debug!("(sync) checkpoint body length: {}", buf.len()); tracing::debug!("(sync) checkpoint body length: {}", buf.len());
let ck = open_deserialize::<S>(&buf, &self.key)?; let ck = open_deserialize::<S>(&buf, &self.key)?;
(*ts, Some(ck)) (*ts, Some(ck))
@ -112,7 +112,7 @@ impl<S: BayouState> Bayou<S> {
} }
if let Some(ck) = checkpoint.1 { if let Some(ck) = checkpoint.1 {
debug!( tracing::debug!(
"(sync) updating checkpoint to loaded state at {:?}", "(sync) updating checkpoint to loaded state at {:?}",
checkpoint.0 checkpoint.0
); );
@ -127,7 +127,7 @@ impl<S: BayouState> Bayou<S> {
// 3. List all operations starting from checkpoint // 3. List all operations starting from checkpoint
let ts_ser = self.checkpoint.0.to_string(); let ts_ser = self.checkpoint.0.to_string();
debug!("(sync) looking up operations starting at {}", ts_ser); tracing::debug!("(sync) looking up operations starting at {}", ts_ser);
let ops_map = self let ops_map = self
.storage .storage
.row_fetch(&storage::Selector::Range { .row_fetch(&storage::Selector::Range {
@ -161,7 +161,7 @@ impl<S: BayouState> Bayou<S> {
} }
} }
ops.sort_by_key(|(ts, _)| *ts); ops.sort_by_key(|(ts, _)| *ts);
debug!("(sync) {} operations", ops.len()); tracing::debug!("(sync) {} operations", ops.len());
if ops.len() < self.history.len() { if ops.len() < self.history.len() {
bail!("Some operations have disappeared from storage!"); bail!("Some operations have disappeared from storage!");
@ -238,20 +238,8 @@ impl<S: BayouState> Bayou<S> {
Ok(()) Ok(())
} }
pub async fn idle_sync(&mut self) -> Result<()> { pub fn notifier(&self) -> std::sync::Weak<Notify> {
tracing::debug!("start idle_sync"); Arc::downgrade(&self.watch.learnt_remote_update)
loop {
tracing::trace!("idle_sync loop");
let fut_notif = self.watch.learnt_remote_update.notified();
if self.last_sync_watch_ct != *self.watch.rx.borrow() {
break
}
fut_notif.await;
}
tracing::trace!("idle_sync done");
self.sync().await?;
Ok(())
} }
/// Applies a new operation on the state. Once this function returns, /// Applies a new operation on the state. Once this function returns,
@ -259,7 +247,7 @@ impl<S: BayouState> Bayou<S> {
/// Make sure to call `.opportunistic_sync()` before doing this, /// Make sure to call `.opportunistic_sync()` before doing this,
/// and even before calculating the `op` argument given here. /// and even before calculating the `op` argument given here.
pub async fn push(&mut self, op: S::Op) -> Result<()> { pub async fn push(&mut self, op: S::Op) -> Result<()> {
debug!("(push) add operation: {:?}", op); tracing::debug!("(push) add operation: {:?}", op);
let ts = Timestamp::after( let ts = Timestamp::after(
self.history self.history
@ -321,18 +309,18 @@ impl<S: BayouState> Bayou<S> {
{ {
Some(i) => i, Some(i) => i,
None => { None => {
debug!("(cp) Oldest operation is too recent to trigger checkpoint"); tracing::debug!("(cp) Oldest operation is too recent to trigger checkpoint");
return Ok(()); return Ok(());
} }
}; };
if i_cp < CHECKPOINT_MIN_OPS { if i_cp < CHECKPOINT_MIN_OPS {
debug!("(cp) Not enough old operations to trigger checkpoint"); tracing::debug!("(cp) Not enough old operations to trigger checkpoint");
return Ok(()); return Ok(());
} }
let ts_cp = self.history[i_cp].0; let ts_cp = self.history[i_cp].0;
debug!( tracing::debug!(
"(cp) we could checkpoint at time {} (index {} in history)", "(cp) we could checkpoint at time {} (index {} in history)",
ts_cp.to_string(), ts_cp.to_string(),
i_cp i_cp
@ -340,13 +328,13 @@ impl<S: BayouState> Bayou<S> {
// Check existing checkpoints: if last one is too recent, don't checkpoint again. // Check existing checkpoints: if last one is too recent, don't checkpoint again.
let existing_checkpoints = self.list_checkpoints().await?; let existing_checkpoints = self.list_checkpoints().await?;
debug!("(cp) listed checkpoints: {:?}", existing_checkpoints); tracing::debug!("(cp) listed checkpoints: {:?}", existing_checkpoints);
if let Some(last_cp) = existing_checkpoints.last() { if let Some(last_cp) = existing_checkpoints.last() {
if (ts_cp.msec as i128 - last_cp.0.msec as i128) if (ts_cp.msec as i128 - last_cp.0.msec as i128)
< CHECKPOINT_INTERVAL.as_millis() as i128 < CHECKPOINT_INTERVAL.as_millis() as i128
{ {
debug!( tracing::debug!(
"(cp) last checkpoint is too recent: {}, not checkpointing", "(cp) last checkpoint is too recent: {}, not checkpointing",
last_cp.0.to_string() last_cp.0.to_string()
); );
@ -354,7 +342,7 @@ impl<S: BayouState> Bayou<S> {
} }
} }
debug!("(cp) saving checkpoint at {}", ts_cp.to_string()); tracing::debug!("(cp) saving checkpoint at {}", ts_cp.to_string());
// Calculate state at time of checkpoint // Calculate state at time of checkpoint
let mut last_known_state = (0, &self.checkpoint.1); let mut last_known_state = (0, &self.checkpoint.1);
@ -370,7 +358,7 @@ impl<S: BayouState> Bayou<S> {
// Serialize and save checkpoint // Serialize and save checkpoint
let cryptoblob = seal_serialize(&state_cp, &self.key)?; let cryptoblob = seal_serialize(&state_cp, &self.key)?;
debug!("(cp) checkpoint body length: {}", cryptoblob.len()); tracing::debug!("(cp) checkpoint body length: {}", cryptoblob.len());
let blob_val = storage::BlobVal::new( let blob_val = storage::BlobVal::new(
storage::BlobRef(format!("{}/checkpoint/{}", self.path, ts_cp.to_string())), storage::BlobRef(format!("{}/checkpoint/{}", self.path, ts_cp.to_string())),
@ -385,7 +373,7 @@ impl<S: BayouState> Bayou<S> {
// Delete blobs // Delete blobs
for (_ts, key) in existing_checkpoints[..last_to_keep].iter() { for (_ts, key) in existing_checkpoints[..last_to_keep].iter() {
debug!("(cp) drop old checkpoint {}", key); tracing::debug!("(cp) drop old checkpoint {}", key);
self.storage self.storage
.blob_rm(&storage::BlobRef(key.to_string())) .blob_rm(&storage::BlobRef(key.to_string()))
.await?; .await?;
@ -440,7 +428,7 @@ struct K2vWatch {
target: storage::RowRef, target: storage::RowRef,
rx: watch::Receiver<storage::RowRef>, rx: watch::Receiver<storage::RowRef>,
propagate_local_update: Notify, propagate_local_update: Notify,
learnt_remote_update: Notify, learnt_remote_update: Arc<Notify>,
} }
impl K2vWatch { impl K2vWatch {
@ -452,7 +440,7 @@ impl K2vWatch {
let (tx, rx) = watch::channel::<storage::RowRef>(target.clone()); let (tx, rx) = watch::channel::<storage::RowRef>(target.clone());
let propagate_local_update = Notify::new(); let propagate_local_update = Notify::new();
let learnt_remote_update = Notify::new(); let learnt_remote_update = Arc::new(Notify::new());
let watch = Arc::new(K2vWatch { target, rx, propagate_local_update, learnt_remote_update }); let watch = Arc::new(K2vWatch { target, rx, propagate_local_update, learnt_remote_update });
@ -466,13 +454,13 @@ impl K2vWatch {
storage: storage::Store, storage: storage::Store,
tx: watch::Sender<storage::RowRef>, tx: watch::Sender<storage::RowRef>,
) { ) {
let mut row = match Weak::upgrade(&self_weak) { let (mut row, remote_update) = match Weak::upgrade(&self_weak) {
Some(this) => this.target.clone(), Some(this) => (this.target.clone(), this.learnt_remote_update.clone()),
None => return, None => return,
}; };
while let Some(this) = Weak::upgrade(&self_weak) { while let Some(this) = Weak::upgrade(&self_weak) {
debug!( tracing::debug!(
"bayou k2v watch bg loop iter ({}, {})", "bayou k2v watch bg loop iter ({}, {})",
this.target.uid.shard, this.target.uid.sort this.target.uid.shard, this.target.uid.sort
); );
@ -491,9 +479,11 @@ impl K2vWatch {
} }
Ok(new_value) => { Ok(new_value) => {
row = new_value.row_ref; row = new_value.row_ref;
if tx.send(row.clone()).is_err() { if let Err(e) = tx.send(row.clone()) {
tracing::warn!(err=?e, "(watch) can't record the new log ref");
break; break;
} }
tracing::debug!(row=?row, "(watch) learnt remote update");
this.learnt_remote_update.notify_waiters(); this.learnt_remote_update.notify_waiters();
} }
} }
@ -505,12 +495,14 @@ impl K2vWatch {
let row_val = storage::RowVal::new(row.clone(), rand); let row_val = storage::RowVal::new(row.clone(), rand);
if let Err(e) = storage.row_insert(vec![row_val]).await if let Err(e) = storage.row_insert(vec![row_val]).await
{ {
error!("Error in bayou k2v watch updater loop: {}", e); tracing::error!("Error in bayou k2v watch updater loop: {}", e);
tokio::time::sleep(Duration::from_secs(30)).await; tokio::time::sleep(Duration::from_secs(30)).await;
} }
} }
); );
} }
info!("bayou k2v watch bg loop exiting"); // unblock listeners
remote_update.notify_waiters();
tracing::info!("bayou k2v watch bg loop exiting");
} }
} }

View file

@ -83,7 +83,7 @@ pub async fn dispatch<'a>(
CommandBody::Idle => { CommandBody::Idle => {
Ok(( Ok((
Response::build().to_req(ctx.req).message("DUMMY command due to anti-pattern in the code").ok()?, Response::build().to_req(ctx.req).message("DUMMY command due to anti-pattern in the code").ok()?,
flow::Transition::Idle(tokio::sync::Notify::new()), flow::Transition::Idle(ctx.req.tag.clone(), tokio::sync::Notify::new()),
)) ))
} }

View file

@ -3,6 +3,7 @@ use std::fmt;
use std::sync::Arc; use std::sync::Arc;
use tokio::sync::Notify; use tokio::sync::Notify;
use imap_codec::imap_types::core::Tag;
use crate::imap::mailbox_view::MailboxView; use crate::imap::mailbox_view::MailboxView;
use crate::mail::user::User; use crate::mail::user::User;
@ -21,7 +22,7 @@ pub enum State {
NotAuthenticated, NotAuthenticated,
Authenticated(Arc<User>), Authenticated(Arc<User>),
Selected(Arc<User>, MailboxView, MailboxPerm), Selected(Arc<User>, MailboxView, MailboxPerm),
Idle(Arc<User>, MailboxView, MailboxPerm, Arc<Notify>), Idle(Arc<User>, MailboxView, MailboxPerm, Tag<'static>, Arc<Notify>),
Logout, Logout,
} }
@ -35,7 +36,7 @@ pub enum Transition {
None, None,
Authenticate(Arc<User>), Authenticate(Arc<User>),
Select(MailboxView, MailboxPerm), Select(MailboxView, MailboxPerm),
Idle(Notify), Idle(Tag<'static>, Notify),
UnIdle, UnIdle,
Unselect, Unselect,
Logout, Logout,
@ -55,10 +56,10 @@ impl State {
(State::Selected(u, _, _) , Transition::Unselect) => { (State::Selected(u, _, _) , Transition::Unselect) => {
State::Authenticated(u.clone()) State::Authenticated(u.clone())
} }
(State::Selected(u, m, p), Transition::Idle(s)) => { (State::Selected(u, m, p), Transition::Idle(t, s)) => {
State::Idle(u, m, p, Arc::new(s)) State::Idle(u, m, p, t, Arc::new(s))
}, },
(State::Idle(u, m, p, _), Transition::UnIdle) => { (State::Idle(u, m, p, _, _), Transition::UnIdle) => {
State::Selected(u, m, p) State::Selected(u, m, p)
}, },
(_, Transition::Logout) => State::Logout, (_, Transition::Logout) => State::Logout,

View file

@ -224,6 +224,12 @@ impl MailboxView {
Ok((summary, conflict_id_or_uid)) Ok((summary, conflict_id_or_uid))
} }
pub async fn idle_sync(&mut self) -> Result<Vec<Body<'static>>> {
self.internal.mailbox.notify().await.upgrade().ok_or(anyhow!("test"))?.notified().await;
self.internal.mailbox.opportunistic_sync().await?;
self.update(UpdateParameters::default()).await
}
pub async fn expunge(&mut self) -> Result<Vec<Body<'static>>> { pub async fn expunge(&mut self) -> Result<Vec<Body<'static>>> {
self.internal.sync().await?; self.internal.sync().await?;
let state = self.internal.peek().await; let state = self.internal.peek().await;

View file

@ -1,4 +1,4 @@
use anyhow::anyhow; use anyhow::{Result, anyhow, bail};
use crate::imap::capability::{ClientCapability, ServerCapability}; use crate::imap::capability::{ClientCapability, ServerCapability};
use crate::imap::command::{anonymous, authenticated, selected}; use crate::imap::command::{anonymous, authenticated, selected};
use crate::imap::flow; use crate::imap::flow;
@ -27,28 +27,40 @@ impl Instance {
pub async fn request(&mut self, req: Request) -> ResponseOrIdle { pub async fn request(&mut self, req: Request) -> ResponseOrIdle {
match req { match req {
Request::Idle => ResponseOrIdle::Response(self.idle().await), Request::Idle => self.idle().await,
Request::ImapCommand(cmd) => self.command(cmd).await, Request::ImapCommand(cmd) => self.command(cmd).await,
} }
} }
pub async fn idle(&mut self) -> Response<'static> { pub async fn idle(&mut self) -> ResponseOrIdle {
let (user, mbx, perm, stop) = match &mut self.state { match self.idle_happy().await {
flow::State::Idle(ref user, ref mut mailbox, ref perm, ref stop) => (user, mailbox, perm, stop), Ok(r) => r,
_ => unreachable!(), Err(e) => {
tracing::error!(err=?e, "something bad happened in idle");
ResponseOrIdle::Response(Response::bye().unwrap())
}
}
}
pub async fn idle_happy(&mut self) -> Result<ResponseOrIdle> {
let (mbx, tag, stop) = match &mut self.state {
flow::State::Idle(_, ref mut mbx, _, tag, stop) => (mbx, tag.clone(), stop.clone()),
_ => bail!("Invalid session state, can't idle"),
}; };
tokio::select! { tokio::select! {
_ = stop.notified() => { _ = stop.notified() => {
return Response::build() self.state.apply(flow::Transition::UnIdle)?;
.tag(imap_codec::imap_types::core::Tag::try_from("FIXME").unwrap()) return Ok(ResponseOrIdle::Response(Response::build()
.tag(tag.clone())
.message("IDLE completed") .message("IDLE completed")
.ok() .ok()?))
.unwrap() },
change = mbx.idle_sync() => {
tracing::debug!("idle event");
return Ok(ResponseOrIdle::IdleEvent(change?));
} }
} }
unimplemented!();
} }
@ -119,7 +131,7 @@ impl Instance {
} }
match &self.state { match &self.state {
flow::State::Idle(_, _, _, n) => ResponseOrIdle::StartIdle(n.clone()), flow::State::Idle(_, _, _, _, n) => ResponseOrIdle::StartIdle(n.clone()),
_ => ResponseOrIdle::Response(resp), _ => ResponseOrIdle::Response(resp),
} }
} }

View file

@ -68,8 +68,8 @@ impl Mailbox {
} }
/// Block until a sync has been done (due to changes in the event log) /// Block until a sync has been done (due to changes in the event log)
pub async fn idle_sync(&self) -> Result<()> { pub async fn notify(&self) -> std::sync::Weak<tokio::sync::Notify> {
self.mbox.write().await.idle_sync().await self.mbox.read().await.notifier()
} }
// ---- Functions for reading the mailbox ---- // ---- Functions for reading the mailbox ----
@ -204,9 +204,8 @@ impl MailboxInternal {
Ok(()) Ok(())
} }
async fn idle_sync(&mut self) -> Result<()> { fn notifier(&self) -> std::sync::Weak<tokio::sync::Notify> {
self.uid_index.idle_sync().await?; self.uid_index.notifier()
Ok(())
} }
// ---- Functions for reading the mailbox ---- // ---- Functions for reading the mailbox ----