You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by ps...@apache.org on 2026/08/13 07:29:43 UTC
(opendal-ofs) branch managed-sync-eval updated: perf: stream incremental managed sync
This is an automated email from the ASF dual-hosted git repository.
PsiACE pushed a commit to branch managed-sync-eval
in repository https://gitbox.apache.org/repos/asf/opendal-ofs.git
The following commit(s) were added to refs/heads/managed-sync-eval by this push:
new 6eaac80 perf: stream incremental managed sync
6eaac80 is described below
commit 6eaac8059a86329e37d1c29cb98af7d850a48f93
Author: PsiACE <ps...@apache.org>
AuthorDate: Thu Aug 13 15:29:26 2026 +0800
perf: stream incremental managed sync
---
crates/core/src/managed/file.rs | 167 ++++++++++++++++++++++++++-
crates/core/src/managed/head.rs | 57 ++++++++--
crates/core/src/managed/mod.rs | 1 +
crates/core/src/managed/namespace.rs | 213 +++++++++++++++++++++++------------
crates/core/src/sync/engine.rs | 34 +++---
crates/core/src/sync/install.rs | 38 ++++++-
crates/core/src/sync/transfer.rs | 94 +++++++++++++++-
7 files changed, 494 insertions(+), 110 deletions(-)
diff --git a/crates/core/src/managed/file.rs b/crates/core/src/managed/file.rs
index 2dc262f..cf33908 100644
--- a/crates/core/src/managed/file.rs
+++ b/crates/core/src/managed/file.rs
@@ -20,7 +20,7 @@
use std::ops::{Bound, Range, RangeBounds};
use opendal::Operator;
-use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncSeek, AsyncSeekExt as _, AsyncWrite};
use crate::Error;
use crate::filesystem::ContentRef;
@@ -40,6 +40,15 @@ pub struct FileExtentReader {
tail: Option<RecordStreamReader<ExtentRef>>,
}
+pub(crate) struct ReusableFile<'a> {
+ pub(crate) reference: FileDataRef,
+ pub(crate) source: &'a mut (dyn ReusableFileSource + 'a),
+}
+
+pub(crate) trait ReusableFileSource: AsyncRead + AsyncSeek + Send + Unpin {}
+
+impl<T> ReusableFileSource for T where T: AsyncRead + AsyncSeek + Send + Unpin {}
+
impl FileDataRef {
/// Open the ordered logical extent stream.
pub async fn extents(&self, operator: &Operator) -> Result<FileExtentReader, Error> {
@@ -166,14 +175,19 @@ impl ManagedVolume {
}
/// Read one logical byte range through its ordered extents.
- pub(crate) async fn read_data(
- &self,
+ pub(crate) async fn read_data<'a>(
+ &'a self,
content: (ContentRef, FileDataRef),
range: impl RangeBounds<u64>,
- destination: &mut (impl AsyncWrite + Send + Unpin),
+ reusable: Option<ReusableFile<'a>>,
+ destination: &'a mut (impl AsyncWrite + Send + Unpin),
) -> Result<(), Error> {
let (content, reference) = content;
let range = logical_range(content.length(), range)?;
+ if let Some(reusable) = reusable {
+ return read_reusing_extents(self, reference, content, range, reusable, destination)
+ .await;
+ }
if self.format.file_storage().access.is_some() {
return self
.file_access()?
@@ -184,6 +198,151 @@ impl ManagedVolume {
}
}
+async fn read_reusing_extents(
+ volume: &ManagedVolume,
+ target: FileDataRef,
+ content: ContentRef,
+ range: Range<u64>,
+ reusable: ReusableFile<'_>,
+ destination: &mut (impl AsyncWrite + Send + Unpin),
+) -> Result<(), Error> {
+ let mut target = target.extents(volume.operator()).await?;
+ let mut existing = reusable.reference.extents(volume.operator()).await?;
+ let mut existing_extent = None::<(ExtentRef, u64, u64)>;
+ let mut existing_offset = 0_u64;
+ let mut target_offset = 0_u64;
+ let mut remote = Vec::new();
+ while let Some(extent) = target.next().await? {
+ validate_extent(volume, &extent)?;
+ let extent_end = logical_extent_end(target_offset, &extent)?;
+ while existing_extent
+ .as_ref()
+ .is_none_or(|(_, _, end)| *end <= target_offset)
+ {
+ let Some(extent) = existing.next().await? else {
+ existing_extent = None;
+ break;
+ };
+ validate_extent(volume, &extent)?;
+ let end = logical_extent_end(existing_offset, &extent)?;
+ existing_extent = Some((extent, existing_offset, end));
+ existing_offset = end;
+ }
+ if extent_end <= range.start || target_offset >= range.end {
+ target_offset = extent_end;
+ continue;
+ }
+ let reusable_offset = existing_extent.as_ref().and_then(|(existing, offset, _)| {
+ (existing.content() == extent.content()).then_some(*offset)
+ });
+ if let Some(reusable_offset) = reusable_offset {
+ flush_remote(volume, &mut remote, range.clone(), destination).await?;
+ let selected = range.start.saturating_sub(target_offset)
+ ..range.end.min(extent_end) - target_offset;
+ copy_reusable(
+ reusable.source,
+ reusable_offset + selected.start,
+ selected.end - selected.start,
+ destination,
+ )
+ .await?;
+ } else {
+ if remote.last().is_some_and(|(previous, _, _)| {
+ previous.range.segment != extent.range.segment
+ || stored_extent_end(previous).ok() != Some(extent.range.offset)
+ }) {
+ flush_remote(volume, &mut remote, range.clone(), destination).await?;
+ }
+ remote.push((extent, target_offset, extent_end));
+ }
+ target_offset = extent_end;
+ }
+ flush_remote(volume, &mut remote, range, destination).await?;
+ if target_offset != content.length() {
+ return Err(Error::corrupt(
+ "read Managed file",
+ "extents do not cover the logical file",
+ ));
+ }
+ Ok(())
+}
+
+async fn copy_reusable(
+ source: &mut (dyn ReusableFileSource + '_),
+ offset: u64,
+ length: u64,
+ destination: &mut (impl AsyncWrite + Send + Unpin),
+) -> Result<(), Error> {
+ source
+ .seek(std::io::SeekFrom::Start(offset))
+ .await
+ .map_err(|error| Error::io("seek reusable Managed file", error))?;
+ let copied = tokio::io::copy(&mut source.take(length), destination)
+ .await
+ .map_err(|error| Error::io("copy reusable Managed file", error))?;
+ if copied != length {
+ return Err(Error::conflict(
+ "copy reusable Managed file",
+ "verified local file changed during installation",
+ ));
+ }
+ Ok(())
+}
+
+async fn flush_remote(
+ volume: &ManagedVolume,
+ run: &mut Vec<(ExtentRef, u64, u64)>,
+ range: Range<u64>,
+ destination: &mut (impl AsyncWrite + Send + Unpin),
+) -> Result<(), Error> {
+ let Some((first, _, _)) = run.first() else {
+ return Ok(());
+ };
+ let segment = first.range.segment;
+ let start = first.range.offset;
+ let end = stored_extent_end(&run.last().expect("remote extent run is not empty").0)?;
+ let mut reader = SegmentRangeReader::open(volume.operator(), segment, start..end).await?;
+ for (extent, offset, extent_end) in run.drain(..) {
+ volume
+ .read_extent_from(
+ &mut reader,
+ extent,
+ range.start.saturating_sub(offset)..range.end.min(extent_end) - offset,
+ destination,
+ )
+ .await?;
+ }
+ Ok(())
+}
+
+fn validate_extent(volume: &ManagedVolume, extent: &ExtentRef) -> Result<(), Error> {
+ if extent.range.segment.class != ObjectClass::DataSegment
+ || extent.range.stored.length() == 0
+ || extent.content().length() == 0
+ || extent.decoded.len() != volume.file_decoding_count()
+ {
+ return Err(Error::corrupt(
+ "read Managed file",
+ "file extent is invalid",
+ ));
+ }
+ stored_extent_end(extent).map(drop)
+}
+
+fn logical_extent_end(offset: u64, extent: &ExtentRef) -> Result<u64, Error> {
+ offset
+ .checked_add(extent.content().length())
+ .ok_or_else(|| Error::corrupt("read Managed file", "logical extent range overflows"))
+}
+
+fn stored_extent_end(extent: &ExtentRef) -> Result<u64, Error> {
+ extent
+ .range
+ .offset
+ .checked_add(extent.range.stored.length())
+ .ok_or_else(|| Error::corrupt("read Managed file", "stored extent range overflows"))
+}
+
async fn verify_empty(source: &mut (impl AsyncRead + Send + Unpin)) -> Result<(), Error> {
let mut byte = [0_u8; 1];
if source
diff --git a/crates/core/src/managed/head.rs b/crates/core/src/managed/head.rs
index 93f3d2c..f3f8d63 100644
--- a/crates/core/src/managed/head.rs
+++ b/crates/core/src/managed/head.rs
@@ -204,18 +204,57 @@ impl ManagedVolume {
}
pub(crate) async fn observe(&self) -> Result<ManagedObservation, Error> {
+ self.observe_with_base(None)
+ .await
+ .map(|(observation, _)| observation)
+ }
+
+ pub(crate) async fn observe_with_base(
+ &self,
+ base: Option<NamespaceRevision>,
+ ) -> Result<(ManagedObservation, Option<Namespace<FileDataRef>>), Error> {
let authority = self.read_authority().await?;
let head = authority.head();
let commit = publication::read_commit(self, head.current_commit).await?;
- let namespace = namespace::read(self, &commit, commit.change_cursor).await?;
- Ok(ManagedObservation {
- namespace,
- authority,
- namespace_revision: head.current_commit,
- reclamation_watermark: head.minimum_retained_cursor,
- gc_epoch: head.gc_epoch,
- commit,
- })
+ let readable_base = base.filter(|base| {
+ let sequence = base.change_cursor.sequence();
+ sequence >= head.minimum_retained_cursor.sequence()
+ && sequence < head.current_commit.change_cursor.sequence()
+ });
+ let (namespace, base_namespace) = match readable_base {
+ Some(base) => {
+ let reference = if base.change_cursor == head.minimum_retained_cursor
+ && base.object.locator.gc_epoch < head.current_commit.object.locator.gc_epoch
+ {
+ head.current_commit
+ } else {
+ base
+ };
+ let base_commit = publication::read_commit(self, reference).await?;
+ let (base_namespace, namespace) = namespace::read_pair(
+ self,
+ (&base_commit, base.change_cursor),
+ (&commit, commit.change_cursor),
+ )
+ .await?;
+ (namespace, Some(base_namespace))
+ }
+ None => (
+ namespace::read(self, &commit, commit.change_cursor).await?,
+ None,
+ ),
+ };
+ Ok((
+ ManagedObservation {
+ namespace,
+ authority,
+ namespace_revision: head.current_commit,
+ reclamation_watermark: head.minimum_retained_cursor,
+ gc_epoch: head.gc_epoch,
+ commit,
+ },
+ base_namespace,
+ ))
}
pub(super) async fn read_authority(&self) -> Result<AuthorityObservation, Error> {
diff --git a/crates/core/src/managed/mod.rs b/crates/core/src/managed/mod.rs
index c963332..1c9fefd 100644
--- a/crates/core/src/managed/mod.rs
+++ b/crates/core/src/managed/mod.rs
@@ -41,6 +41,7 @@ pub use authority::{
};
pub use data::FileDataRef;
pub use file::FileExtentReader;
+pub(crate) use file::ReusableFile;
pub(crate) use format::ManagedFormat;
pub use gc::GcOutcome;
pub(crate) use head::ManagedObservation;
diff --git a/crates/core/src/managed/namespace.rs b/crates/core/src/managed/namespace.rs
index fce7ab9..6d02451 100644
--- a/crates/core/src/managed/namespace.rs
+++ b/crates/core/src/managed/namespace.rs
@@ -32,10 +32,9 @@ use crate::namespace::Namespace;
use crate::workset::{MergeRuns, Spool, Workspace};
use super::data::FileDataRef;
-use super::head::{ManagedVolume, NamespaceRevision};
+use super::head::ManagedVolume;
use super::layout::{NamespaceChangeSegment, NamespaceCommit};
use super::object::{GcEpoch, ObjectClass};
-use super::publication;
use super::stream::{self, RecordStreamReader, RecordStreamWriter, StreamKind, StreamRef};
pub(super) async fn write_genesis(
@@ -62,33 +61,6 @@ pub(super) async fn write_genesis(
.await
}
-impl ManagedVolume {
- pub(crate) async fn namespace(
- &self,
- revision: NamespaceRevision,
- ) -> Result<Namespace<FileDataRef>, Error> {
- let authority = self.read_authority().await?;
- let head = authority.head();
- if revision.change_cursor.sequence() < head.minimum_retained_cursor.sequence()
- || revision.change_cursor.sequence() > head.current_commit.change_cursor.sequence()
- {
- return Err(Error::invalid(
- "read Managed namespace",
- "requested change cursor is outside the retained interval",
- ));
- }
- let reference = if revision.change_cursor == head.minimum_retained_cursor
- && revision.object.locator.gc_epoch < head.current_commit.object.locator.gc_epoch
- {
- head.current_commit
- } else {
- revision
- };
- let commit = publication::read_commit(self, reference).await?;
- read(self, &commit, revision.change_cursor).await
- }
-}
-
pub(super) async fn write_snapshot(
volume: &ManagedVolume,
namespace: &Namespace<FileDataRef>,
@@ -242,9 +214,88 @@ pub(super) async fn read(
volume: &ManagedVolume,
commit: &NamespaceCommit,
view_cursor: ChangeCursor,
+) -> Result<Namespace<FileDataRef>, Error> {
+ let workspace = Workspace::create(volume.worksets)?;
+ let snapshot = download_snapshot(
+ volume,
+ &workspace,
+ commit.namespace_snapshot.stream,
+ commit.namespace_snapshot.change_cursor,
+ )
+ .await?;
+ read_from_snapshot(volume, commit, view_cursor, &workspace, snapshot).await
+}
+
+pub(super) async fn read_pair(
+ volume: &ManagedVolume,
+ older: (&NamespaceCommit, ChangeCursor),
+ newer: (&NamespaceCommit, ChangeCursor),
+) -> Result<(Namespace<FileDataRef>, Namespace<FileDataRef>), Error> {
+ if older.0.namespace_snapshot.change_cursor != newer.0.namespace_snapshot.change_cursor
+ || older.0.namespace_snapshot.stream != newer.0.namespace_snapshot.stream
+ {
+ return tokio::try_join!(
+ read(volume, older.0, older.1),
+ read(volume, newer.0, newer.1)
+ );
+ }
+ let workspace = Workspace::create(volume.worksets)?;
+ let snapshot = download_snapshot(
+ volume,
+ &workspace,
+ newer.0.namespace_snapshot.stream,
+ newer.0.namespace_snapshot.change_cursor,
+ )
+ .await?;
+ let mut segments = Vec::new();
+ for (commit, _) in [older, newer] {
+ for segment in commit.namespace_changes.iter().copied() {
+ validate_change_segment(segment, commit.change_cursor)?;
+ if !segments.iter().any(|current: &NamespaceChangeSegment| {
+ current.end_cursor == segment.end_cursor && current.stream == segment.stream
+ }) {
+ segments.push(segment);
+ }
+ }
+ }
+ let workspace_ref = &workspace;
+ let commit_cursor = newer.0.change_cursor;
+ let downloads = futures::stream::iter(segments)
+ .map(|segment| async move {
+ Ok::<_, Error>(DownloadedChange {
+ end_cursor: segment.end_cursor,
+ stream: segment.stream,
+ records: download_changes(volume, workspace_ref, segment, commit_cursor).await?,
+ })
+ })
+ .buffer_unordered(volume.stream_concurrency);
+ futures::pin_mut!(downloads);
+ let mut changes = Vec::new();
+ while let Some(download) = downloads.next().await {
+ changes.push(download?);
+ }
+ Ok((
+ read_from_downloads(
+ volume,
+ older.0,
+ older.1,
+ &workspace,
+ snapshot.clone(),
+ &changes,
+ )?,
+ read_from_downloads(volume, newer.0, newer.1, &workspace, snapshot, &changes)?,
+ ))
+}
+
+async fn read_from_snapshot(
+ volume: &ManagedVolume,
+ commit: &NamespaceCommit,
+ view_cursor: ChangeCursor,
+ workspace: &Workspace,
+ snapshot: Spool<VersionedRecord>,
) -> Result<Namespace<FileDataRef>, Error> {
if view_cursor == commit.namespace_snapshot.change_cursor {
- return read_snapshot(volume, commit, view_cursor).await;
+ return finish_view(volume, commit, view_cursor, workspace, snapshot);
}
if view_cursor != commit.change_cursor {
return Err(Error::corrupt(
@@ -264,53 +315,57 @@ pub(super) async fn read(
"snapshot is newer than the requested view",
));
}
- let workspace = Workspace::create(volume.worksets)?;
- let mut streams = MergeRuns::new(workspace.merge_width());
- streams.push(
- download_snapshot(
- volume,
- &workspace,
- commit.namespace_snapshot.stream,
- commit.namespace_snapshot.change_cursor,
- )
- .await?,
- |group| merge_group(&workspace, group, view_cursor),
- )?;
+ let commit_cursor = commit.change_cursor;
let downloads = futures::stream::iter(commit.namespace_changes.iter().copied())
- .map(|segment| download_changes(volume, &workspace, segment, commit.change_cursor))
+ .map(|segment| async move {
+ Ok::<_, Error>(DownloadedChange {
+ end_cursor: segment.end_cursor,
+ stream: segment.stream,
+ records: download_changes(volume, workspace, segment, commit_cursor).await?,
+ })
+ })
.buffer_unordered(volume.stream_concurrency);
futures::pin_mut!(downloads);
- while let Some(stream) = downloads.next().await {
- streams.push(stream?, |group| merge_group(&workspace, group, view_cursor))?;
+ let mut changes = Vec::new();
+ while let Some(download) = downloads.next().await {
+ changes.push(download?);
}
- let merged = streams
- .finish(|group| merge_group(&workspace, group, view_cursor))?
- .ok_or_else(|| Error::corrupt("read Managed namespace", "namespace has no streams"))?;
- finish_view(volume, commit, view_cursor, &workspace, merged)
+ read_from_downloads(volume, commit, view_cursor, workspace, snapshot, &changes)
}
-async fn read_snapshot(
+fn read_from_downloads(
volume: &ManagedVolume,
commit: &NamespaceCommit,
view_cursor: ChangeCursor,
+ workspace: &Workspace,
+ snapshot: Spool<VersionedRecord>,
+ changes: &[DownloadedChange],
) -> Result<Namespace<FileDataRef>, Error> {
- let reference = commit.namespace_snapshot.stream;
- reference.require(
- StreamKind::NAMESPACE_SNAPSHOT,
- ObjectClass::NamespaceSegment,
- )?;
- let workspace = Workspace::create(volume.worksets)?;
- let mut output = workspace.writer("namespace")?;
- let mut remote =
- RecordStreamReader::<NamespaceRecord<FileDataRef>>::open(&volume.operator, reference)
- .await?;
- let mut previous = None::<String>;
- let mut root_seen = false;
- while let Some(record) = remote.next().await? {
- validate_record(volume, &record, &mut previous, &mut root_seen)?;
- output.write(&record)?;
+ if view_cursor == commit.namespace_snapshot.change_cursor {
+ return finish_view(volume, commit, view_cursor, workspace, snapshot);
}
- finish_namespace(volume, commit, view_cursor, output.finish()?, root_seen)
+ let mut streams = MergeRuns::new(workspace.merge_width());
+ streams.push(snapshot, |group| merge_group(workspace, group, view_cursor))?;
+ for segment in &commit.namespace_changes {
+ let stream = changes
+ .iter()
+ .find(|download| {
+ download.end_cursor == segment.end_cursor && download.stream == segment.stream
+ })
+ .ok_or_else(|| {
+ Error::corrupt(
+ "read Managed namespace",
+ "namespace change stream was not downloaded",
+ )
+ })?;
+ streams.push(stream.records.clone(), |group| {
+ merge_group(workspace, group, view_cursor)
+ })?;
+ }
+ let merged = streams
+ .finish(|group| merge_group(workspace, group, view_cursor))?
+ .ok_or_else(|| Error::corrupt("read Managed namespace", "namespace has no streams"))?;
+ finish_view(volume, commit, view_cursor, workspace, merged)
}
fn finish_view(
@@ -394,12 +449,7 @@ async fn download_changes(
) -> Result<Spool<VersionedRecord>, Error> {
let reference = segment.stream;
reference.require(StreamKind::NAMESPACE_CHANGES, ObjectClass::NamespaceSegment)?;
- if segment.end_cursor > commit_cursor || segment.source_bytes == 0 {
- return Err(Error::corrupt(
- "read Managed namespace",
- "namespace change descriptor is invalid",
- ));
- }
+ validate_change_segment(segment, commit_cursor)?;
let mut remote =
RecordStreamReader::<NamespaceRecord<FileDataRef>>::open(&volume.operator, reference)
.await?;
@@ -417,6 +467,25 @@ async fn download_changes(
local.finish()
}
+fn validate_change_segment(
+ segment: NamespaceChangeSegment,
+ commit_cursor: ChangeCursor,
+) -> Result<(), Error> {
+ if segment.end_cursor > commit_cursor || segment.source_bytes == 0 {
+ return Err(Error::corrupt(
+ "read Managed namespace",
+ "namespace change descriptor is invalid",
+ ));
+ }
+ Ok(())
+}
+
+struct DownloadedChange {
+ end_cursor: ChangeCursor,
+ stream: StreamRef,
+ records: Spool<VersionedRecord>,
+}
+
fn merge_group(
workspace: &Workspace,
streams: &[Spool<VersionedRecord>],
diff --git a/crates/core/src/sync/engine.rs b/crates/core/src/sync/engine.rs
index fd706a5..52ae0cc 100644
--- a/crates/core/src/sync/engine.rs
+++ b/crates/core/src/sync/engine.rs
@@ -19,15 +19,13 @@ use std::collections::BTreeSet;
use std::num::NonZeroUsize;
use std::path::Path;
-use serde::de::DeserializeOwned;
-
use crate::Error;
use crate::filesystem::ChangeCursor;
use crate::managed::{FileDataRef, ManagedVolume, NamespaceRevision};
use crate::namespace::Namespace;
use super::ReplicaState;
-use super::install::install;
+use super::install::{InstalledData, install};
use super::reconcile::{ReconcilePlan, reconcile};
use super::scan::{ScannedTree, scan};
use super::state_file;
@@ -74,7 +72,6 @@ impl SyncEngine {
));
}
- let observed = self.volume.observe().await?;
let stored = state_file::load(state_path)?;
if let Some(state) = &stored {
if state.root() != root {
@@ -89,14 +86,20 @@ impl SyncEngine {
"replica state belongs to a different volume",
));
}
- if state.authority_id() != observed.authority_id()
- || state.authority_name() != self.volume.authority_name()
- {
- return Err(Error::invalid(
- "synchronize replica",
- "replica state belongs to a different namespace authority",
- ));
- }
+ }
+ let base_revision = stored
+ .as_ref()
+ .filter(|state| state.installation().is_none() && state.pending_publication().is_none())
+ .map(ReplicaState::common_revision);
+ let (observed, loaded_base) = self.volume.observe_with_base(base_revision).await?;
+ if let Some(state) = &stored
+ && (state.authority_id() != observed.authority_id()
+ || state.authority_name() != self.volume.authority_name())
+ {
+ return Err(Error::invalid(
+ "synchronize replica",
+ "replica state belongs to a different namespace authority",
+ ));
}
if let Some((target, published)) = stored.as_ref().and_then(ReplicaState::installation) {
return self
@@ -185,11 +188,6 @@ impl SyncEngine {
}
let common_revision = state.common_revision();
- let loaded_base = if observed.revision() == common_revision {
- None
- } else {
- Some(self.volume.namespace(common_revision).await?)
- };
let base = loaded_base.as_ref().unwrap_or(&observed.namespace);
let local = scan(
&root,
@@ -289,7 +287,7 @@ impl SyncEngine {
}
}
- pub(super) async fn install_and_advance<C: DeserializeOwned>(
+ pub(super) async fn install_and_advance<C: InstalledData>(
&self,
root: &Path,
state_path: &Path,
diff --git a/crates/core/src/sync/install.rs b/crates/core/src/sync/install.rs
index e80af55..6de6a33 100644
--- a/crates/core/src/sync/install.rs
+++ b/crates/core/src/sync/install.rs
@@ -31,7 +31,7 @@ use super::local_fs::{self, DirectoryDurability, StoredPath};
use super::segment_install::{self, Installation as SegmentInstallation};
use super::transfer::materialize_file;
-pub(crate) async fn install<C: DeserializeOwned>(
+pub(crate) async fn install<C: InstalledData>(
root: &Path,
current: Option<&Namespace<C>>,
target: &Namespace<FileDataRef>,
@@ -51,7 +51,7 @@ pub(crate) async fn repair(
apply::<FileDataRef>(root, None, target, volume, transfer_concurrency, true).await
}
-async fn apply<C: DeserializeOwned>(
+async fn apply<C: InstalledData>(
root: &Path,
current: Option<&Namespace<C>>,
target: &Namespace<FileDataRef>,
@@ -139,6 +139,13 @@ async fn apply<C: DeserializeOwned>(
continue;
}
let executable = node.attributes.executable;
+ let reusable = matching_current.and_then(|current| {
+ current.value.as_ref().and_then(|current| {
+ current
+ .file()
+ .and_then(|(_, _, data)| data.reference().cloned())
+ })
+ });
if let Some(extent) = data
.tail()
.is_none()
@@ -164,6 +171,7 @@ async fn apply<C: DeserializeOwned>(
destination,
content,
data.clone(),
+ reusable,
executable,
authoritative,
));
@@ -200,6 +208,7 @@ async fn install_file(
destination: std::path::PathBuf,
fingerprint: crate::filesystem::ContentRef,
content: FileDataRef,
+ reusable: Option<FileDataRef>,
executable: bool,
authoritative: bool,
) -> Result<Option<std::path::PathBuf>, Error> {
@@ -210,7 +219,14 @@ async fn install_file(
local_fs::remove_replaced_directory(&destination)?;
crate::fault::check("during-install")?;
}
- materialize_file(volume, (fingerprint, content), &destination, executable).await?;
+ materialize_file(
+ volume,
+ (fingerprint, content),
+ reusable,
+ &destination,
+ executable,
+ )
+ .await?;
crate::fault::check("during-install")?;
Ok(Some(destination))
}
@@ -248,6 +264,22 @@ fn namespace_removals<C: DeserializeOwned>(
removed.finish()
}
+pub(crate) trait InstalledData: DeserializeOwned {
+ fn reference(&self) -> Option<&FileDataRef>;
+}
+
+impl InstalledData for FileDataRef {
+ fn reference(&self) -> Option<&FileDataRef> {
+ Some(self)
+ }
+}
+
+impl InstalledData for Option<FileDataRef> {
+ fn reference(&self) -> Option<&FileDataRef> {
+ self.as_ref()
+ }
+}
+
fn repair_removals(
root: &Path,
target: &Namespace<FileDataRef>,
diff --git a/crates/core/src/sync/transfer.rs b/crates/core/src/sync/transfer.rs
index de61e9d..6322931 100644
--- a/crates/core/src/sync/transfer.rs
+++ b/crates/core/src/sync/transfer.rs
@@ -16,14 +16,18 @@
// under the License.
use std::path::Path;
+use std::pin::Pin;
+use std::task::{Context, Poll};
use blake3::Hasher;
use tokio::fs::File;
-use tokio::io::AsyncReadExt as _;
+use tokio::io::{AsyncReadExt as _, AsyncWrite};
use crate::Error;
use crate::filesystem::{ContentRef, Digest, OperationId};
-use crate::managed::{DataSegmentPlacement, FileDataRef, GcEpoch, KnownContent, ManagedVolume};
+use crate::managed::{
+ DataSegmentPlacement, FileDataRef, GcEpoch, KnownContent, ManagedVolume, ReusableFile,
+};
const IO_BUFFER_BYTES: usize = 256 * 1024;
@@ -88,14 +92,32 @@ pub(super) async fn publish_file_into(
pub(super) async fn materialize_file(
volume: &ManagedVolume,
content: (ContentRef, FileDataRef),
+ reusable: Option<FileDataRef>,
destination: &Path,
executable: bool,
) -> Result<(), Error> {
let mut staging = StagedFile::create(destination).await?;
+ let mut source = match reusable {
+ Some(reference) => Some((
+ reference,
+ File::open(destination).await.map_err(|error| {
+ Error::from_io("open reusable replica file", Some(destination), error)
+ })?,
+ )),
+ None => None,
+ };
+ let reusable = source.as_mut().map(|(reference, source)| ReusableFile {
+ reference: reference.clone(),
+ source,
+ });
+ let expected = content.0;
+ let staging_path = staging.path().to_owned();
+ let mut writer = ContentWriter::new(staging.writer());
let result = volume
- .read_data(content, .., staging.writer())
+ .read_data(content, .., reusable, &mut writer)
.await
- .map_err(|error| error.with_context("path", staging.path().display()));
+ .and_then(|()| writer.finish(expected))
+ .map_err(|error| error.with_context("path", staging_path.display()));
if let Err(error) = result {
staging.abort().await;
return Err(error);
@@ -103,6 +125,70 @@ pub(super) async fn materialize_file(
staging.commit(executable).await
}
+struct ContentWriter<'a> {
+ inner: &'a mut File,
+ hasher: Hasher,
+ length: u64,
+}
+
+impl<'a> ContentWriter<'a> {
+ fn new(inner: &'a mut File) -> Self {
+ Self {
+ inner,
+ hasher: Hasher::new(),
+ length: 0,
+ }
+ }
+
+ fn finish(self, expected: ContentRef) -> Result<(), Error> {
+ let observed = ContentRef::new(
+ Digest::from_bytes(self.hasher.finalize().into()),
+ self.length,
+ );
+ if observed != expected {
+ return Err(Error::conflict(
+ "install Managed file",
+ "materialized bytes do not match the target content",
+ ));
+ }
+ Ok(())
+ }
+}
+
+impl AsyncWrite for ContentWriter<'_> {
+ fn poll_write(
+ mut self: Pin<&mut Self>,
+ context: &mut Context<'_>,
+ bytes: &[u8],
+ ) -> Poll<std::io::Result<usize>> {
+ match Pin::new(&mut *self.inner).poll_write(context, bytes) {
+ Poll::Ready(Ok(written)) => {
+ self.hasher.update(&bytes[..written]);
+ self.length = self
+ .length
+ .checked_add(written as u64)
+ .ok_or_else(|| std::io::Error::other("materialized file length overflows"))?;
+ Poll::Ready(Ok(written))
+ }
+ result => result,
+ }
+ }
+
+ fn poll_flush(
+ mut self: Pin<&mut Self>,
+ context: &mut Context<'_>,
+ ) -> Poll<std::io::Result<()>> {
+ Pin::new(&mut *self.inner).poll_flush(context)
+ }
+
+ fn poll_shutdown(
+ mut self: Pin<&mut Self>,
+ context: &mut Context<'_>,
+ ) -> Poll<std::io::Result<()>> {
+ Pin::new(&mut *self.inner).poll_shutdown(context)
+ }
+}
+
/// One atomic local-file installation shared by every data placement.
pub(super) struct StagedFile {
destination: std::path::PathBuf,