1use super::{
4 BlobReadError, DecodeLimits, ExtentKind, ExtentRangeRequest, ExtentReadError, ExtentSlice,
5 FileKind, FilePayload, FileRecord, plan_extent_range_async, read_blob_range_async,
6};
7use crate::AsyncObjectStore;
8use crate::cancellation::CancellationToken;
9use crate::performance::{OperationFailure, WorkBudget, WorkCounters, WorkError};
10use crate::storage::ByteRange;
11use bytes::Bytes;
12use thiserror::Error;
13
14#[derive(Clone, Copy, Debug, Eq, PartialEq)]
16pub struct FileRangeRequest {
17 pub record: FileRecord,
19 pub range: ByteRange,
21 pub maximum_spans: u32,
23 pub limits: DecodeLimits,
25 pub budget: WorkBudget,
27}
28
29#[derive(Clone, Debug, Eq, PartialEq)]
31pub struct FileRangeRead {
32 pub bytes: Bytes,
34 pub work: WorkCounters,
36}
37
38pub async fn read_file_range_async<S: AsyncObjectStore>(
45 store: &S,
46 request: FileRangeRequest,
47 cancellation: &CancellationToken,
48) -> Result<FileRangeRead, FileRangeReadFailure> {
49 cancellation
50 .check()
51 .map_err(|_| failed(FileRangeReadError::Cancelled, WorkCounters::default()))?;
52 if request.record.kind != FileKind::Regular {
53 return Err(failed(
54 FileRangeReadError::NotRegular,
55 WorkCounters::default(),
56 ));
57 }
58 match request.record.payload {
59 FilePayload::InlineRegular(data) => read_inline(data.as_bytes(), &request),
60 FilePayload::Regular {
61 logical_bytes,
62 extents,
63 } => read_sparse(store, logical_bytes, extents, &request, cancellation).await,
64 _ => Err(failed(
65 FileRangeReadError::NotRegular,
66 WorkCounters::default(),
67 )),
68 }
69}
70
71fn read_inline(
72 bytes: &[u8],
73 request: &FileRangeRequest,
74) -> Result<FileRangeRead, FileRangeReadFailure> {
75 let end = request
76 .range
77 .offset
78 .checked_add(request.range.length)
79 .ok_or_else(|| failed(FileRangeReadError::InvalidRange, WorkCounters::default()))?;
80 if end > u64::try_from(bytes.len()).unwrap_or(u64::MAX) {
81 return Err(failed(
82 FileRangeReadError::InvalidRange,
83 WorkCounters::default(),
84 ));
85 }
86 let start = usize::try_from(request.range.offset)
87 .map_err(|_| failed(FileRangeReadError::InvalidRange, WorkCounters::default()))?;
88 let end = usize::try_from(end)
89 .map_err(|_| failed(FileRangeReadError::InvalidRange, WorkCounters::default()))?;
90 let length = u64::try_from(end - start).unwrap_or(u64::MAX);
91 let work = WorkCounters {
92 bytes_copied: length,
93 output_bytes: length,
94 items_returned: 1,
95 allocation_operations: u64::from(length != 0),
96 peak_allocation_bytes: length,
97 ..WorkCounters::default()
98 };
99 work.verify(request.budget)
100 .map_err(|error| failed(error.into(), WorkCounters::default()))?;
101 #[allow(
102 clippy::indexing_slicing,
103 reason = "end <= bytes.len() is checked above (end > bytes.len() as u64 returns InvalidRange before this point) and start <= end since start and end are both derived from the same validated request.range, so bytes[start..end] cannot panic"
104 )]
105 let region = &bytes[start..end];
106 Ok(FileRangeRead {
107 bytes: Bytes::copy_from_slice(region),
108 work,
109 })
110}
111
112async fn read_sparse<S: AsyncObjectStore>(
113 store: &S,
114 logical_bytes: u64,
115 extents: crate::ObjectId,
116 request: &FileRangeRequest,
117 cancellation: &CancellationToken,
118) -> Result<FileRangeRead, FileRangeReadFailure> {
119 let plan = plan_extent_range_async(
120 store,
121 ExtentRangeRequest {
122 root: extents,
123 file_size: logical_bytes,
124 range: request.range,
125 maximum_spans: request.maximum_spans,
126 limits: request.limits,
127 budget: request.budget,
128 },
129 cancellation,
130 )
131 .await
132 .map_err(|failure| failure.map_with_prior_work(WorkCounters::default(), Into::into))?;
133 let mut work = plan.work;
134 let output_len = usize::try_from(request.range.length)
135 .map_err(|_| failed(FileRangeReadError::InvalidRange, work))?;
136 let live_output = request.range.length;
137 let initial_peak = plan
138 .retained_allocation_bytes
139 .checked_add(live_output)
140 .ok_or_else(|| failed(FileRangeReadError::Work(WorkError::Overflow), work))?;
141 work = add(
142 work,
143 WorkCounters {
144 allocation_operations: u64::from(output_len != 0),
145 peak_allocation_bytes: initial_peak,
146 ..WorkCounters::default()
147 },
148 )?;
149 work.verify(request.budget)
150 .map_err(|error| failed(error.into(), work))?;
151 let mut output = Vec::new();
152 output
153 .try_reserve_exact(output_len)
154 .map_err(|_| failed(FileRangeReadError::AllocationFailed, work))?;
155 output.resize(output_len, 0);
156 for span in &plan.spans {
157 work = copy_content_span(
158 store,
159 span,
160 work,
161 initial_peak,
162 &mut output,
163 request,
164 cancellation,
165 )
166 .await?;
167 }
168 work = add(
169 work,
170 WorkCounters {
171 output_bytes: request.range.length,
172 items_returned: 1,
173 ..WorkCounters::default()
174 },
175 )?;
176 work.verify(request.budget)
177 .map_err(|error| failed(error.into(), work))?;
178 Ok(FileRangeRead {
179 bytes: Bytes::from(output),
180 work,
181 })
182}
183
184async fn copy_content_span<S: AsyncObjectStore>(
186 store: &S,
187 span: &ExtentSlice,
188 mut work: WorkCounters,
189 initial_peak: u64,
190 output: &mut [u8],
191 request: &FileRangeRequest,
192 cancellation: &CancellationToken,
193) -> Result<WorkCounters, FileRangeReadFailure> {
194 let ExtentKind::Content {
195 object,
196 object_offset,
197 } = span.kind
198 else {
199 return Ok(work);
200 };
201 let nested = read_blob_range_async(
202 store,
203 object,
204 ByteRange {
205 offset: object_offset,
206 length: span.length,
207 },
208 request.limits,
209 remaining(work, request.budget)?,
210 cancellation,
211 )
212 .await
213 .map_err(|failure| failure.map_with_prior_work(work, Into::into))?;
214 let simultaneous = initial_peak
215 .checked_add(nested.work.peak_allocation_bytes)
216 .ok_or_else(|| failed(FileRangeReadError::Work(WorkError::Overflow), work))?;
217 let mut nested_work = nested.work;
218 nested_work.peak_allocation_bytes = 0;
219 nested_work.output_bytes = 0;
220 work = add(work, nested_work)?;
221 work.peak_allocation_bytes = work.peak_allocation_bytes.max(simultaneous);
222 let destination = usize::try_from(span.offset - request.range.offset)
223 .map_err(|_| failed(FileRangeReadError::InvalidRange, work))?;
224 let end = destination
225 .checked_add(nested.bytes.len())
226 .ok_or_else(|| failed(FileRangeReadError::InvalidRange, work))?;
227 #[allow(
228 clippy::indexing_slicing,
229 reason = "RangeMachine::finish (range.rs) only returns an ExtentPlan when spans exactly and contiguously tile [request.range.offset, request.range.offset + request.range.length) with no gaps or overlaps (ExtentReadError::IncompleteCoverage otherwise), and BlobRangeMachine::finish (blob.rs) only returns Ok when nested.bytes.len() == span.length exactly (BlobReadError::IncompleteCoverage otherwise); together these guarantee destination..end falls within output, which was sized to request.range.length"
230 )]
231 output[destination..end].copy_from_slice(&nested.bytes);
232 work = add(
233 work,
234 WorkCounters {
235 bytes_copied: span.length,
236 ..WorkCounters::default()
237 },
238 )?;
239 work.verify(request.budget)
240 .map_err(|error| failed(error.into(), work))?;
241 Ok(work)
242}
243
244fn add(prior: WorkCounters, next: WorkCounters) -> Result<WorkCounters, FileRangeReadFailure> {
245 prior
246 .checked_add(next)
247 .map_err(|error| failed(error.into(), prior))
248}
249
250fn remaining(work: WorkCounters, budget: WorkBudget) -> Result<WorkBudget, FileRangeReadFailure> {
251 work.remaining(budget)
252 .map_err(|error| failed(error.into(), work))
253}
254
255pub type FileRangeReadFailure = OperationFailure<FileRangeReadError>;
257
258#[derive(Debug, Error)]
260pub enum FileRangeReadError {
261 #[error("file range read requires a regular file")]
263 NotRegular,
264 #[error("file range is invalid")]
266 InvalidRange,
267 #[error("file range read was cancelled")]
269 Cancelled,
270 #[error("file range output allocation failed")]
272 AllocationFailed,
273 #[error(transparent)]
275 Extent(#[from] ExtentReadError),
276 #[error(transparent)]
278 Blob(#[from] BlobReadError),
279 #[error(transparent)]
281 Work(#[from] WorkError),
282}
283
284fn failed(error: FileRangeReadError, work: WorkCounters) -> FileRangeReadFailure {
285 OperationFailure::new(error, work)
286}
287
288#[cfg(test)]
289#[path = "tests/file_read.rs"]
290mod tests;