Line data Source code
1 : //!
2 : //! VirtualFile is like a normal File, but it's not bound directly to
3 : //! a file descriptor. Instead, the file is opened when it's read from,
4 : //! and if too many files are open globally in the system, least-recently
5 : //! used ones are closed.
6 : //!
7 : //! To track which files have been recently used, we use the clock algorithm
8 : //! with a 'recently_used' flag on each slot.
9 : //!
10 : //! This is similar to PostgreSQL's virtual file descriptor facility in
11 : //! src/backend/storage/file/fd.c
12 : //!
13 : use crate::context::RequestContext;
14 : use crate::metrics::{StorageIoOperation, STORAGE_IO_SIZE, STORAGE_IO_TIME_METRIC};
15 :
16 : use crate::page_cache::PageWriteGuard;
17 : use crate::tenant::TENANTS_SEGMENT_NAME;
18 : use camino::{Utf8Path, Utf8PathBuf};
19 : use once_cell::sync::OnceCell;
20 : use pageserver_api::shard::TenantShardId;
21 : use std::fs::File;
22 : use std::io::{Error, ErrorKind, Seek, SeekFrom};
23 : use tokio_epoll_uring::{BoundedBuf, IoBuf, IoBufMut, Slice};
24 :
25 : use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, OwnedFd, RawFd};
26 : use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
27 : use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
28 : use tokio::time::Instant;
29 :
30 : pub use pageserver_api::models::virtual_file as api;
31 : pub(crate) mod io_engine;
32 : pub use io_engine::feature_test as io_engine_feature_test;
33 : pub use io_engine::FeatureTestResult as IoEngineFeatureTestResult;
34 : mod metadata;
35 : mod open_options;
36 : use self::owned_buffers_io::write::OwnedAsyncWriter;
37 : pub(crate) use io_engine::IoEngineKind;
38 : pub(crate) use metadata::Metadata;
39 : pub(crate) use open_options::*;
40 :
41 : pub(crate) mod owned_buffers_io {
42 : //! Abstractions for IO with owned buffers.
43 : //!
44 : //! Not actually tied to [`crate::virtual_file`] specifically, but, it's the primary
45 : //! reason we need this abstraction.
46 : //!
47 : //! Over time, this could move into the `tokio-epoll-uring` crate, maybe `uring-common`,
48 : //! but for the time being we're proving out the primitives in the neon.git repo
49 : //! for faster iteration.
50 :
51 : pub(crate) mod write;
52 : pub(crate) mod util {
53 : pub(crate) mod size_tracking_writer;
54 : }
55 : }
56 :
57 : ///
58 : /// A virtual file descriptor. You can use this just like std::fs::File, but internally
59 : /// the underlying file is closed if the system is low on file descriptors,
60 : /// and re-opened when it's accessed again.
61 : ///
62 : /// Like with std::fs::File, multiple threads can read/write the file concurrently,
63 : /// holding just a shared reference the same VirtualFile, using the read_at() / write_at()
64 : /// functions from the FileExt trait. But the functions from the Read/Write/Seek traits
65 : /// require a mutable reference, because they modify the "current position".
66 : ///
67 : /// Each VirtualFile has a physical file descriptor in the global OPEN_FILES array, at the
68 : /// slot that 'handle points to, if the underlying file is currently open. If it's not
69 : /// currently open, the 'handle' can still point to the slot where it was last kept. The
70 : /// 'tag' field is used to detect whether the handle still is valid or not.
71 : ///
72 : #[derive(Debug)]
73 : pub struct VirtualFile {
74 : /// Lazy handle to the global file descriptor cache. The slot that this points to
75 : /// might contain our File, or it may be empty, or it may contain a File that
76 : /// belongs to a different VirtualFile.
77 : handle: RwLock<SlotHandle>,
78 :
79 : /// Current file position
80 : pos: u64,
81 :
82 : /// File path and options to use to open it.
83 : ///
84 : /// Note: this only contains the options needed to re-open it. For example,
85 : /// if a new file is created, we only pass the create flag when it's initially
86 : /// opened, in the VirtualFile::create() function, and strip the flag before
87 : /// storing it here.
88 : pub path: Utf8PathBuf,
89 : open_options: OpenOptions,
90 :
91 : // These are strings becase we only use them for metrics, and those expect strings.
92 : // It makes no sense for us to constantly turn the `TimelineId` and `TenantId` into
93 : // strings.
94 : tenant_id: String,
95 : shard_id: String,
96 : timeline_id: String,
97 : }
98 :
99 : #[derive(Debug, PartialEq, Clone, Copy)]
100 : struct SlotHandle {
101 : /// Index into OPEN_FILES.slots
102 : index: usize,
103 :
104 : /// Value of 'tag' in the slot. If slot's tag doesn't match, then the slot has
105 : /// been recycled and no longer contains the FD for this virtual file.
106 : tag: u64,
107 : }
108 :
109 : /// OPEN_FILES is the global array that holds the physical file descriptors that
110 : /// are currently open. Each slot in the array is protected by a separate lock,
111 : /// so that different files can be accessed independently. The lock must be held
112 : /// in write mode to replace the slot with a different file, but a read mode
113 : /// is enough to operate on the file, whether you're reading or writing to it.
114 : ///
115 : /// OPEN_FILES starts in uninitialized state, and it's initialized by
116 : /// the virtual_file::init() function. It must be called exactly once at page
117 : /// server startup.
118 : static OPEN_FILES: OnceCell<OpenFiles> = OnceCell::new();
119 :
120 : struct OpenFiles {
121 : slots: &'static [Slot],
122 :
123 : /// clock arm for the clock algorithm
124 : next: AtomicUsize,
125 : }
126 :
127 : struct Slot {
128 : inner: RwLock<SlotInner>,
129 :
130 : /// has this file been used since last clock sweep?
131 : recently_used: AtomicBool,
132 : }
133 :
134 : struct SlotInner {
135 : /// Counter that's incremented every time a different file is stored here.
136 : /// To avoid the ABA problem.
137 : tag: u64,
138 :
139 : /// the underlying file
140 : file: Option<OwnedFd>,
141 : }
142 :
143 : /// Impl of [`tokio_epoll_uring::IoBuf`] and [`tokio_epoll_uring::IoBufMut`] for [`PageWriteGuard`].
144 : struct PageWriteGuardBuf {
145 : page: PageWriteGuard<'static>,
146 : init_up_to: usize,
147 : }
148 : // Safety: the [`PageWriteGuard`] gives us exclusive ownership of the page cache slot,
149 : // and the location remains stable even if [`Self`] or the [`PageWriteGuard`] is moved.
150 : unsafe impl tokio_epoll_uring::IoBuf for PageWriteGuardBuf {
151 113980 : fn stable_ptr(&self) -> *const u8 {
152 113980 : self.page.as_ptr()
153 113980 : }
154 341940 : fn bytes_init(&self) -> usize {
155 341940 : self.init_up_to
156 341940 : }
157 113980 : fn bytes_total(&self) -> usize {
158 113980 : self.page.len()
159 113980 : }
160 : }
161 : // Safety: see above, plus: the ownership of [`PageWriteGuard`] means exclusive access,
162 : // hence it's safe to hand out the `stable_mut_ptr()`.
163 : unsafe impl tokio_epoll_uring::IoBufMut for PageWriteGuardBuf {
164 113980 : fn stable_mut_ptr(&mut self) -> *mut u8 {
165 113980 : self.page.as_mut_ptr()
166 113980 : }
167 :
168 113980 : unsafe fn set_init(&mut self, pos: usize) {
169 113980 : assert!(pos <= self.page.len());
170 113980 : self.init_up_to = pos;
171 113980 : }
172 : }
173 :
174 : impl OpenFiles {
175 : /// Find a slot to use, evicting an existing file descriptor if needed.
176 : ///
177 : /// On return, we hold a lock on the slot, and its 'tag' has been updated
178 : /// recently_used has been set. It's all ready for reuse.
179 189109 : async fn find_victim_slot(&self) -> (SlotHandle, RwLockWriteGuard<SlotInner>) {
180 189109 : //
181 189109 : // Run the clock algorithm to find a slot to replace.
182 189109 : //
183 189109 : let num_slots = self.slots.len();
184 189109 : let mut retries = 0;
185 : let mut slot;
186 : let mut slot_guard;
187 : let index;
188 1971005 : loop {
189 1971005 : let next = self.next.fetch_add(1, Ordering::AcqRel) % num_slots;
190 1971005 : slot = &self.slots[next];
191 1971005 :
192 1971005 : // If the recently_used flag on this slot is set, continue the clock
193 1971005 : // sweep. Otherwise try to use this slot. If we cannot acquire the
194 1971005 : // lock, also continue the clock sweep.
195 1971005 : //
196 1971005 : // We only continue in this manner for a while, though. If we loop
197 1971005 : // through the array twice without finding a victim, just pick the
198 1971005 : // next slot and wait until we can reuse it. This way, we avoid
199 1971005 : // spinning in the extreme case that all the slots are busy with an
200 1971005 : // I/O operation.
201 1971005 : if retries < num_slots * 2 {
202 1902675 : if !slot.recently_used.swap(false, Ordering::Release) {
203 1697887 : if let Ok(guard) = slot.inner.try_write() {
204 120779 : slot_guard = guard;
205 120779 : index = next;
206 120779 : break;
207 1577108 : }
208 204788 : }
209 1781896 : retries += 1;
210 : } else {
211 68330 : slot_guard = slot.inner.write().await;
212 68330 : index = next;
213 68330 : break;
214 : }
215 : }
216 :
217 : //
218 : // We now have the victim slot locked. If it was in use previously, close the
219 : // old file.
220 : //
221 189109 : if let Some(old_file) = slot_guard.file.take() {
222 184717 : // the normal path of dropping VirtualFile uses "close", use "close-by-replace" here to
223 184717 : // distinguish the two.
224 184717 : STORAGE_IO_TIME_METRIC
225 184717 : .get(StorageIoOperation::CloseByReplace)
226 184717 : .observe_closure_duration(|| drop(old_file));
227 184717 : }
228 :
229 : // Prepare the slot for reuse and return it
230 189109 : slot_guard.tag += 1;
231 189109 : slot.recently_used.store(true, Ordering::Relaxed);
232 189109 : (
233 189109 : SlotHandle {
234 189109 : index,
235 189109 : tag: slot_guard.tag,
236 189109 : },
237 189109 : slot_guard,
238 189109 : )
239 189109 : }
240 : }
241 :
242 : /// Identify error types that should alwways terminate the process. Other
243 : /// error types may be elegible for retry.
244 0 : pub(crate) fn is_fatal_io_error(e: &std::io::Error) -> bool {
245 0 : use nix::errno::Errno::*;
246 0 : match e.raw_os_error().map(nix::errno::from_i32) {
247 : Some(EIO) => {
248 : // Terminate on EIO because we no longer trust the device to store
249 : // data safely, or to uphold persistence guarantees on fsync.
250 0 : true
251 : }
252 : Some(EROFS) => {
253 : // Terminate on EROFS because a filesystem is usually remounted
254 : // readonly when it has experienced some critical issue, so the same
255 : // logic as EIO applies.
256 0 : true
257 : }
258 : Some(EACCES) => {
259 : // Terminate on EACCESS because we should always have permissions
260 : // for our own data dir: if we don't, then we can't do our job and
261 : // need administrative intervention to fix permissions. Terminating
262 : // is the best way to make sure we stop cleanly rather than going
263 : // into infinite retry loops, and will make it clear to the outside
264 : // world that we need help.
265 0 : true
266 : }
267 : _ => {
268 : // Treat all other local file I/O errors are retryable. This includes:
269 : // - ENOSPC: we stay up and wait for eviction to free some space
270 : // - EINVAL, EBADF, EBADFD: this is a code bug, not a filesystem/hardware issue
271 : // - WriteZero, Interrupted: these are used internally VirtualFile
272 0 : false
273 : }
274 : }
275 0 : }
276 :
277 : /// Call this when the local filesystem gives us an error with an external
278 : /// cause: this includes EIO, EROFS, and EACCESS: all these indicate either
279 : /// bad storage or bad configuration, and we can't fix that from inside
280 : /// a running process.
281 0 : pub(crate) fn on_fatal_io_error(e: &std::io::Error, context: &str) -> ! {
282 0 : tracing::error!("Fatal I/O error: {e}: {context})");
283 0 : std::process::abort();
284 : }
285 :
286 : pub(crate) trait MaybeFatalIo<T> {
287 : fn maybe_fatal_err(self, context: &str) -> std::io::Result<T>;
288 : fn fatal_err(self, context: &str) -> T;
289 : }
290 :
291 : impl<T> MaybeFatalIo<T> for std::io::Result<T> {
292 : /// Terminate the process if the result is an error of a fatal type, else pass it through
293 : ///
294 : /// This is appropriate for writes, where we typically want to die on EIO/ACCES etc, but
295 : /// not on ENOSPC.
296 22 : fn maybe_fatal_err(self, context: &str) -> std::io::Result<T> {
297 22 : if let Err(e) = &self {
298 0 : if is_fatal_io_error(e) {
299 0 : on_fatal_io_error(e, context);
300 0 : }
301 22 : }
302 22 : self
303 22 : }
304 :
305 : /// Terminate the process on any I/O error.
306 : ///
307 : /// This is appropriate for reads on files that we know exist: they should always work.
308 2350 : fn fatal_err(self, context: &str) -> T {
309 2350 : match self {
310 2350 : Ok(v) => v,
311 0 : Err(e) => {
312 0 : on_fatal_io_error(&e, context);
313 : }
314 : }
315 2350 : }
316 : }
317 :
318 : /// Observe duration for the given storage I/O operation
319 : ///
320 : /// Unlike `observe_closure_duration`, this supports async,
321 : /// where "support" means that we measure wall clock time.
322 : macro_rules! observe_duration {
323 : ($op:expr, $($body:tt)*) => {{
324 : let instant = Instant::now();
325 : let result = $($body)*;
326 : let elapsed = instant.elapsed().as_secs_f64();
327 : STORAGE_IO_TIME_METRIC
328 : .get($op)
329 : .observe(elapsed);
330 : result
331 : }}
332 : }
333 :
334 : macro_rules! with_file {
335 : ($this:expr, $op:expr, | $ident:ident | $($body:tt)*) => {{
336 : let $ident = $this.lock_file().await?;
337 : observe_duration!($op, $($body)*)
338 : }};
339 : ($this:expr, $op:expr, | mut $ident:ident | $($body:tt)*) => {{
340 : let mut $ident = $this.lock_file().await?;
341 : observe_duration!($op, $($body)*)
342 : }};
343 : }
344 :
345 : impl VirtualFile {
346 : /// Open a file in read-only mode. Like File::open.
347 2274 : pub async fn open<P: AsRef<Utf8Path>>(
348 2274 : path: P,
349 2274 : ctx: &RequestContext,
350 2274 : ) -> Result<VirtualFile, std::io::Error> {
351 2274 : Self::open_with_options(path.as_ref(), OpenOptions::new().read(true), ctx).await
352 2274 : }
353 :
354 : /// Create a new file for writing. If the file exists, it will be truncated.
355 : /// Like File::create.
356 1353 : pub async fn create<P: AsRef<Utf8Path>>(
357 1353 : path: P,
358 1353 : ctx: &RequestContext,
359 1353 : ) -> Result<VirtualFile, std::io::Error> {
360 1353 : Self::open_with_options(
361 1353 : path.as_ref(),
362 1353 : OpenOptions::new().write(true).create(true).truncate(true),
363 1353 : ctx,
364 1353 : )
365 698 : .await
366 1353 : }
367 :
368 : /// Open a file with given options.
369 : ///
370 : /// Note: If any custom flags were set in 'open_options' through OpenOptionsExt,
371 : /// they will be applied also when the file is subsequently re-opened, not only
372 : /// on the first time. Make sure that's sane!
373 5515 : pub async fn open_with_options<P: AsRef<Utf8Path>>(
374 5515 : path: P,
375 5515 : open_options: &OpenOptions,
376 5515 : _ctx: &RequestContext, /* TODO: carry a pointer to the metrics in the RequestContext instead of the parsing https://github.com/neondatabase/neon/issues/6107 */
377 5515 : ) -> Result<VirtualFile, std::io::Error> {
378 5515 : let path_ref = path.as_ref();
379 5515 : let path_str = path_ref.to_string();
380 5515 : let parts = path_str.split('/').collect::<Vec<&str>>();
381 5515 : let (tenant_id, shard_id, timeline_id) =
382 5515 : if parts.len() > 5 && parts[parts.len() - 5] == TENANTS_SEGMENT_NAME {
383 3897 : let tenant_shard_part = parts[parts.len() - 4];
384 3897 : let (tenant_id, shard_id) = match tenant_shard_part.parse::<TenantShardId>() {
385 3897 : Ok(tenant_shard_id) => (
386 3897 : tenant_shard_id.tenant_id.to_string(),
387 3897 : format!("{}", tenant_shard_id.shard_slug()),
388 3897 : ),
389 : Err(_) => {
390 : // Malformed path: this ID is just for observability, so tolerate it
391 : // and pass through
392 0 : (tenant_shard_part.to_string(), "*".to_string())
393 : }
394 : };
395 3897 : (tenant_id, shard_id, parts[parts.len() - 2].to_string())
396 : } else {
397 1618 : ("*".to_string(), "*".to_string(), "*".to_string())
398 : };
399 5515 : let (handle, mut slot_guard) = get_open_files().find_victim_slot().await;
400 :
401 : // NB: there is also StorageIoOperation::OpenAfterReplace which is for the case
402 : // where our caller doesn't get to use the returned VirtualFile before its
403 : // slot gets re-used by someone else.
404 5515 : let file = observe_duration!(StorageIoOperation::Open, {
405 5515 : open_options.open(path_ref.as_std_path()).await?
406 : });
407 :
408 : // Strip all options other than read and write.
409 : //
410 : // It would perhaps be nicer to check just for the read and write flags
411 : // explicitly, but OpenOptions doesn't contain any functions to read flags,
412 : // only to set them.
413 5515 : let mut reopen_options = open_options.clone();
414 5515 : reopen_options.create(false);
415 5515 : reopen_options.create_new(false);
416 5515 : reopen_options.truncate(false);
417 5515 :
418 5515 : let vfile = VirtualFile {
419 5515 : handle: RwLock::new(handle),
420 5515 : pos: 0,
421 5515 : path: path_ref.to_path_buf(),
422 5515 : open_options: reopen_options,
423 5515 : tenant_id,
424 5515 : shard_id,
425 5515 : timeline_id,
426 5515 : };
427 5515 :
428 5515 : // TODO: Under pressure, it's likely the slot will get re-used and
429 5515 : // the underlying file closed before they get around to using it.
430 5515 : // => https://github.com/neondatabase/neon/issues/6065
431 5515 : slot_guard.file.replace(file);
432 5515 :
433 5515 : Ok(vfile)
434 5515 : }
435 :
436 : /// Async version of [`::utils::crashsafe::overwrite`].
437 : ///
438 : /// # NB:
439 : ///
440 : /// Doesn't actually use the [`VirtualFile`] file descriptor cache, but,
441 : /// it did at an earlier time.
442 : /// And it will use this module's [`io_engine`] in the near future, so, leaving it here.
443 28 : pub async fn crashsafe_overwrite<B: BoundedBuf<Buf = Buf> + Send, Buf: IoBuf + Send>(
444 28 : final_path: Utf8PathBuf,
445 28 : tmp_path: Utf8PathBuf,
446 28 : content: B,
447 28 : ) -> std::io::Result<()> {
448 28 : // TODO: use tokio_epoll_uring if configured as `io_engine`.
449 28 : // See https://github.com/neondatabase/neon/issues/6663
450 28 :
451 28 : tokio::task::spawn_blocking(move || {
452 28 : let slice_storage;
453 28 : let content_len = content.bytes_init();
454 28 : let content = if content.bytes_init() > 0 {
455 28 : slice_storage = Some(content.slice(0..content_len));
456 28 : slice_storage.as_deref().expect("just set it to Some()")
457 : } else {
458 0 : &[]
459 : };
460 28 : utils::crashsafe::overwrite(&final_path, &tmp_path, content)
461 28 : })
462 28 : .await
463 28 : .expect("blocking task is never aborted")
464 28 : }
465 :
466 : /// Call File::sync_all() on the underlying File.
467 2720 : pub async fn sync_all(&self) -> Result<(), Error> {
468 2720 : with_file!(self, StorageIoOperation::Fsync, |file_guard| {
469 2720 : let (_file_guard, res) = io_engine::get().sync_all(file_guard).await;
470 2720 : res
471 : })
472 2720 : }
473 :
474 : /// Call File::sync_data() on the underlying File.
475 0 : pub async fn sync_data(&self) -> Result<(), Error> {
476 0 : with_file!(self, StorageIoOperation::Fsync, |file_guard| {
477 0 : let (_file_guard, res) = io_engine::get().sync_data(file_guard).await;
478 0 : res
479 : })
480 0 : }
481 :
482 1561 : pub async fn metadata(&self) -> Result<Metadata, Error> {
483 1561 : with_file!(self, StorageIoOperation::Metadata, |file_guard| {
484 1561 : let (_file_guard, res) = io_engine::get().metadata(file_guard).await;
485 1561 : res
486 : })
487 1561 : }
488 :
489 : /// Helper function internal to `VirtualFile` that looks up the underlying File,
490 : /// opens it and evicts some other File if necessary. The passed parameter is
491 : /// assumed to be a function available for the physical `File`.
492 : ///
493 : /// We are doing it via a macro as Rust doesn't support async closures that
494 : /// take on parameters with lifetimes.
495 1489951 : async fn lock_file(&self) -> Result<FileGuard, Error> {
496 1489951 : let open_files = get_open_files();
497 :
498 183594 : let mut handle_guard = {
499 : // Read the cached slot handle, and see if the slot that it points to still
500 : // contains our File.
501 : //
502 : // We only need to hold the handle lock while we read the current handle. If
503 : // another thread closes the file and recycles the slot for a different file,
504 : // we will notice that the handle we read is no longer valid and retry.
505 1489951 : let mut handle = *self.handle.read().await;
506 1580347 : loop {
507 1580347 : // Check if the slot contains our File
508 1580347 : {
509 1580347 : let slot = &open_files.slots[handle.index];
510 1580347 : let slot_guard = slot.inner.read().await;
511 1580347 : if slot_guard.tag == handle.tag && slot_guard.file.is_some() {
512 : // Found a cached file descriptor.
513 1306357 : slot.recently_used.store(true, Ordering::Relaxed);
514 1306357 : return Ok(FileGuard { slot_guard });
515 273990 : }
516 : }
517 :
518 : // The slot didn't contain our File. We will have to open it ourselves,
519 : // but before that, grab a write lock on handle in the VirtualFile, so
520 : // that no other thread will try to concurrently open the same file.
521 273990 : let handle_guard = self.handle.write().await;
522 :
523 : // If another thread changed the handle while we were not holding the lock,
524 : // then the handle might now be valid again. Loop back to retry.
525 273990 : if *handle_guard != handle {
526 90396 : handle = *handle_guard;
527 90396 : continue;
528 183594 : }
529 183594 : break handle_guard;
530 : }
531 : };
532 :
533 : // We need to open the file ourselves. The handle in the VirtualFile is
534 : // now locked in write-mode. Find a free slot to put it in.
535 183594 : let (handle, mut slot_guard) = open_files.find_victim_slot().await;
536 :
537 : // Re-open the physical file.
538 : // NB: we use StorageIoOperation::OpenAferReplace for this to distinguish this
539 : // case from StorageIoOperation::Open. This helps with identifying thrashing
540 : // of the virtual file descriptor cache.
541 183594 : let file = observe_duration!(StorageIoOperation::OpenAfterReplace, {
542 183594 : self.open_options.open(self.path.as_std_path()).await?
543 : });
544 :
545 : // Store the File in the slot and update the handle in the VirtualFile
546 : // to point to it.
547 183594 : slot_guard.file.replace(file);
548 183594 :
549 183594 : *handle_guard = handle;
550 183594 :
551 183594 : return Ok(FileGuard {
552 183594 : slot_guard: slot_guard.downgrade(),
553 183594 : });
554 1489951 : }
555 :
556 6 : pub fn remove(self) {
557 6 : let path = self.path.clone();
558 6 : drop(self);
559 6 : std::fs::remove_file(path).expect("failed to remove the virtual file");
560 6 : }
561 :
562 4701 : pub async fn seek(&mut self, pos: SeekFrom) -> Result<u64, Error> {
563 4701 : match pos {
564 4691 : SeekFrom::Start(offset) => {
565 4691 : self.pos = offset;
566 4691 : }
567 4 : SeekFrom::End(offset) => {
568 4 : self.pos = with_file!(self, StorageIoOperation::Seek, |mut file_guard| file_guard
569 4 : .with_std_file_mut(|std_file| std_file.seek(SeekFrom::End(offset))))?
570 : }
571 6 : SeekFrom::Current(offset) => {
572 6 : let pos = self.pos as i128 + offset as i128;
573 6 : if pos < 0 {
574 2 : return Err(Error::new(
575 2 : ErrorKind::InvalidInput,
576 2 : "offset would be negative",
577 2 : ));
578 4 : }
579 4 : if pos > u64::MAX as i128 {
580 0 : return Err(Error::new(ErrorKind::InvalidInput, "offset overflow"));
581 4 : }
582 4 : self.pos = pos as u64;
583 : }
584 : }
585 4697 : Ok(self.pos)
586 4701 : }
587 :
588 334182 : pub async fn read_exact_at<B>(
589 334182 : &self,
590 334182 : buf: B,
591 334182 : offset: u64,
592 334182 : ctx: &RequestContext,
593 334182 : ) -> Result<B, Error>
594 334182 : where
595 334182 : B: IoBufMut + Send,
596 334182 : {
597 334182 : let (buf, res) = read_exact_at_impl(buf, offset, None, |buf, offset| {
598 334182 : self.read_at(buf, offset, ctx)
599 334182 : })
600 609714 : .await;
601 334182 : res.map(|()| buf)
602 334182 : }
603 :
604 38656 : pub async fn read_exact_at_n<B>(
605 38656 : &self,
606 38656 : buf: B,
607 38656 : offset: u64,
608 38656 : count: usize,
609 38656 : ctx: &RequestContext,
610 38656 : ) -> Result<B, Error>
611 38656 : where
612 38656 : B: IoBufMut + Send,
613 38656 : {
614 38656 : let (buf, res) = read_exact_at_impl(buf, offset, Some(count), |buf, offset| {
615 38656 : self.read_at(buf, offset, ctx)
616 38656 : })
617 19569 : .await;
618 38656 : res.map(|()| buf)
619 38656 : }
620 :
621 : /// Like [`Self::read_exact_at`] but for [`PageWriteGuard`].
622 113980 : pub async fn read_exact_at_page(
623 113980 : &self,
624 113980 : page: PageWriteGuard<'static>,
625 113980 : offset: u64,
626 113980 : ctx: &RequestContext,
627 113980 : ) -> Result<PageWriteGuard<'static>, Error> {
628 113980 : let buf = PageWriteGuardBuf {
629 113980 : page,
630 113980 : init_up_to: 0,
631 113980 : };
632 113980 : let res = self.read_exact_at(buf, offset, ctx).await;
633 113980 : res.map(|PageWriteGuardBuf { page, .. }| page)
634 113980 : .map_err(|e| Error::new(ErrorKind::Other, e))
635 113980 : }
636 :
637 : // Copied from https://doc.rust-lang.org/1.72.0/src/std/os/unix/fs.rs.html#219-235
638 4 : pub async fn write_all_at<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
639 4 : &self,
640 4 : buf: B,
641 4 : mut offset: u64,
642 4 : ctx: &RequestContext,
643 4 : ) -> (B::Buf, Result<(), Error>) {
644 4 : let buf_len = buf.bytes_init();
645 4 : if buf_len == 0 {
646 0 : return (Slice::into_inner(buf.slice_full()), Ok(()));
647 4 : }
648 4 : let mut buf = buf.slice(0..buf_len);
649 8 : while !buf.is_empty() {
650 : let res;
651 4 : (buf, res) = self.write_at(buf, offset, ctx).await;
652 0 : match res {
653 : Ok(0) => {
654 0 : return (
655 0 : Slice::into_inner(buf),
656 0 : Err(Error::new(
657 0 : std::io::ErrorKind::WriteZero,
658 0 : "failed to write whole buffer",
659 0 : )),
660 0 : );
661 : }
662 4 : Ok(n) => {
663 4 : buf = buf.slice(n..);
664 4 : offset += n as u64;
665 4 : }
666 0 : Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
667 0 : Err(e) => return (Slice::into_inner(buf), Err(e)),
668 : }
669 : }
670 4 : (Slice::into_inner(buf), Ok(()))
671 4 : }
672 :
673 : /// Writes `buf.slice(0..buf.bytes_init())`.
674 : /// Returns the IoBuf that is underlying the BoundedBuf `buf`.
675 : /// I.e., the returned value's `bytes_init()` method returns something different than the `bytes_init()` that was passed in.
676 : /// It's quite brittle and easy to mis-use, so, we return the size in the Ok() variant.
677 1112412 : pub async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
678 1112412 : &mut self,
679 1112412 : buf: B,
680 1112412 : ctx: &RequestContext,
681 1112412 : ) -> (B::Buf, Result<usize, Error>) {
682 1112412 : let nbytes = buf.bytes_init();
683 1112412 : if nbytes == 0 {
684 32 : return (Slice::into_inner(buf.slice_full()), Ok(0));
685 1112380 : }
686 1112380 : let mut buf = buf.slice(0..nbytes);
687 2224758 : while !buf.is_empty() {
688 : let res;
689 1112380 : (buf, res) = self.write(buf, ctx).await;
690 2 : match res {
691 : Ok(0) => {
692 0 : return (
693 0 : Slice::into_inner(buf),
694 0 : Err(Error::new(
695 0 : std::io::ErrorKind::WriteZero,
696 0 : "failed to write whole buffer",
697 0 : )),
698 0 : );
699 : }
700 1112378 : Ok(n) => {
701 1112378 : buf = buf.slice(n..);
702 1112378 : }
703 2 : Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {}
704 2 : Err(e) => return (Slice::into_inner(buf), Err(e)),
705 : }
706 : }
707 1112378 : (Slice::into_inner(buf), Ok(nbytes))
708 1112412 : }
709 :
710 1112380 : async fn write<B: IoBuf + Send>(
711 1112380 : &mut self,
712 1112380 : buf: Slice<B>,
713 1112380 : ctx: &RequestContext,
714 1112380 : ) -> (Slice<B>, Result<usize, std::io::Error>) {
715 1112380 : let pos = self.pos;
716 1112380 : let (buf, res) = self.write_at(buf, pos, ctx).await;
717 1112380 : let n = match res {
718 1112378 : Ok(n) => n,
719 2 : Err(e) => return (buf, Err(e)),
720 : };
721 1112378 : self.pos += n as u64;
722 1112378 : (buf, Ok(n))
723 1112380 : }
724 :
725 373282 : pub(crate) async fn read_at<B>(
726 373282 : &self,
727 373282 : buf: B,
728 373282 : offset: u64,
729 373282 : _ctx: &RequestContext, /* TODO: use for metrics: https://github.com/neondatabase/neon/issues/6107 */
730 373282 : ) -> (B, Result<usize, Error>)
731 373282 : where
732 373282 : B: tokio_epoll_uring::BoundedBufMut + Send,
733 373282 : {
734 442808 : let file_guard = match self.lock_file().await {
735 373282 : Ok(file_guard) => file_guard,
736 0 : Err(e) => return (buf, Err(e)),
737 : };
738 :
739 373282 : observe_duration!(StorageIoOperation::Read, {
740 373282 : let ((_file_guard, buf), res) = io_engine::get().read_at(file_guard, offset, buf).await;
741 373282 : if let Ok(size) = res {
742 373280 : STORAGE_IO_SIZE
743 373280 : .with_label_values(&[
744 373280 : "read",
745 373280 : &self.tenant_id,
746 373280 : &self.shard_id,
747 373280 : &self.timeline_id,
748 373280 : ])
749 373280 : .add(size as i64);
750 373280 : }
751 373282 : (buf, res)
752 : })
753 373282 : }
754 :
755 1112384 : async fn write_at<B: IoBuf + Send>(
756 1112384 : &self,
757 1112384 : buf: Slice<B>,
758 1112384 : offset: u64,
759 1112384 : _ctx: &RequestContext, /* TODO: use for metrics: https://github.com/neondatabase/neon/issues/6107 */
760 1112384 : ) -> (Slice<B>, Result<usize, Error>) {
761 1112384 : let file_guard = match self.lock_file().await {
762 1112384 : Ok(file_guard) => file_guard,
763 0 : Err(e) => return (buf, Err(e)),
764 : };
765 1112384 : observe_duration!(StorageIoOperation::Write, {
766 1112384 : let ((_file_guard, buf), result) =
767 1112384 : io_engine::get().write_at(file_guard, offset, buf).await;
768 1112384 : if let Ok(size) = result {
769 1112382 : STORAGE_IO_SIZE
770 1112382 : .with_label_values(&[
771 1112382 : "write",
772 1112382 : &self.tenant_id,
773 1112382 : &self.shard_id,
774 1112382 : &self.timeline_id,
775 1112382 : ])
776 1112382 : .add(size as i64);
777 1112382 : }
778 1112384 : (buf, result)
779 : })
780 1112384 : }
781 : }
782 :
783 : // Adapted from https://doc.rust-lang.org/1.72.0/src/std/os/unix/fs.rs.html#117-135
784 372848 : pub async fn read_exact_at_impl<B, F, Fut>(
785 372848 : buf: B,
786 372848 : mut offset: u64,
787 372848 : count: Option<usize>,
788 372848 : mut read_at: F,
789 372848 : ) -> (B, std::io::Result<()>)
790 372848 : where
791 372848 : B: IoBufMut + Send,
792 372848 : F: FnMut(tokio_epoll_uring::Slice<B>, u64) -> Fut,
793 372848 : Fut: std::future::Future<Output = (tokio_epoll_uring::Slice<B>, std::io::Result<usize>)>,
794 372848 : {
795 372848 : let mut buf: tokio_epoll_uring::Slice<B> = match count {
796 38658 : Some(count) => {
797 38658 : assert!(count <= buf.bytes_total());
798 38658 : assert!(count > 0);
799 38658 : buf.slice(..count) // may include uninitialized memory
800 : }
801 334190 : None => buf.slice_full(), // includes all the uninitialized memory
802 : };
803 :
804 745698 : while buf.bytes_total() != 0 {
805 : let res;
806 629283 : (buf, res) = read_at(buf, offset).await;
807 0 : match res {
808 2 : Ok(0) => break,
809 372850 : Ok(n) => {
810 372850 : buf = buf.slice(n..);
811 372850 : offset += n as u64;
812 372850 : }
813 0 : Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {}
814 0 : Err(e) => return (buf.into_inner(), Err(e)),
815 : }
816 : }
817 : // NB: don't use `buf.is_empty()` here; it is from the
818 : // `impl Deref for Slice { Target = [u8] }`; the &[u8]
819 : // returned by it only covers the initialized portion of `buf`.
820 : // Whereas we're interested in ensuring that we filled the entire
821 : // buffer that the user passed in.
822 372848 : if buf.bytes_total() != 0 {
823 2 : (
824 2 : buf.into_inner(),
825 2 : Err(std::io::Error::new(
826 2 : std::io::ErrorKind::UnexpectedEof,
827 2 : "failed to fill whole buffer",
828 2 : )),
829 2 : )
830 : } else {
831 372846 : assert_eq!(buf.len(), buf.bytes_total());
832 372846 : (buf.into_inner(), Ok(()))
833 : }
834 372848 : }
835 :
836 : #[cfg(test)]
837 : mod test_read_exact_at_impl {
838 :
839 : use std::{collections::VecDeque, sync::Arc};
840 :
841 : use tokio_epoll_uring::{BoundedBuf, BoundedBufMut};
842 :
843 : use super::read_exact_at_impl;
844 :
845 : struct Expectation {
846 : offset: u64,
847 : bytes_total: usize,
848 : result: std::io::Result<Vec<u8>>,
849 : }
850 : struct MockReadAt {
851 : expectations: VecDeque<Expectation>,
852 : }
853 :
854 : impl MockReadAt {
855 14 : async fn read_at(
856 14 : &mut self,
857 14 : mut buf: tokio_epoll_uring::Slice<Vec<u8>>,
858 14 : offset: u64,
859 14 : ) -> (tokio_epoll_uring::Slice<Vec<u8>>, std::io::Result<usize>) {
860 14 : let exp = self
861 14 : .expectations
862 14 : .pop_front()
863 14 : .expect("read_at called but we have no expectations left");
864 14 : assert_eq!(exp.offset, offset);
865 14 : assert_eq!(exp.bytes_total, buf.bytes_total());
866 14 : match exp.result {
867 14 : Ok(bytes) => {
868 14 : assert!(bytes.len() <= buf.bytes_total());
869 14 : buf.put_slice(&bytes);
870 14 : (buf, Ok(bytes.len()))
871 : }
872 0 : Err(e) => (buf, Err(e)),
873 : }
874 14 : }
875 : }
876 :
877 : impl Drop for MockReadAt {
878 10 : fn drop(&mut self) {
879 10 : assert_eq!(self.expectations.len(), 0);
880 10 : }
881 : }
882 :
883 : #[tokio::test]
884 2 : async fn test_basic() {
885 2 : let buf = Vec::with_capacity(5);
886 2 : let mock_read_at = Arc::new(tokio::sync::Mutex::new(MockReadAt {
887 2 : expectations: VecDeque::from(vec![Expectation {
888 2 : offset: 0,
889 2 : bytes_total: 5,
890 2 : result: Ok(vec![b'a', b'b', b'c', b'd', b'e']),
891 2 : }]),
892 2 : }));
893 2 : let (buf, res) = read_exact_at_impl(buf, 0, None, |buf, offset| {
894 2 : let mock_read_at = Arc::clone(&mock_read_at);
895 2 : async move { mock_read_at.lock().await.read_at(buf, offset).await }
896 2 : })
897 2 : .await;
898 2 : assert!(res.is_ok());
899 2 : assert_eq!(buf, vec![b'a', b'b', b'c', b'd', b'e']);
900 2 : }
901 :
902 : #[tokio::test]
903 2 : async fn test_with_count() {
904 2 : let buf = Vec::with_capacity(5);
905 2 : let mock_read_at = Arc::new(tokio::sync::Mutex::new(MockReadAt {
906 2 : expectations: VecDeque::from(vec![Expectation {
907 2 : offset: 0,
908 2 : bytes_total: 3,
909 2 : result: Ok(vec![b'a', b'b', b'c']),
910 2 : }]),
911 2 : }));
912 2 :
913 2 : let (buf, res) = read_exact_at_impl(buf, 0, Some(3), |buf, offset| {
914 2 : let mock_read_at = Arc::clone(&mock_read_at);
915 2 : async move { mock_read_at.lock().await.read_at(buf, offset).await }
916 2 : })
917 2 : .await;
918 2 : assert!(res.is_ok());
919 2 : assert_eq!(buf, vec![b'a', b'b', b'c']);
920 2 : }
921 :
922 : #[tokio::test]
923 2 : async fn test_empty_buf_issues_no_syscall() {
924 2 : let buf = Vec::new();
925 2 : let mock_read_at = Arc::new(tokio::sync::Mutex::new(MockReadAt {
926 2 : expectations: VecDeque::new(),
927 2 : }));
928 2 : let (_buf, res) = read_exact_at_impl(buf, 0, None, |buf, offset| {
929 0 : let mock_read_at = Arc::clone(&mock_read_at);
930 2 : async move { mock_read_at.lock().await.read_at(buf, offset).await }
931 2 : })
932 2 : .await;
933 2 : assert!(res.is_ok());
934 2 : }
935 :
936 : #[tokio::test]
937 2 : async fn test_two_read_at_calls_needed_until_buf_filled() {
938 2 : let buf = Vec::with_capacity(4);
939 2 : let mock_read_at = Arc::new(tokio::sync::Mutex::new(MockReadAt {
940 2 : expectations: VecDeque::from(vec![
941 2 : Expectation {
942 2 : offset: 0,
943 2 : bytes_total: 4,
944 2 : result: Ok(vec![b'a', b'b']),
945 2 : },
946 2 : Expectation {
947 2 : offset: 2,
948 2 : bytes_total: 2,
949 2 : result: Ok(vec![b'c', b'd']),
950 2 : },
951 2 : ]),
952 2 : }));
953 4 : let (buf, res) = read_exact_at_impl(buf, 0, None, |buf, offset| {
954 4 : let mock_read_at = Arc::clone(&mock_read_at);
955 4 : async move { mock_read_at.lock().await.read_at(buf, offset).await }
956 4 : })
957 2 : .await;
958 2 : assert!(res.is_ok());
959 2 : assert_eq!(buf, vec![b'a', b'b', b'c', b'd']);
960 2 : }
961 :
962 : #[tokio::test]
963 2 : async fn test_eof_before_buffer_full() {
964 2 : let buf = Vec::with_capacity(3);
965 2 : let mock_read_at = Arc::new(tokio::sync::Mutex::new(MockReadAt {
966 2 : expectations: VecDeque::from(vec![
967 2 : Expectation {
968 2 : offset: 0,
969 2 : bytes_total: 3,
970 2 : result: Ok(vec![b'a']),
971 2 : },
972 2 : Expectation {
973 2 : offset: 1,
974 2 : bytes_total: 2,
975 2 : result: Ok(vec![b'b']),
976 2 : },
977 2 : Expectation {
978 2 : offset: 2,
979 2 : bytes_total: 1,
980 2 : result: Ok(vec![]),
981 2 : },
982 2 : ]),
983 2 : }));
984 6 : let (_buf, res) = read_exact_at_impl(buf, 0, None, |buf, offset| {
985 6 : let mock_read_at = Arc::clone(&mock_read_at);
986 6 : async move { mock_read_at.lock().await.read_at(buf, offset).await }
987 6 : })
988 2 : .await;
989 2 : let Err(err) = res else {
990 2 : panic!("should return an error");
991 2 : };
992 2 : assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
993 2 : assert_eq!(format!("{err}"), "failed to fill whole buffer");
994 2 : // buffer contents on error are unspecified
995 2 : }
996 : }
997 :
998 : struct FileGuard {
999 : slot_guard: RwLockReadGuard<'static, SlotInner>,
1000 : }
1001 :
1002 : impl AsRef<OwnedFd> for FileGuard {
1003 1489951 : fn as_ref(&self) -> &OwnedFd {
1004 1489951 : // This unwrap is safe because we only create `FileGuard`s
1005 1489951 : // if we know that the file is Some.
1006 1489951 : self.slot_guard.file.as_ref().unwrap()
1007 1489951 : }
1008 : }
1009 :
1010 : impl FileGuard {
1011 : /// Soft deprecation: we'll move VirtualFile to async APIs and remove this function eventually.
1012 744967 : fn with_std_file<F, R>(&self, with: F) -> R
1013 744967 : where
1014 744967 : F: FnOnce(&File) -> R,
1015 744967 : {
1016 744967 : // SAFETY:
1017 744967 : // - lifetime of the fd: `file` doesn't outlive the OwnedFd stored in `self`.
1018 744967 : // - `&` usage below: `self` is `&`, hence Rust typesystem guarantees there are is no `&mut`
1019 744967 : let file = unsafe { File::from_raw_fd(self.as_ref().as_raw_fd()) };
1020 744967 : let res = with(&file);
1021 744967 : let _ = file.into_raw_fd();
1022 744967 : res
1023 744967 : }
1024 : /// Soft deprecation: we'll move VirtualFile to async APIs and remove this function eventually.
1025 4 : fn with_std_file_mut<F, R>(&mut self, with: F) -> R
1026 4 : where
1027 4 : F: FnOnce(&mut File) -> R,
1028 4 : {
1029 4 : // SAFETY:
1030 4 : // - lifetime of the fd: `file` doesn't outlive the OwnedFd stored in `self`.
1031 4 : // - &mut usage below: `self` is `&mut`, hence this call is the only task/thread that has control over the underlying fd
1032 4 : let mut file = unsafe { File::from_raw_fd(self.as_ref().as_raw_fd()) };
1033 4 : let res = with(&mut file);
1034 4 : let _ = file.into_raw_fd();
1035 4 : res
1036 4 : }
1037 : }
1038 :
1039 : impl tokio_epoll_uring::IoFd for FileGuard {
1040 744980 : unsafe fn as_fd(&self) -> RawFd {
1041 744980 : let owned_fd: &OwnedFd = self.as_ref();
1042 744980 : owned_fd.as_raw_fd()
1043 744980 : }
1044 : }
1045 :
1046 : #[cfg(test)]
1047 : impl VirtualFile {
1048 20200 : pub(crate) async fn read_blk(
1049 20200 : &self,
1050 20200 : blknum: u32,
1051 20200 : ctx: &RequestContext,
1052 20200 : ) -> Result<crate::tenant::block_io::BlockLease<'_>, std::io::Error> {
1053 20200 : use crate::page_cache::PAGE_SZ;
1054 20200 : let buf = vec![0; PAGE_SZ];
1055 20200 : let buf = self
1056 20200 : .read_exact_at(buf, blknum as u64 * (PAGE_SZ as u64), ctx)
1057 10255 : .await?;
1058 20200 : Ok(crate::tenant::block_io::BlockLease::Vec(buf))
1059 20200 : }
1060 :
1061 224 : async fn read_to_end(&mut self, buf: &mut Vec<u8>, ctx: &RequestContext) -> Result<(), Error> {
1062 224 : let mut tmp = vec![0; 128];
1063 : loop {
1064 : let res;
1065 444 : (tmp, res) = self.read_at(tmp, self.pos, ctx).await;
1066 2 : match res {
1067 222 : Ok(0) => return Ok(()),
1068 220 : Ok(n) => {
1069 220 : self.pos += n as u64;
1070 220 : buf.extend_from_slice(&tmp[..n]);
1071 220 : }
1072 2 : Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {}
1073 2 : Err(e) => return Err(e),
1074 : }
1075 : }
1076 224 : }
1077 : }
1078 :
1079 : impl Drop for VirtualFile {
1080 : /// If a VirtualFile is dropped, close the underlying file if it was open.
1081 4742 : fn drop(&mut self) {
1082 4742 : let handle = self.handle.get_mut();
1083 4742 :
1084 4742 : fn clean_slot(slot: &Slot, mut slot_guard: RwLockWriteGuard<'_, SlotInner>, tag: u64) {
1085 4742 : if slot_guard.tag == tag {
1086 4742 : slot.recently_used.store(false, Ordering::Relaxed);
1087 4742 : // there is also operation "close-by-replace" for closes done on eviction for
1088 4742 : // comparison.
1089 4742 : if let Some(fd) = slot_guard.file.take() {
1090 4157 : STORAGE_IO_TIME_METRIC
1091 4157 : .get(StorageIoOperation::Close)
1092 4157 : .observe_closure_duration(|| drop(fd));
1093 4157 : }
1094 4742 : }
1095 4742 : }
1096 4742 :
1097 4742 : // We don't have async drop so we cannot directly await the lock here.
1098 4742 : // Instead, first do a best-effort attempt at closing the underlying
1099 4742 : // file descriptor by using `try_write`, and if that fails, spawn
1100 4742 : // a tokio task to do it asynchronously: we just want it to be
1101 4742 : // cleaned up eventually.
1102 4742 : // Most of the time, the `try_lock` should succeed though,
1103 4742 : // as we have `&mut self` access. In other words, if the slot
1104 4742 : // is still occupied by our file, there should be no access from
1105 4742 : // other I/O operations; the only other possible place to lock
1106 4742 : // the slot is the lock algorithm looking for free slots.
1107 4742 : let slot = &get_open_files().slots[handle.index];
1108 4742 : if let Ok(slot_guard) = slot.inner.try_write() {
1109 4742 : clean_slot(slot, slot_guard, handle.tag);
1110 4742 : } else {
1111 0 : let tag = handle.tag;
1112 0 : tokio::spawn(async move {
1113 0 : let slot_guard = slot.inner.write().await;
1114 0 : clean_slot(slot, slot_guard, tag);
1115 0 : });
1116 0 : };
1117 4742 : }
1118 : }
1119 :
1120 : impl OwnedAsyncWriter for VirtualFile {
1121 : #[inline(always)]
1122 3 : async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
1123 3 : &mut self,
1124 3 : buf: B,
1125 3 : ctx: &RequestContext,
1126 3 : ) -> std::io::Result<(usize, B::Buf)> {
1127 3 : let (buf, res) = VirtualFile::write_all(self, buf, ctx).await;
1128 3 : res.map(move |v| (v, buf))
1129 3 : }
1130 : }
1131 :
1132 : impl OpenFiles {
1133 159 : fn new(num_slots: usize) -> OpenFiles {
1134 159 : let mut slots = Box::new(Vec::with_capacity(num_slots));
1135 1590 : for _ in 0..num_slots {
1136 1590 : let slot = Slot {
1137 1590 : recently_used: AtomicBool::new(false),
1138 1590 : inner: RwLock::new(SlotInner { tag: 0, file: None }),
1139 1590 : };
1140 1590 : slots.push(slot);
1141 1590 : }
1142 :
1143 159 : OpenFiles {
1144 159 : next: AtomicUsize::new(0),
1145 159 : slots: Box::leak(slots),
1146 159 : }
1147 159 : }
1148 : }
1149 :
1150 : ///
1151 : /// Initialize the virtual file module. This must be called once at page
1152 : /// server startup.
1153 : ///
1154 : #[cfg(not(test))]
1155 0 : pub fn init(num_slots: usize, engine: IoEngineKind) {
1156 0 : if OPEN_FILES.set(OpenFiles::new(num_slots)).is_err() {
1157 0 : panic!("virtual_file::init called twice");
1158 0 : }
1159 0 : io_engine::init(engine);
1160 0 : crate::metrics::virtual_file_descriptor_cache::SIZE_MAX.set(num_slots as u64);
1161 0 : }
1162 :
1163 : const TEST_MAX_FILE_DESCRIPTORS: usize = 10;
1164 :
1165 : // Get a handle to the global slots array.
1166 1500208 : fn get_open_files() -> &'static OpenFiles {
1167 1500208 : //
1168 1500208 : // In unit tests, page server startup doesn't happen and no one calls
1169 1500208 : // virtual_file::init(). Initialize it here, with a small array.
1170 1500208 : //
1171 1500208 : // This applies to the virtual file tests below, but all other unit
1172 1500208 : // tests too, so the virtual file facility is always usable in
1173 1500208 : // unit tests.
1174 1500208 : //
1175 1500208 : if cfg!(test) {
1176 1500208 : OPEN_FILES.get_or_init(|| OpenFiles::new(TEST_MAX_FILE_DESCRIPTORS))
1177 : } else {
1178 0 : OPEN_FILES.get().expect("virtual_file::init not called yet")
1179 : }
1180 1500208 : }
1181 :
1182 : #[cfg(test)]
1183 : mod tests {
1184 : use crate::context::DownloadBehavior;
1185 : use crate::task_mgr::TaskKind;
1186 :
1187 : use super::*;
1188 : use rand::seq::SliceRandom;
1189 : use rand::thread_rng;
1190 : use rand::Rng;
1191 : use std::io::Write;
1192 : use std::os::unix::fs::FileExt;
1193 : use std::sync::Arc;
1194 :
1195 : enum MaybeVirtualFile {
1196 : VirtualFile(VirtualFile),
1197 : File(File),
1198 : }
1199 :
1200 : impl From<VirtualFile> for MaybeVirtualFile {
1201 6 : fn from(vf: VirtualFile) -> Self {
1202 6 : MaybeVirtualFile::VirtualFile(vf)
1203 6 : }
1204 : }
1205 :
1206 : impl MaybeVirtualFile {
1207 404 : async fn read_exact_at(
1208 404 : &self,
1209 404 : mut buf: Vec<u8>,
1210 404 : offset: u64,
1211 404 : ctx: &RequestContext,
1212 404 : ) -> Result<Vec<u8>, Error> {
1213 404 : match self {
1214 203 : MaybeVirtualFile::VirtualFile(file) => file.read_exact_at(buf, offset, ctx).await,
1215 202 : MaybeVirtualFile::File(file) => file.read_exact_at(&mut buf, offset).map(|()| buf),
1216 : }
1217 404 : }
1218 8 : async fn write_all_at<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
1219 8 : &self,
1220 8 : buf: B,
1221 8 : offset: u64,
1222 8 : ctx: &RequestContext,
1223 8 : ) -> Result<(), Error> {
1224 8 : match self {
1225 4 : MaybeVirtualFile::VirtualFile(file) => {
1226 4 : let (_buf, res) = file.write_all_at(buf, offset, ctx).await;
1227 4 : res
1228 : }
1229 4 : MaybeVirtualFile::File(file) => {
1230 4 : let buf_len = buf.bytes_init();
1231 4 : if buf_len == 0 {
1232 0 : return Ok(());
1233 4 : }
1234 4 : file.write_all_at(&buf.slice(0..buf_len), offset)
1235 : }
1236 : }
1237 8 : }
1238 36 : async fn seek(&mut self, pos: SeekFrom) -> Result<u64, Error> {
1239 36 : match self {
1240 18 : MaybeVirtualFile::VirtualFile(file) => file.seek(pos).await,
1241 18 : MaybeVirtualFile::File(file) => file.seek(pos),
1242 : }
1243 36 : }
1244 8 : async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
1245 8 : &mut self,
1246 8 : buf: B,
1247 8 : ctx: &RequestContext,
1248 8 : ) -> Result<(), Error> {
1249 8 : match self {
1250 4 : MaybeVirtualFile::VirtualFile(file) => {
1251 4 : let (_buf, res) = file.write_all(buf, ctx).await;
1252 4 : res.map(|_| ())
1253 : }
1254 4 : MaybeVirtualFile::File(file) => {
1255 4 : let buf_len = buf.bytes_init();
1256 4 : if buf_len == 0 {
1257 0 : return Ok(());
1258 4 : }
1259 4 : file.write_all(&buf.slice(0..buf_len))
1260 : }
1261 : }
1262 8 : }
1263 :
1264 : // Helper function to slurp contents of a file, starting at the current position,
1265 : // into a string
1266 442 : async fn read_string(&mut self, ctx: &RequestContext) -> Result<String, Error> {
1267 442 : use std::io::Read;
1268 442 : let mut buf = String::new();
1269 442 : match self {
1270 224 : MaybeVirtualFile::VirtualFile(file) => {
1271 224 : let mut buf = Vec::new();
1272 226 : file.read_to_end(&mut buf, ctx).await?;
1273 222 : return Ok(String::from_utf8(buf).unwrap());
1274 : }
1275 218 : MaybeVirtualFile::File(file) => {
1276 218 : file.read_to_string(&mut buf)?;
1277 : }
1278 : }
1279 216 : Ok(buf)
1280 442 : }
1281 :
1282 : // Helper function to slurp a portion of a file into a string
1283 404 : async fn read_string_at(
1284 404 : &mut self,
1285 404 : pos: u64,
1286 404 : len: usize,
1287 404 : ctx: &RequestContext,
1288 404 : ) -> Result<String, Error> {
1289 404 : let buf = vec![0; len];
1290 404 : let buf = self.read_exact_at(buf, pos, ctx).await?;
1291 404 : Ok(String::from_utf8(buf).unwrap())
1292 404 : }
1293 : }
1294 :
1295 : #[tokio::test]
1296 2 : async fn test_virtual_files() -> anyhow::Result<()> {
1297 2 : // The real work is done in the test_files() helper function. This
1298 2 : // allows us to run the same set of tests against a native File, and
1299 2 : // VirtualFile. We trust the native Files and wouldn't need to test them,
1300 2 : // but this allows us to verify that the operations return the same
1301 2 : // results with VirtualFiles as with native Files. (Except that with
1302 2 : // native files, you will run out of file descriptors if the ulimit
1303 2 : // is low enough.)
1304 2 : struct A;
1305 2 :
1306 2 : impl Adapter for A {
1307 206 : async fn open(
1308 206 : path: Utf8PathBuf,
1309 206 : opts: OpenOptions,
1310 206 : ctx: &RequestContext,
1311 206 : ) -> Result<MaybeVirtualFile, anyhow::Error> {
1312 206 : let vf = VirtualFile::open_with_options(&path, &opts, ctx).await?;
1313 206 : Ok(MaybeVirtualFile::VirtualFile(vf))
1314 206 : }
1315 2 : }
1316 531 : test_files::<A>("virtual_files").await
1317 2 : }
1318 :
1319 : #[tokio::test]
1320 2 : async fn test_physical_files() -> anyhow::Result<()> {
1321 2 : struct B;
1322 2 :
1323 2 : impl Adapter for B {
1324 206 : async fn open(
1325 206 : path: Utf8PathBuf,
1326 206 : opts: OpenOptions,
1327 206 : _ctx: &RequestContext,
1328 206 : ) -> Result<MaybeVirtualFile, anyhow::Error> {
1329 206 : Ok(MaybeVirtualFile::File({
1330 206 : let owned_fd = opts.open(path.as_std_path()).await?;
1331 206 : File::from(owned_fd)
1332 2 : }))
1333 206 : }
1334 2 : }
1335 2 :
1336 104 : test_files::<B>("physical_files").await
1337 2 : }
1338 :
1339 : /// This is essentially a closure which returns a MaybeVirtualFile, but because rust edition
1340 : /// 2024 is not yet out with new lifetime capture or outlives rules, this is a async function
1341 : /// in trait which benefits from the new lifetime capture rules already.
1342 : trait Adapter {
1343 : async fn open(
1344 : path: Utf8PathBuf,
1345 : opts: OpenOptions,
1346 : ctx: &RequestContext,
1347 : ) -> Result<MaybeVirtualFile, anyhow::Error>;
1348 : }
1349 :
1350 4 : async fn test_files<A>(testname: &str) -> anyhow::Result<()>
1351 4 : where
1352 4 : A: Adapter,
1353 4 : {
1354 4 : let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error);
1355 4 : let testdir = crate::config::PageServerConf::test_repo_dir(testname);
1356 4 : std::fs::create_dir_all(&testdir)?;
1357 :
1358 4 : let path_a = testdir.join("file_a");
1359 4 : let mut file_a = A::open(
1360 4 : path_a.clone(),
1361 4 : OpenOptions::new()
1362 4 : .write(true)
1363 4 : .create(true)
1364 4 : .truncate(true)
1365 4 : .to_owned(),
1366 4 : &ctx,
1367 4 : )
1368 4 : .await?;
1369 4 : file_a.write_all(b"foobar".to_vec(), &ctx).await?;
1370 :
1371 : // cannot read from a file opened in write-only mode
1372 4 : let _ = file_a.read_string(&ctx).await.unwrap_err();
1373 :
1374 : // Close the file and re-open for reading
1375 4 : let mut file_a = A::open(path_a, OpenOptions::new().read(true).to_owned(), &ctx).await?;
1376 :
1377 : // cannot write to a file opened in read-only mode
1378 4 : let _ = file_a.write_all(b"bar".to_vec(), &ctx).await.unwrap_err();
1379 4 :
1380 4 : // Try simple read
1381 4 : assert_eq!("foobar", file_a.read_string(&ctx).await?);
1382 :
1383 : // It's positioned at the EOF now.
1384 4 : assert_eq!("", file_a.read_string(&ctx).await?);
1385 :
1386 : // Test seeks.
1387 4 : assert_eq!(file_a.seek(SeekFrom::Start(1)).await?, 1);
1388 4 : assert_eq!("oobar", file_a.read_string(&ctx).await?);
1389 :
1390 4 : assert_eq!(file_a.seek(SeekFrom::End(-2)).await?, 4);
1391 4 : assert_eq!("ar", file_a.read_string(&ctx).await?);
1392 :
1393 4 : assert_eq!(file_a.seek(SeekFrom::Start(1)).await?, 1);
1394 4 : assert_eq!(file_a.seek(SeekFrom::Current(2)).await?, 3);
1395 4 : assert_eq!("bar", file_a.read_string(&ctx).await?);
1396 :
1397 4 : assert_eq!(file_a.seek(SeekFrom::Current(-5)).await?, 1);
1398 4 : assert_eq!("oobar", file_a.read_string(&ctx).await?);
1399 :
1400 : // Test erroneous seeks to before byte 0
1401 4 : file_a.seek(SeekFrom::End(-7)).await.unwrap_err();
1402 4 : assert_eq!(file_a.seek(SeekFrom::Start(1)).await?, 1);
1403 4 : file_a.seek(SeekFrom::Current(-2)).await.unwrap_err();
1404 4 :
1405 4 : // the erroneous seek should have left the position unchanged
1406 4 : assert_eq!("oobar", file_a.read_string(&ctx).await?);
1407 :
1408 : // Create another test file, and try FileExt functions on it.
1409 4 : let path_b = testdir.join("file_b");
1410 4 : let mut file_b = A::open(
1411 4 : path_b.clone(),
1412 4 : OpenOptions::new()
1413 4 : .read(true)
1414 4 : .write(true)
1415 4 : .create(true)
1416 4 : .truncate(true)
1417 4 : .to_owned(),
1418 4 : &ctx,
1419 4 : )
1420 2 : .await?;
1421 4 : file_b.write_all_at(b"BAR".to_vec(), 3, &ctx).await?;
1422 4 : file_b.write_all_at(b"FOO".to_vec(), 0, &ctx).await?;
1423 :
1424 4 : assert_eq!(file_b.read_string_at(2, 3, &ctx).await?, "OBA");
1425 :
1426 : // Open a lot of files, enough to cause some evictions. (Or to be precise,
1427 : // open the same file many times. The effect is the same.)
1428 : //
1429 : // leave file_a positioned at offset 1 before we start
1430 4 : assert_eq!(file_a.seek(SeekFrom::Start(1)).await?, 1);
1431 :
1432 4 : let mut vfiles = Vec::new();
1433 404 : for _ in 0..100 {
1434 400 : let mut vfile = A::open(
1435 400 : path_b.clone(),
1436 400 : OpenOptions::new().read(true).to_owned(),
1437 400 : &ctx,
1438 400 : )
1439 200 : .await?;
1440 400 : assert_eq!("FOOBAR", vfile.read_string(&ctx).await?);
1441 400 : vfiles.push(vfile);
1442 : }
1443 :
1444 : // make sure we opened enough files to definitely cause evictions.
1445 4 : assert!(vfiles.len() > TEST_MAX_FILE_DESCRIPTORS * 2);
1446 :
1447 : // The underlying file descriptor for 'file_a' should be closed now. Try to read
1448 : // from it again. We left the file positioned at offset 1 above.
1449 4 : assert_eq!("oobar", file_a.read_string(&ctx).await?);
1450 :
1451 : // Check that all the other FDs still work too. Use them in random order for
1452 : // good measure.
1453 4 : vfiles.as_mut_slice().shuffle(&mut thread_rng());
1454 400 : for vfile in vfiles.iter_mut() {
1455 400 : assert_eq!("OOBAR", vfile.read_string_at(1, 5, &ctx).await?);
1456 : }
1457 :
1458 4 : Ok(())
1459 4 : }
1460 :
1461 : /// Test using VirtualFiles from many threads concurrently. This tests both using
1462 : /// a lot of VirtualFiles concurrently, causing evictions, and also using the same
1463 : /// VirtualFile from multiple threads concurrently.
1464 : #[tokio::test]
1465 2 : async fn test_vfile_concurrency() -> Result<(), Error> {
1466 2 : const SIZE: usize = 8 * 1024;
1467 2 : const VIRTUAL_FILES: usize = 100;
1468 2 : const THREADS: usize = 100;
1469 2 : const SAMPLE: [u8; SIZE] = [0xADu8; SIZE];
1470 2 :
1471 2 : let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error);
1472 2 : let testdir = crate::config::PageServerConf::test_repo_dir("vfile_concurrency");
1473 2 : std::fs::create_dir_all(&testdir)?;
1474 2 :
1475 2 : // Create a test file.
1476 2 : let test_file_path = testdir.join("concurrency_test_file");
1477 2 : {
1478 2 : let file = File::create(&test_file_path)?;
1479 2 : file.write_all_at(&SAMPLE, 0)?;
1480 2 : }
1481 2 :
1482 2 : // Open the file many times.
1483 2 : let mut files = Vec::new();
1484 202 : for _ in 0..VIRTUAL_FILES {
1485 200 : let f = VirtualFile::open_with_options(
1486 200 : &test_file_path,
1487 200 : OpenOptions::new().read(true),
1488 200 : &ctx,
1489 200 : )
1490 101 : .await?;
1491 200 : files.push(f);
1492 2 : }
1493 2 : let files = Arc::new(files);
1494 2 :
1495 2 : // Launch many threads, and use the virtual files concurrently in random order.
1496 2 : let rt = tokio::runtime::Builder::new_multi_thread()
1497 2 : .worker_threads(THREADS)
1498 2 : .thread_name("test_vfile_concurrency thread")
1499 2 : .build()
1500 2 : .unwrap();
1501 2 : let mut hdls = Vec::new();
1502 202 : for _threadno in 0..THREADS {
1503 200 : let files = files.clone();
1504 200 : let ctx = ctx.detached_child(TaskKind::UnitTest, DownloadBehavior::Error);
1505 200 : let hdl = rt.spawn(async move {
1506 200 : let mut buf = vec![0u8; SIZE];
1507 200 : let mut rng = rand::rngs::OsRng;
1508 200000 : for _ in 1..1000 {
1509 199800 : let f = &files[rng.gen_range(0..files.len())];
1510 534633 : buf = f.read_exact_at(buf, 0, &ctx).await.unwrap();
1511 199800 : assert!(buf == SAMPLE);
1512 2 : }
1513 200 : });
1514 200 : hdls.push(hdl);
1515 200 : }
1516 202 : for hdl in hdls {
1517 200 : hdl.await?;
1518 2 : }
1519 2 : std::mem::forget(rt);
1520 2 :
1521 2 : Ok(())
1522 2 : }
1523 :
1524 : #[tokio::test]
1525 2 : async fn test_atomic_overwrite_basic() {
1526 2 : let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error);
1527 2 : let testdir = crate::config::PageServerConf::test_repo_dir("test_atomic_overwrite_basic");
1528 2 : std::fs::create_dir_all(&testdir).unwrap();
1529 2 :
1530 2 : let path = testdir.join("myfile");
1531 2 : let tmp_path = testdir.join("myfile.tmp");
1532 2 :
1533 2 : VirtualFile::crashsafe_overwrite(path.clone(), tmp_path.clone(), b"foo".to_vec())
1534 2 : .await
1535 2 : .unwrap();
1536 2 : let mut file = MaybeVirtualFile::from(VirtualFile::open(&path, &ctx).await.unwrap());
1537 2 : let post = file.read_string(&ctx).await.unwrap();
1538 2 : assert_eq!(post, "foo");
1539 2 : assert!(!tmp_path.exists());
1540 2 : drop(file);
1541 2 :
1542 2 : VirtualFile::crashsafe_overwrite(path.clone(), tmp_path.clone(), b"bar".to_vec())
1543 2 : .await
1544 2 : .unwrap();
1545 2 : let mut file = MaybeVirtualFile::from(VirtualFile::open(&path, &ctx).await.unwrap());
1546 2 : let post = file.read_string(&ctx).await.unwrap();
1547 2 : assert_eq!(post, "bar");
1548 2 : assert!(!tmp_path.exists());
1549 2 : drop(file);
1550 2 : }
1551 :
1552 : #[tokio::test]
1553 2 : async fn test_atomic_overwrite_preexisting_tmp() {
1554 2 : let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error);
1555 2 : let testdir =
1556 2 : crate::config::PageServerConf::test_repo_dir("test_atomic_overwrite_preexisting_tmp");
1557 2 : std::fs::create_dir_all(&testdir).unwrap();
1558 2 :
1559 2 : let path = testdir.join("myfile");
1560 2 : let tmp_path = testdir.join("myfile.tmp");
1561 2 :
1562 2 : std::fs::write(&tmp_path, "some preexisting junk that should be removed").unwrap();
1563 2 : assert!(tmp_path.exists());
1564 2 :
1565 2 : VirtualFile::crashsafe_overwrite(path.clone(), tmp_path.clone(), b"foo".to_vec())
1566 2 : .await
1567 2 : .unwrap();
1568 2 :
1569 2 : let mut file = MaybeVirtualFile::from(VirtualFile::open(&path, &ctx).await.unwrap());
1570 2 : let post = file.read_string(&ctx).await.unwrap();
1571 2 : assert_eq!(post, "foo");
1572 2 : assert!(!tmp_path.exists());
1573 2 : drop(file);
1574 2 : }
1575 : }
|
