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/14 05:47:42 UTC
(opendal-ofs) 04/12: refactor: compose managed file access as one chain
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
commit 32a18ea42f136b00cab075d4be2d5dc6f32c15ed
Author: PsiACE <ps...@apache.org>
AuthorDate: Fri Aug 14 02:11:12 2026 +0800
refactor: compose managed file access as one chain
---
crates/core/src/managed/extension.rs | 57 +++++++-
crates/core/src/managed/file.rs | 267 +++++++++++++----------------------
crates/core/src/managed/head.rs | 18 +--
crates/core/src/managed/mod.rs | 17 ++-
crates/core/src/sync/publication.rs | 23 ++-
crates/core/src/sync/transfer.rs | 2 +-
crates/extensions/fastcdc/src/lib.rs | 49 +++++--
crates/extras/src/lib.rs | 74 +++++-----
8 files changed, 250 insertions(+), 257 deletions(-)
diff --git a/crates/core/src/managed/extension.rs b/crates/core/src/managed/extension.rs
index 6f2646d..b21b2a1 100644
--- a/crates/core/src/managed/extension.rs
+++ b/crates/core/src/managed/extension.rs
@@ -399,7 +399,18 @@ impl<T: ExtendedExtentAccess> ExtentAccess for T {
/// Logical file access. Layout extensions are built over an extent access.
pub trait FileAccess: Send + Sync + fmt::Debug + Unpin + 'static {
/// Describe the persisted layout and extent encoding.
- fn info(&self) -> FileAccessInfo;
+ fn info(&self) -> Option<FileAccessInfo>;
+
+ /// Describe the physical-to-logical extent decoding chain.
+ fn extent_decodings(&self) -> Vec<ExtensionFormat>;
+
+ /// Append one bounded extent through the composed encoding chain.
+ fn write_extent_into<'a>(
+ &'a self,
+ context: &'a AccessContext,
+ placement: &'a mut DataSegmentPlacement<'_>,
+ source: &'a mut (dyn AsyncRead + Send + Unpin),
+ ) -> impl Future<Output = Result<ExtentRef, Error>> + Send + 'a;
/// Publish one logical file into a shared segment placement session.
fn write_into<'a>(
@@ -432,7 +443,7 @@ pub trait FileAccess: Send + Sync + fmt::Debug + Unpin + 'static {
}
/// Typed file-layout extension over an extent access.
-pub trait FilePartitionExtension<A: ExtentAccess> {
+pub trait FilePartitionExtension<A: FileAccess> {
/// Resulting statically composed file access.
type ExtendedAccess: FileAccess;
@@ -461,10 +472,25 @@ pub trait ExtendedFileAccess: Send + Sync + fmt::Debug + Unpin + 'static {
fn inner(&self) -> &Self::Inner;
/// Forward the persisted access description by default.
- fn info(&self) -> FileAccessInfo {
+ fn info(&self) -> Option<FileAccessInfo> {
self.inner().info()
}
+ /// Forward the extent decoding description by default.
+ fn extent_decodings(&self) -> Vec<ExtensionFormat> {
+ self.inner().extent_decodings()
+ }
+
+ /// Forward encoded extent placement by default.
+ fn write_extent_into<'a>(
+ &'a self,
+ context: &'a AccessContext,
+ placement: &'a mut DataSegmentPlacement<'_>,
+ source: &'a mut (dyn AsyncRead + Send + Unpin),
+ ) -> impl Future<Output = Result<ExtentRef, Error>> + Send + 'a {
+ self.inner().write_extent_into(context, placement, source)
+ }
+
/// Forward publication into the caller's placement session.
fn write_into<'a>(
&'a self,
@@ -504,10 +530,23 @@ pub trait ExtendedFileAccess: Send + Sync + fmt::Debug + Unpin + 'static {
}
impl<T: ExtendedFileAccess> FileAccess for T {
- fn info(&self) -> FileAccessInfo {
+ fn info(&self) -> Option<FileAccessInfo> {
ExtendedFileAccess::info(self)
}
+ fn extent_decodings(&self) -> Vec<ExtensionFormat> {
+ ExtendedFileAccess::extent_decodings(self)
+ }
+
+ fn write_extent_into<'a>(
+ &'a self,
+ context: &'a AccessContext,
+ placement: &'a mut DataSegmentPlacement<'_>,
+ source: &'a mut (dyn AsyncRead + Send + Unpin),
+ ) -> impl Future<Output = Result<ExtentRef, Error>> + Send + 'a {
+ ExtendedFileAccess::write_extent_into(self, context, placement, source)
+ }
+
fn write_into<'a>(
&'a self,
context: &'a AccessContext,
@@ -544,7 +583,9 @@ impl<T: ExtendedFileAccess> FileAccess for T {
/// Object-safe file access used only by `ManagedVolume`.
pub trait FileAccessDyn: Send + Sync + fmt::Debug + Unpin {
/// Dyn form of [`FileAccess::info`].
- fn info_dyn(&self) -> FileAccessInfo;
+ fn info_dyn(&self) -> Option<FileAccessInfo>;
+ /// Dyn form of [`FileAccess::extent_decodings`].
+ fn extent_decodings_dyn(&self) -> Vec<ExtensionFormat>;
/// Dyn form of [`FileAccess::write_into`].
fn write_into_dyn<'a>(
&'a self,
@@ -574,10 +615,14 @@ pub trait FileAccessDyn: Send + Sync + fmt::Debug + Unpin {
}
impl<A: FileAccess> FileAccessDyn for A {
- fn info_dyn(&self) -> FileAccessInfo {
+ fn info_dyn(&self) -> Option<FileAccessInfo> {
self.info()
}
+ fn extent_decodings_dyn(&self) -> Vec<ExtensionFormat> {
+ self.extent_decodings()
+ }
+
fn write_into_dyn<'a>(
&'a self,
context: &'a AccessContext,
diff --git a/crates/core/src/managed/file.rs b/crates/core/src/managed/file.rs
index b2f745e..934d944 100644
--- a/crates/core/src/managed/file.rs
+++ b/crates/core/src/managed/file.rs
@@ -25,8 +25,8 @@ use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncSeek, AsyncSeekExt as _, Asyn
use crate::Error;
use crate::filesystem::ContentRef;
-use super::data::{FileDataRef, FileExtentWriter};
-use super::extension::{ExtentAccess as _, ExtentRef, IdentityExtentAccess, SegmentRangeRef};
+use super::data::FileDataRef;
+use super::extension::{AccessContext, ExtentRef, FileAccess, IdentityExtentAccess};
use super::object::{GcEpoch, ObjectClass};
use super::stream::RecordStreamReader;
use super::{
@@ -39,6 +39,89 @@ pub struct FileExtentReader {
tail: Option<RecordStreamReader<ExtentRef>>,
}
+/// Whole-file access used at the inside of the extension chain.
+#[derive(Clone, Copy, Debug, Default)]
+pub struct WholeFileAccess<A = IdentityExtentAccess> {
+ inner: A,
+}
+
+impl<A> WholeFileAccess<A> {
+ /// Build whole-file access over one statically composed extent access.
+ pub const fn new(inner: A) -> Self {
+ Self { inner }
+ }
+}
+
+impl<A: super::extension::ExtentAccess> FileAccess for WholeFileAccess<A> {
+ fn info(&self) -> Option<super::extension::FileAccessInfo> {
+ None
+ }
+
+ fn extent_decodings(&self) -> Vec<super::extension::ExtensionFormat> {
+ self.inner.info()
+ }
+
+ async fn write_extent_into(
+ &self,
+ context: &AccessContext,
+ placement: &mut DataSegmentPlacement<'_>,
+ source: &mut (dyn AsyncRead + Send + Unpin),
+ ) -> Result<ExtentRef, Error> {
+ self.inner.write_into(context, placement, source).await
+ }
+
+ async fn write_into(
+ &self,
+ context: &AccessContext,
+ placement: &mut DataSegmentPlacement<'_>,
+ source: &mut (dyn AsyncRead + Send + Unpin),
+ _known: &KnownContent,
+ ) -> Result<(ContentRef, FileDataRef), Error> {
+ let extent = self.write_extent_into(context, placement, source).await?;
+ let content = extent.content();
+ let reference = if content.length() == 0 {
+ FileDataRef::empty()
+ } else {
+ FileDataRef::single(extent)
+ };
+ Ok((content, reference))
+ }
+
+ async fn read(
+ &self,
+ context: &AccessContext,
+ reference: FileDataRef,
+ content: ContentRef,
+ range: Range<u64>,
+ destination: &mut (dyn AsyncWrite + Send + Unpin),
+ ) -> Result<(), Error> {
+ reference.validate(content, self.extent_decodings().len())?;
+ let Some(extent) = reference.inline_extent() else {
+ return Ok(());
+ };
+ if reference.tail().is_some() || extent.content() != content {
+ return Err(Error::corrupt(
+ "read Managed file",
+ "core file layout is not one whole-file extent",
+ ));
+ }
+ self.inner.read(context, extent, range, destination).await
+ }
+
+ async fn read_extent_from(
+ &self,
+ context: &AccessContext,
+ reader: &mut DataSegmentReader,
+ reference: ExtentRef,
+ range: Range<u64>,
+ destination: &mut (dyn AsyncWrite + Send + Unpin),
+ ) -> Result<(), Error> {
+ self.inner
+ .read_from(context, reader, reference, range, destination)
+ .await
+ }
+}
+
pub(crate) struct ReusableFile<'a> {
pub(crate) reference: FileDataRef,
pub(crate) source: &'a mut (dyn ReusableFileSource + 'a),
@@ -82,19 +165,14 @@ impl ManagedVolume {
placement: &mut DataSegmentPlacement<'_>,
source: &mut (impl AsyncRead + Send + Unpin),
known: &KnownContent,
- ) -> Result<(ContentRef, FileDataRef), Error> {
- let (content, reference) = match self.file_access.as_deref() {
- Some(access) => {
- access
- .write_into_dyn(&self.access_context, placement, source, known)
- .await
- }
- None => write_identity_file(self, placement, source, known).await,
- }?;
- Ok(match known.file(content)? {
- Some(known) => (content, known),
- None => (content, reference),
- })
+ ) -> Result<(ContentRef, FileDataRef, bool), Error> {
+ let (content, reference) = self
+ .file_access
+ .write_into_dyn(&self.access_context, placement, source, known)
+ .await?;
+ let known = known.file(content)?;
+ let reused = known.is_some();
+ Ok((content, known.unwrap_or(reference), reused))
}
pub(crate) async fn read_extent_from(
@@ -104,24 +182,9 @@ impl ManagedVolume {
range: Range<u64>,
destination: &mut (impl AsyncWrite + Send + Unpin),
) -> Result<(), Error> {
- match self.file_access.as_deref() {
- Some(access) => {
- access
- .read_extent_from_dyn(
- &self.access_context,
- reader,
- reference,
- range,
- destination,
- )
- .await
- }
- None => {
- IdentityExtentAccess
- .read_from(&self.access_context, reader, reference, range, destination)
- .await
- }
- }
+ self.file_access
+ .read_extent_from_dyn(&self.access_context, reader, reference, range, destination)
+ .await
}
/// Publish one file through the volume's partition and encoding policy.
@@ -133,14 +196,14 @@ impl ManagedVolume {
) -> Result<(ContentRef, FileDataRef), Error> {
let mut placement = self.data_placement(gc_epoch, u64::MAX);
let result = self.publish_data_into(&mut placement, source, known).await;
- let (content, reference) = match result {
+ let (content, reference, reused) = match result {
Ok(result) => result,
Err(error) => {
placement.abort().await;
return Err(error);
}
};
- if known.file(content)?.is_some() || content.length() == 0 {
+ if reused || content.length() == 0 {
placement.abort().await;
} else {
placement.finish().await?;
@@ -162,73 +225,10 @@ impl ManagedVolume {
return read_reusing_extents(self, reference, content, range, reusable, destination)
.await;
}
- if self.format.file_storage().access.is_some() {
- return self
- .file_access()?
- .read_dyn(&self.access_context, reference, content, range, destination)
- .await;
- }
- read_identity_extents(self, reference, content, range, destination).await
- }
-}
-
-async fn write_identity_file(
- volume: &ManagedVolume,
- placement: &mut DataSegmentPlacement<'_>,
- source: &mut (impl AsyncRead + Send + Unpin),
- known: &KnownContent,
-) -> Result<(ContentRef, FileDataRef), Error> {
- let target = volume.transfer_window_bytes();
- let mut logical_hasher = blake3::Hasher::new();
- let mut logical_length = 0_u64;
- let mut extents = FileExtentWriter::new(volume.operator(), placement.gc_epoch());
- loop {
- let mut bytes = Vec::with_capacity(target);
- while bytes.len() < target {
- let read = source
- .take((target - bytes.len()) as u64)
- .read_to_end(&mut bytes)
- .await
- .map_err(|error| Error::io("read Managed file stream", error))?;
- if read == 0 {
- break;
- }
- }
- if bytes.is_empty() {
- break;
- }
- logical_hasher.update(&bytes);
- let length = bytes.len() as u64;
- logical_length = logical_length
- .checked_add(length)
- .ok_or_else(|| Error::invalid("publish Managed file", "file length overflows"))?;
- let content = ContentRef::new(
- crate::filesystem::Digest::from_bytes(blake3::hash(&bytes).into()),
- length,
- );
- let extent = match known.extent(content)? {
- Some(extent) => extent,
- None => {
- let mut source = bytes.as_slice();
- IdentityExtentAccess
- .write_into(&volume.access_context, placement, &mut source)
- .await?
- }
- };
- if extent.content() != content || !extent.decoded.is_empty() {
- return Err(Error::corrupt(
- "publish Managed file",
- "identity extent does not match the observed stream",
- ));
- }
- extents.write(extent).await?;
+ self.file_access
+ .read_dyn(&self.access_context, reference, content, range, destination)
+ .await
}
- let content = ContentRef::new(
- crate::filesystem::Digest::from_bytes(logical_hasher.finalize().into()),
- logical_length,
- );
- let data = extents.close().await?;
- Ok((content, data))
}
async fn read_reusing_extents(
@@ -376,71 +376,6 @@ fn stored_extent_end(extent: &ExtentRef) -> Result<u64, Error> {
.ok_or_else(|| Error::corrupt("read Managed file", "stored extent range overflows"))
}
-async fn read_identity_extents(
- volume: &ManagedVolume,
- reference: FileDataRef,
- content: ContentRef,
- range: Range<u64>,
- destination: &mut (impl AsyncWrite + Send + Unpin),
-) -> Result<(), Error> {
- let mut extents = reference.extents(volume.operator()).await?;
- let mut logical_offset = 0_u64;
- while let Some(extent) = extents.next().await? {
- if !extent.decoded.is_empty() || extent.range.segment.class != ObjectClass::DataSegment {
- return Err(Error::corrupt(
- "read Managed file",
- "core file extent has an unsupported encoding",
- ));
- }
- let extent_end = logical_offset
- .checked_add(extent.range.stored.length())
- .ok_or_else(|| Error::corrupt("read Managed file", "extent range overflows"))?;
- if logical_offset < range.end && range.start < extent_end {
- copy_identity_range(
- volume,
- extent.range,
- range.start.saturating_sub(logical_offset)
- ..range.end.min(extent_end) - logical_offset,
- destination,
- )
- .await?;
- }
- logical_offset = extent_end;
- }
- if logical_offset != content.length() {
- return Err(Error::corrupt(
- "read Managed file",
- "extents do not cover the logical file",
- ));
- }
- Ok(())
-}
-
-async fn copy_identity_range(
- volume: &ManagedVolume,
- reference: SegmentRangeRef,
- range: Range<u64>,
- destination: &mut (impl AsyncWrite + Send + Unpin),
-) -> Result<(), Error> {
- let start = reference
- .offset
- .checked_add(range.start)
- .ok_or_else(|| Error::corrupt("read Managed file", "segment range start overflows"))?;
- let end = reference
- .offset
- .checked_add(range.end)
- .ok_or_else(|| Error::corrupt("read Managed file", "segment range end overflows"))?;
- let mut reader =
- SegmentRangeReader::open(volume.operator(), reference.segment, start..end).await?;
- if range.start == 0 && range.end == reference.stored.length() {
- reader.copy_file(reference.stored, destination).await
- } else {
- reader
- .copy_bytes(range.end - range.start, destination)
- .await
- }
-}
-
fn logical_range(file_size: u64, range: impl RangeBounds<u64>) -> Result<Range<u64>, Error> {
let start = match range.start_bound() {
Bound::Included(offset) => *offset,
diff --git a/crates/core/src/managed/head.rs b/crates/core/src/managed/head.rs
index ffba586..cfb25e9 100644
--- a/crates/core/src/managed/head.rs
+++ b/crates/core/src/managed/head.rs
@@ -43,7 +43,7 @@ pub struct ManagedVolume {
pub(super) read_gap_bytes: usize,
pub(super) worksets: WorksetOptions,
pub(super) data_placement: Option<Arc<dyn DataPlacement>>,
- pub(super) file_access: Option<Arc<dyn FileAccessDyn>>,
+ pub(super) file_access: Arc<dyn FileAccessDyn>,
pub(super) access_context: AccessContext,
pub(super) authority_access: Arc<dyn AuthorityAccessDyn>,
pub(super) authority_name: String,
@@ -115,7 +115,7 @@ impl ManagedVolume {
read_gap_bytes: usize,
worksets: WorksetOptions,
data_placement: Option<Arc<dyn DataPlacement>>,
- file_access: Option<Arc<dyn FileAccessDyn>>,
+ file_access: Arc<dyn FileAccessDyn>,
authority_access: Arc<dyn AuthorityAccessDyn>,
authority_name: String,
) -> Self {
@@ -149,15 +149,6 @@ impl ManagedVolume {
.map(|placement| placement.segment_target_bytes().get())
}
- pub(super) fn file_access(&self) -> Result<&dyn FileAccessDyn, Error> {
- self.file_access.as_deref().ok_or_else(|| {
- Error::unsupported(
- "use Managed file extension",
- "the volume file extension access is unavailable",
- )
- })
- }
-
pub(crate) const fn stream_concurrency(&self) -> usize {
self.stream_concurrency
}
@@ -182,10 +173,7 @@ impl ManagedVolume {
}
pub(crate) fn file_decoding_count(&self) -> usize {
- match &self.format.file_storage().access {
- Some(info) => info.decodings.len(),
- None => 0,
- }
+ self.file_access.extent_decodings_dyn().len()
}
pub(super) async fn initialize(&self) -> Result<(), Error> {
diff --git a/crates/core/src/managed/mod.rs b/crates/core/src/managed/mod.rs
index c3d66fa..3e60bdf 100644
--- a/crates/core/src/managed/mod.rs
+++ b/crates/core/src/managed/mod.rs
@@ -40,8 +40,8 @@ pub use authority::{
AuthorityObservation, AuthorityRoot, AuthorityRoots, CollectionFence, DefaultAuthorityAccess,
};
pub use data::{FileDataRef, FileExtentWriter};
-pub use file::FileExtentReader;
pub(crate) use file::ReusableFile;
+pub use file::{FileExtentReader, WholeFileAccess};
pub(crate) use format::ManagedFormat;
pub use gc::GcOutcome;
pub(crate) use head::ManagedObservation;
@@ -74,7 +74,7 @@ pub struct ManagedMetadata {
read_gap_bytes: usize,
worksets: WorksetOptions,
data_placement: Option<Arc<dyn DataPlacement>>,
- file_access: Option<Arc<dyn FileAccessDyn>>,
+ file_access: Arc<dyn FileAccessDyn>,
authority_access: Arc<dyn AuthorityAccessDyn>,
authority_name: String,
}
@@ -103,7 +103,9 @@ impl ManagedMetadata {
read_gap_bytes: 1024 * 1024,
worksets: WorksetOptions::new(work_memory_mib, stream_concurrency)?,
data_placement: None,
- file_access: None,
+ file_access: extension::type_erase(WholeFileAccess::new(
+ extension::IdentityExtentAccess,
+ )),
authority_access: Arc::new(DefaultAuthorityAccess),
authority_name: DEFAULT_AUTHORITY.to_owned(),
})
@@ -135,7 +137,7 @@ impl ManagedMetadata {
/// Configure one statically composed file extension access.
pub fn with_file_extension(mut self, access: impl FileAccess) -> Self {
- self.file_access = Some(extension::type_erase(access));
+ self.file_access = extension::type_erase(access);
self
}
@@ -160,7 +162,7 @@ impl ManagedMetadata {
.data_placement
.as_ref()
.map(|placement| placement.info()),
- access: self.file_access.as_ref().map(|access| access.info_dyn()),
+ access: self.file_access.info_dyn(),
};
let desired = ManagedFormat::new(
VolumeId::generate(),
@@ -244,8 +246,9 @@ impl ManagedMetadata {
));
}
}
- match (&format.file_storage().access, self.file_access.as_ref()) {
- (Some(expected), Some(access)) if access.info_dyn() == *expected => {}
+ let configured_access = self.file_access.info_dyn();
+ match (&format.file_storage().access, configured_access.as_ref()) {
+ (Some(expected), Some(configured)) if configured == expected => {}
(Some(_), _) => {
return Err(Error::unsupported(
"open Managed volume",
diff --git a/crates/core/src/sync/publication.rs b/crates/core/src/sync/publication.rs
index d985070..a1eff26 100644
--- a/crates/core/src/sync/publication.rs
+++ b/crates/core/src/sync/publication.rs
@@ -200,11 +200,11 @@ async fn read_known_content(
let NamespaceValue::RegularFile { content, data, .. } = node.value else {
continue;
};
+ files.write(&(content, data.clone()))?;
if let Some(extent) = data.inline_extent() {
records.write(&(extent.content(), extent))?;
}
if let Some(tail) = data.tail() {
- files.write(&(content, tail.object.locator, data.clone()))?;
tails.write(&tail)?;
}
}
@@ -240,15 +240,8 @@ async fn read_known_content(
let records = workset::sort(workspace, &records.finish()?, |(content, extent)| {
(*content, extent.range.segment, extent.range.offset)
})?;
- let files = workset::sort(workspace, &files.finish()?, |(content, locator, _)| {
- (*content, *locator)
- })?;
- let mut known_files = workspace.writer("known-files")?;
- let mut files = files.reader()?;
- while let Some((content, _, data)) = files.next()? {
- known_files.write(&(content, data))?;
- }
- KnownContent::build(workspace, &known_files.finish()?, &records)
+ let files = workset::sort(workspace, &files.finish()?, |(content, _)| *content)?;
+ KnownContent::build(workspace, &files, &records)
}
async fn read_extent_tail(
@@ -290,10 +283,12 @@ fn schedule_placement_lane(
let result = async {
let mut output = workspace.writer("shared-segment-publication-results")?;
let mut files = files.reader()?;
+ let mut retains_segment = false;
while let Some(file) = files.next()? {
- let (content, data) =
+ let (content, data, reused) =
publish_file_into(&volume, &mut placement, &root.join(&file.path), &known)
.await?;
+ retains_segment |= !reused && content.length() != 0;
output.write(&LocalRecord {
path: file.path,
kind: NodeKind::RegularFile,
@@ -301,7 +296,11 @@ fn schedule_placement_lane(
file: Some(LocalFile { content, data }),
})?;
}
- placement.finish().await?;
+ if retains_segment {
+ placement.finish().await?;
+ } else {
+ placement.abort().await;
+ }
output.finish()
}
.await;
diff --git a/crates/core/src/sync/transfer.rs b/crates/core/src/sync/transfer.rs
index d93253e..314bed6 100644
--- a/crates/core/src/sync/transfer.rs
+++ b/crates/core/src/sync/transfer.rs
@@ -77,7 +77,7 @@ pub(super) async fn publish_file_into(
placement: &mut DataSegmentPlacement<'_>,
path: &Path,
known: &KnownContent,
-) -> Result<(ContentRef, FileDataRef), Error> {
+) -> Result<(ContentRef, FileDataRef, bool), Error> {
let mut file = File::open(path)
.await
.map_err(|error| Error::from_io("publish local file", Some(path), error))?;
diff --git a/crates/extensions/fastcdc/src/lib.rs b/crates/extensions/fastcdc/src/lib.rs
index 988637f..433e5cc 100644
--- a/crates/extensions/fastcdc/src/lib.rs
+++ b/crates/extensions/fastcdc/src/lib.rs
@@ -24,8 +24,8 @@ use fastcdc::v2020::{
};
use ofs_core::filesystem::{ContentRef, Digest};
use ofs_core::managed::extension::{
- AccessContext, ExtensionFormat, ExtensionId, ExtentAccess, ExtentRef, FileAccess,
- FileAccessInfo, FilePartitionExtension,
+ AccessContext, ExtensionFormat, ExtensionId, ExtentRef, FileAccess, FileAccessInfo,
+ FilePartitionExtension,
};
use ofs_core::managed::{
DataSegmentPlacement, DataSegmentReader, FileDataRef, FileExtentWriter, KnownContent,
@@ -129,7 +129,7 @@ impl FastCdcExtension {
}
}
-impl<A: ExtentAccess> FilePartitionExtension<A> for FastCdcExtension {
+impl<A: FileAccess> FilePartitionExtension<A> for FastCdcExtension {
type ExtendedAccess = FastCdcAccess<A>;
fn extend(&self, inner: A) -> Self::ExtendedAccess {
@@ -140,22 +140,37 @@ impl<A: ExtentAccess> FilePartitionExtension<A> for FastCdcExtension {
}
}
-/// FastCDC file access composed over one extent encoding.
+/// FastCDC file access layered over the common whole-file access.
#[derive(Clone, Debug)]
pub struct FastCdcAccess<A> {
inner: A,
configuration: Configuration,
}
-impl<A: ExtentAccess> FileAccess for FastCdcAccess<A> {
- fn info(&self) -> FileAccessInfo {
- FileAccessInfo {
+impl<A: FileAccess> FileAccess for FastCdcAccess<A> {
+ fn info(&self) -> Option<FileAccessInfo> {
+ Some(FileAccessInfo {
partitioning: ExtensionFormat {
id: FASTCDC_EXTENSION_ID,
configuration: self.configuration.encode(),
},
- decodings: self.inner.info(),
- }
+ decodings: self.inner.extent_decodings(),
+ })
+ }
+
+ fn extent_decodings(&self) -> Vec<ExtensionFormat> {
+ self.inner.extent_decodings()
+ }
+
+ async fn write_extent_into(
+ &self,
+ context: &AccessContext,
+ placement: &mut DataSegmentPlacement<'_>,
+ source: &mut (dyn AsyncRead + Send + Unpin),
+ ) -> Result<ExtentRef, Error> {
+ self.inner
+ .write_extent_into(context, placement, source)
+ .await
}
async fn write_into(
@@ -171,7 +186,7 @@ impl<A: ExtentAccess> FileAccess for FastCdcAccess<A> {
let mut chunks = std::pin::pin!(chunker.as_stream());
let mut logical_hasher = blake3::Hasher::new();
let mut logical_length = 0_u64;
- let decoding_count = self.inner.info().len();
+ let decoding_count = self.inner.extent_decodings().len();
while let Some(chunk) = chunks.next().await {
let chunk = chunk.map_err(|error| {
@@ -206,7 +221,7 @@ impl<A: ExtentAccess> FileAccess for FastCdcAccess<A> {
None => {
let mut bytes = chunk.data.as_slice();
self.inner
- .write_into(context, placement, &mut bytes)
+ .write_extent_into(context, placement, &mut bytes)
.await?
}
};
@@ -253,7 +268,7 @@ impl<A: ExtentAccess> FileAccess for FastCdcAccess<A> {
let mut extents = reference.extents(context.operator()).await?;
let mut pending = None;
let mut expected_offset = 0_u64;
- let decoding_count = self.inner.info().len();
+ let decoding_count = self.inner.extent_decodings().len();
let maximum = self.configuration.2;
loop {
let extent = match pending.take() {
@@ -296,7 +311,7 @@ impl<A: ExtentAccess> FileAccess for FastCdcAccess<A> {
DataSegmentReader::open(context.operator(), segment, run_start..run_end).await?;
for (extent, offset, extent_end) in run {
self.inner
- .read_from(
+ .read_extent_from(
context,
&mut reader,
extent,
@@ -327,9 +342,13 @@ impl<A: ExtentAccess> FileAccess for FastCdcAccess<A> {
range: Range<u64>,
destination: &mut (dyn AsyncWrite + Send + Unpin),
) -> Result<(), Error> {
- validate_extent(&reference, self.inner.info().len(), self.configuration.2)?;
+ validate_extent(
+ &reference,
+ self.inner.extent_decodings().len(),
+ self.configuration.2,
+ )?;
self.inner
- .read_from(context, reader, reference, range, destination)
+ .read_extent_from(context, reader, reference, range, destination)
.await
}
}
diff --git a/crates/extras/src/lib.rs b/crates/extras/src/lib.rs
index 25f281c..11cfec1 100644
--- a/crates/extras/src/lib.rs
+++ b/crates/extras/src/lib.rs
@@ -23,15 +23,14 @@ use std::num::NonZeroU64;
use ofs_core::filesystem::VolumeId;
#[cfg(feature = "zstd")]
use ofs_core::managed::extension::ExtentExtension as _;
-#[cfg(all(feature = "fastcdc", feature = "tracing"))]
+#[cfg(feature = "tracing")]
use ofs_core::managed::extension::FileAccessExtension as _;
#[cfg(feature = "fastcdc")]
-use ofs_core::managed::extension::{
- FileAccess, FileAccessInfo, FilePartitionExtension as _, IdentityExtentAccess,
-};
+use ofs_core::managed::extension::FilePartitionExtension as _;
+use ofs_core::managed::extension::{FileAccess, FileAccessInfo, IdentityExtentAccess};
#[cfg(feature = "branch")]
use ofs_core::managed::{AuthorityExtension as _, DefaultAuthorityAccess};
-use ofs_core::managed::{ManagedMetadata, ManagedVolume};
+use ofs_core::managed::{ManagedMetadata, ManagedVolume, WholeFileAccess};
use ofs_core::{Error, ErrorKind};
#[cfg(feature = "branch")]
@@ -44,7 +43,7 @@ use ofs_ext_fastcdc::FastCdcExtension;
pub use ofs_ext_pack::PACK_EXTENSION_ID;
#[cfg(feature = "pack")]
use ofs_ext_pack::PackExtension;
-#[cfg(all(feature = "fastcdc", feature = "tracing"))]
+#[cfg(feature = "tracing")]
use ofs_ext_tracing::TracingExtension;
#[cfg(feature = "zstd")]
use ofs_ext_zstd::ZstdExtension;
@@ -133,33 +132,29 @@ fn configure_new_placement(
metadata
}
-#[cfg(feature = "fastcdc")]
fn configure_new_file_access(
metadata: ManagedMetadata,
options: ExtensionOptions,
) -> Result<ManagedMetadata, Error> {
+ let _ = options;
+ #[cfg(feature = "zstd")]
+ let access =
+ WholeFileAccess::new(ZstdExtension::new(options.zstd_level)?.extend(IdentityExtentAccess));
+ #[cfg(not(feature = "zstd"))]
+ let access = WholeFileAccess::new(IdentityExtentAccess);
+
+ #[cfg(feature = "fastcdc")]
let partitioning = FastCdcExtension::new(
options.fastcdc_minimum_bytes,
options.fastcdc_average_bytes,
options.fastcdc_maximum_bytes,
)?;
- #[cfg(feature = "zstd")]
- let access =
- partitioning.extend(ZstdExtension::new(options.zstd_level)?.extend(IdentityExtentAccess));
- #[cfg(not(feature = "zstd"))]
- let access = partitioning.extend(IdentityExtentAccess);
+ #[cfg(feature = "fastcdc")]
+ let access = partitioning.extend(access);
Ok(with_file_access(metadata, access))
}
-#[cfg(not(feature = "fastcdc"))]
-fn configure_new_file_access(
- metadata: ManagedMetadata,
- _options: ExtensionOptions,
-) -> Result<ManagedMetadata, Error> {
- Ok(metadata)
-}
-
#[cfg(feature = "pack")]
fn configure_recorded_placement(
metadata: ManagedMetadata,
@@ -186,24 +181,36 @@ fn configure_recorded_placement(
Ok(metadata)
}
-#[cfg(feature = "fastcdc")]
fn configure_recorded_file_access(
metadata: ManagedMetadata,
info: Option<&FileAccessInfo>,
) -> Result<ManagedMetadata, Error> {
let Some(info) = info else {
- return Ok(metadata);
+ return Ok(with_file_access(
+ metadata,
+ WholeFileAccess::new(IdentityExtentAccess),
+ ));
};
+ configure_recorded_partitioning(metadata, info)
+}
+
+#[cfg(feature = "fastcdc")]
+fn configure_recorded_partitioning(
+ metadata: ManagedMetadata,
+ info: &FileAccessInfo,
+) -> Result<ManagedMetadata, Error> {
let partitioning = FastCdcExtension::from_format(&info.partitioning)?;
match info.decodings.as_slice() {
[] => Ok(with_file_access(
metadata,
- partitioning.extend(IdentityExtentAccess),
+ partitioning.extend(WholeFileAccess::new(IdentityExtentAccess)),
)),
#[cfg(feature = "zstd")]
[decoding] => Ok(with_file_access(
metadata,
- partitioning.extend(ZstdExtension::from_format(decoding)?.extend(IdentityExtentAccess)),
+ partitioning.extend(WholeFileAccess::new(
+ ZstdExtension::from_format(decoding)?.extend(IdentityExtentAccess),
+ )),
)),
_ => Err(Error::new(
ErrorKind::Unsupported,
@@ -214,21 +221,18 @@ fn configure_recorded_file_access(
}
#[cfg(not(feature = "fastcdc"))]
-fn configure_recorded_file_access(
+fn configure_recorded_partitioning(
metadata: ManagedMetadata,
- info: Option<&ofs_core::managed::extension::FileAccessInfo>,
+ _info: &FileAccessInfo,
) -> Result<ManagedMetadata, Error> {
- if info.is_some() {
- return Err(Error::new(
- ErrorKind::Unsupported,
- "open Managed volume",
- "the file access extension is not compiled",
- ));
- }
- Ok(metadata)
+ let _ = metadata;
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "open Managed volume",
+ "the file access extension is not compiled",
+ ))
}
-#[cfg(feature = "fastcdc")]
fn with_file_access(metadata: ManagedMetadata, access: impl FileAccess) -> ManagedMetadata {
#[cfg(feature = "tracing")]
let access = TracingExtension::new().extend(access);