diff --git a/src/app/materialize.rs b/src/app/materialize.rs index 944c834..33b7969 100644 --- a/src/app/materialize.rs +++ b/src/app/materialize.rs @@ -1,5 +1,5 @@ use std::{ - collections::{BTreeMap, BTreeSet}, + collections::{BTreeMap, BTreeSet, VecDeque}, fs, path::Path, str::FromStr, @@ -27,15 +27,13 @@ pub(super) async fn apply_remote_manifest( let mut synchronized_count = 0; let ignore_rules = IgnoreRules::load(&context.folder.path)?; let actions = crate::domain::reconcile_manifests(local, remote); - let (mut pending_actions, downloads) = + let (mut materialization_plan, downloads) = plan_remote_materialization(&ignore_rules, context.folder, actions, remote); - let mut available_blobs = BTreeSet::new(); materialize_ready_actions( context, &mut merged, &mut synchronized_count, - &mut pending_actions, - &available_blobs, + &mut materialization_plan, ) .await?; receive_and_materialize_files( @@ -43,8 +41,7 @@ pub(super) async fn apply_remote_manifest( &peer, &mut merged, &mut synchronized_count, - &mut pending_actions, - &mut available_blobs, + &mut materialization_plan, downloads, ) .await?; @@ -128,21 +125,6 @@ fn action_is_ignored( ) } -async fn materialize_conflict( - context: &SyncContext<'_>, - merged: &mut Manifest, - winner: &Entry, - loser: &Entry, - conflict_path: &str, -) -> anyhow::Result { - let mut writes = 0; - let mut conflict_entry = loser.clone(); - conflict_entry.path = conflict_path.to_owned(); - writes += materialize_entry(context, merged, conflict_entry).await?; - writes += materialize_entry(context, merged, winner.clone()).await?; - Ok(writes) -} - pub(super) async fn recover_pending_materialization( context: &SyncContext<'_>, ) -> anyhow::Result<()> { @@ -155,8 +137,13 @@ pub(super) async fn recover_pending_materialization( if pending.resulting_manifest.folder_id != context.folder.id { anyhow::bail!("pending materialization belongs to another folder"); } - tracing::warn!(path = %pending.entry.path, "Recovering interrupted filesystem update"); - apply_entry(context, &pending.entry).await?; + tracing::warn!( + entry_count = pending.entries.len(), + "Recovering interrupted filesystem updates" + ); + for entry in &pending.entries { + apply_entry(context, entry).await?; + } context .state_store .save_manifest(&pending.resulting_manifest)?; @@ -170,17 +157,30 @@ async fn materialize_entry( context: &SyncContext<'_>, manifest: &mut Manifest, entry: Entry, +) -> anyhow::Result { + materialize_entries(context, manifest, vec![entry]).await +} + +async fn materialize_entries( + context: &SyncContext<'_>, + manifest: &mut Manifest, + entries: Vec, ) -> anyhow::Result { let mut resulting_manifest = manifest.clone(); - resulting_manifest - .entries - .insert(entry.path.clone(), entry.clone()); + for entry in &entries { + resulting_manifest + .entries + .insert(entry.path.clone(), entry.clone()); + } context.state_store.save_pending_materialization( context.folder.id, - &entry, + &entries, &resulting_manifest, )?; - let writes = apply_entry(context, &entry).await?; + let mut writes = 0; + for entry in &entries { + writes += apply_entry(context, entry).await?; + } context.state_store.save_manifest(&resulting_manifest)?; context .state_store @@ -244,7 +244,13 @@ struct RemoteFile { struct PendingAction { action: ReconciliationAction, - required_blobs: BTreeSet, + missing_blob_count: usize, +} + +struct MaterializationPlan { + pending_actions: Vec>, + actions_waiting_by_blob: BTreeMap>, + ready_actions: VecDeque, } fn plan_remote_materialization( @@ -252,17 +258,40 @@ fn plan_remote_materialization( folder: &FolderConfig, actions: Vec, remote: &Manifest, -) -> (Vec, Vec) { +) -> (MaterializationPlan, Vec) { let mut files = BTreeMap::new(); - let pending_actions = actions + let mut pending_actions = Vec::new(); + let mut actions_waiting_by_blob = BTreeMap::new(); + let mut ready_actions = VecDeque::new(); + for action in actions .into_iter() .filter(|action| !action_is_ignored(folder, ignore_rules, action)) - .map(|action| PendingAction { - required_blobs: collect_required_blobs(&mut files, remote, &action), + { + let required_blobs = collect_required_blobs(&mut files, remote, &action); + let action_id = pending_actions.len(); + if required_blobs.is_empty() { + ready_actions.push_back(action_id); + } else { + for blob_hash in &required_blobs { + actions_waiting_by_blob + .entry(blob_hash.clone()) + .or_insert_with(Vec::new) + .push(action_id); + } + } + pending_actions.push(Some(PendingAction { action, - }) - .collect(); - (pending_actions, files.into_values().collect()) + missing_blob_count: required_blobs.len(), + })); + } + ( + MaterializationPlan { + pending_actions, + actions_waiting_by_blob, + ready_actions, + }, + files.into_values().collect(), + ) } fn collect_required_blobs( @@ -289,34 +318,74 @@ async fn materialize_ready_actions( context: &SyncContext<'_>, merged: &mut Manifest, synchronized_count: &mut usize, - pending_actions: &mut Vec, - available_blobs: &BTreeSet, + plan: &mut MaterializationPlan, ) -> anyhow::Result<()> { - while let Some(position) = pending_actions - .iter() - .position(|pending| pending.required_blobs.is_subset(available_blobs)) - { - let pending = pending_actions.remove(position); - *synchronized_count += materialize_action(context, merged, pending.action).await?; + while let Some(actions) = plan.take_ready_actions() { + *synchronized_count += materialize_action_batch(context, merged, actions).await?; } Ok(()) } -async fn materialize_action( +impl MaterializationPlan { + fn note_completed_blob(&mut self, blob_hash: String) { + let Some(waiting_actions) = self.actions_waiting_by_blob.remove(&blob_hash) else { + return; + }; + for action_id in waiting_actions { + let Some(pending) = self.pending_actions[action_id].as_mut() else { + continue; + }; + pending.missing_blob_count -= 1; + if pending.missing_blob_count == 0 { + self.ready_actions.push_back(action_id); + } + } + } + + fn take_ready_actions(&mut self) -> Option> { + const MATERIALIZATION_BATCH_SIZE: usize = 64; + let mut actions = Vec::new(); + while actions.len() < MATERIALIZATION_BATCH_SIZE { + let Some(action_id) = self.ready_actions.pop_front() else { + break; + }; + let Some(pending) = self.pending_actions[action_id].take() else { + continue; + }; + actions.push(pending.action); + } + (!actions.is_empty()).then_some(actions) + } +} + +async fn materialize_action_batch( context: &SyncContext<'_>, merged: &mut Manifest, - action: ReconciliationAction, + actions: Vec, ) -> anyhow::Result { + let entries = actions + .into_iter() + .flat_map(entries_for_action) + .collect::>(); + if entries.is_empty() { + return Ok(0); + } + materialize_entries(context, merged, entries).await +} + +fn entries_for_action(action: ReconciliationAction) -> Vec { match action { - ReconciliationAction::ApplyRemote { entry } => { - materialize_entry(context, merged, entry).await - } - ReconciliationAction::KeepLocal { .. } => Ok(0), + ReconciliationAction::ApplyRemote { entry } => vec![entry], + ReconciliationAction::KeepLocal { .. } => Vec::new(), ReconciliationAction::ResolveConflict { winner, loser, conflict_path, - } => materialize_conflict(context, merged, &winner, &loser, &conflict_path).await, + } => { + let mut conflict_entry = loser; + conflict_entry.path = conflict_path; + vec![conflict_entry, winner] + } } } @@ -325,8 +394,7 @@ async fn receive_and_materialize_files( peer: &iroh::EndpointAddr, merged: &mut Manifest, synchronized_count: &mut usize, - pending_actions: &mut Vec, - available_blobs: &mut BTreeSet, + plan: &mut MaterializationPlan, downloads: Vec, ) -> anyhow::Result<()> { let mut downloads = stream::iter(downloads) @@ -344,15 +412,8 @@ async fn receive_and_materialize_files( while let Some(result) = downloads.next().await { match result { Ok(blob_hash) => { - available_blobs.insert(blob_hash); - materialize_ready_actions( - context, - merged, - synchronized_count, - pending_actions, - available_blobs, - ) - .await?; + plan.note_completed_blob(blob_hash); + materialize_ready_actions(context, merged, synchronized_count, plan).await?; } Err(error) => { tracing::warn!(error = ?error, "Could not receive file; deferring dependent changes") @@ -466,21 +527,21 @@ mod tests { mode: FolderMode::SendReceive, }; - let (pending_actions, downloads) = plan_remote_materialization( + let (materialization_plan, downloads) = plan_remote_materialization( &IgnoreRules::load(directory.path())?, &folder, actions, &remote, ); - assert_eq!(pending_actions.len(), 2); + assert_eq!(materialization_plan.pending_actions.len(), 2); assert_eq!( - pending_actions[0].required_blobs, - ["first-hash".to_owned()].into() + materialization_plan.actions_waiting_by_blob["first-hash"], + vec![0] ); assert_eq!( - pending_actions[1].required_blobs, - ["second-hash".to_owned()].into() + materialization_plan.actions_waiting_by_blob["second-hash"], + vec![1] ); assert_eq!(downloads.len(), 2); Ok(()) diff --git a/src/storage.rs b/src/storage.rs index 58f2960..ac0da30 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -60,7 +60,7 @@ pub struct FolderConfig { #[derive(Clone, Debug)] pub struct PendingMaterialization { - pub entry: Entry, + pub entries: Vec, pub resulting_manifest: Manifest, } diff --git a/src/storage/materializations.rs b/src/storage/materializations.rs index eff5ec4..daec64a 100644 --- a/src/storage/materializations.rs +++ b/src/storage/materializations.rs @@ -9,12 +9,12 @@ impl StateStore { pub fn save_pending_materialization( &self, folder_id: FolderId, - entry: &Entry, + entries: &[Entry], resulting_manifest: &Manifest, ) -> anyhow::Result<()> { self.connection.execute( - "INSERT INTO pending_materializations (folder_id, entry, resulting_manifest) VALUES (?1, ?2, ?3) ON CONFLICT(folder_id) DO UPDATE SET entry = excluded.entry, resulting_manifest = excluded.resulting_manifest", - params![folder_id.to_string(), serde_json::to_string(entry)?, serde_json::to_string(resulting_manifest)?], + "INSERT INTO pending_materializations (folder_id, entries, resulting_manifest) VALUES (?1, ?2, ?3) ON CONFLICT(folder_id) DO UPDATE SET entries = excluded.entries, resulting_manifest = excluded.resulting_manifest", + params![folder_id.to_string(), serde_json::to_string(entries)?, serde_json::to_string(resulting_manifest)?], )?; Ok(()) } @@ -24,11 +24,11 @@ impl StateStore { folder_id: FolderId, ) -> anyhow::Result> { self.connection.query_row( - "SELECT entry, resulting_manifest FROM pending_materializations WHERE folder_id = ?1", + "SELECT entries, resulting_manifest FROM pending_materializations WHERE folder_id = ?1", params![folder_id.to_string()], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)), - ).optional()?.map(|(entry, manifest)| Ok(PendingMaterialization { - entry: serde_json::from_str(&entry)?, + ).optional()?.map(|(entries, manifest)| Ok(PendingMaterialization { + entries: serde_json::from_str(&entries)?, resulting_manifest: serde_json::from_str(&manifest)?, })).transpose() } diff --git a/src/storage/schema.rs b/src/storage/schema.rs index 60dd4dd..6fda89b 100644 --- a/src/storage/schema.rs +++ b/src/storage/schema.rs @@ -57,7 +57,7 @@ const TABLE_STATEMENTS: &[&str] = &[ );", "CREATE TABLE IF NOT EXISTS pending_materializations ( folder_id TEXT PRIMARY KEY NOT NULL, - entry TEXT NOT NULL, + entries TEXT NOT NULL, resulting_manifest TEXT NOT NULL );", ]; diff --git a/src/storage/tests.rs b/src/storage/tests.rs index f6cb4a5..01996d0 100644 --- a/src/storage/tests.rs +++ b/src/storage/tests.rs @@ -198,16 +198,19 @@ fn saves_and_clears_pending_materialization() -> anyhow::Result<()> { let store = StateStore::open(&paths)?; let folder = store.register_folder(&folder_path)?; let mut resulting_manifest = store.load_manifest(folder.id)?; - let entry = test_file_entry("note.txt"); - resulting_manifest - .entries - .insert(entry.path.clone(), entry.clone()); + let entries = vec![test_file_entry("note.txt"), test_file_entry("todo.txt")]; + resulting_manifest.entries.extend( + entries + .iter() + .cloned() + .map(|entry| (entry.path.clone(), entry)), + ); - store.save_pending_materialization(folder.id, &entry, &resulting_manifest)?; + store.save_pending_materialization(folder.id, &entries, &resulting_manifest)?; let pending = store .pending_materialization(folder.id)? .expect("journal exists"); - assert_eq!(pending.entry, entry); + assert_eq!(pending.entries, entries); assert_eq!(pending.resulting_manifest, resulting_manifest); store.clear_pending_materialization(folder.id)?;