TLA 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::metrics::{StorageIoOperation, STORAGE_IO_SIZE, STORAGE_IO_TIME_METRIC};
14 : use crate::tenant::TENANTS_SEGMENT_NAME;
15 : use camino::{Utf8Path, Utf8PathBuf};
16 : use once_cell::sync::OnceCell;
17 : use std::fs::{self, File, OpenOptions};
18 : use std::io::{Error, ErrorKind, Seek, SeekFrom};
19 : use std::os::unix::fs::FileExt;
20 : use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
21 : use std::sync::{RwLock, RwLockWriteGuard};
22 : use utils::fs_ext;
23 :
24 : ///
25 : /// A virtual file descriptor. You can use this just like std::fs::File, but internally
26 : /// the underlying file is closed if the system is low on file descriptors,
27 : /// and re-opened when it's accessed again.
28 : ///
29 : /// Like with std::fs::File, multiple threads can read/write the file concurrently,
30 : /// holding just a shared reference the same VirtualFile, using the read_at() / write_at()
31 : /// functions from the FileExt trait. But the functions from the Read/Write/Seek traits
32 : /// require a mutable reference, because they modify the "current position".
33 : ///
34 : /// Each VirtualFile has a physical file descriptor in the global OPEN_FILES array, at the
35 : /// slot that 'handle points to, if the underlying file is currently open. If it's not
36 : /// currently open, the 'handle' can still point to the slot where it was last kept. The
37 : /// 'tag' field is used to detect whether the handle still is valid or not.
38 : ///
39 UBC 0 : #[derive(Debug)]
40 : pub struct VirtualFile {
41 : /// Lazy handle to the global file descriptor cache. The slot that this points to
42 : /// might contain our File, or it may be empty, or it may contain a File that
43 : /// belongs to a different VirtualFile.
44 : handle: RwLock<SlotHandle>,
45 :
46 : /// Current file position
47 : pos: u64,
48 :
49 : /// File path and options to use to open it.
50 : ///
51 : /// Note: this only contains the options needed to re-open it. For example,
52 : /// if a new file is created, we only pass the create flag when it's initially
53 : /// opened, in the VirtualFile::create() function, and strip the flag before
54 : /// storing it here.
55 : pub path: Utf8PathBuf,
56 : open_options: OpenOptions,
57 :
58 : // These are strings becase we only use them for metrics, and those expect strings.
59 : // It makes no sense for us to constantly turn the `TimelineId` and `TenantId` into
60 : // strings.
61 : tenant_id: String,
62 : timeline_id: String,
63 : }
64 :
65 CBC 125783 : #[derive(Debug, PartialEq, Clone, Copy)]
66 : struct SlotHandle {
67 : /// Index into OPEN_FILES.slots
68 : index: usize,
69 :
70 : /// Value of 'tag' in the slot. If slot's tag doesn't match, then the slot has
71 : /// been recycled and no longer contains the FD for this virtual file.
72 : tag: u64,
73 : }
74 :
75 : /// OPEN_FILES is the global array that holds the physical file descriptors that
76 : /// are currently open. Each slot in the array is protected by a separate lock,
77 : /// so that different files can be accessed independently. The lock must be held
78 : /// in write mode to replace the slot with a different file, but a read mode
79 : /// is enough to operate on the file, whether you're reading or writing to it.
80 : ///
81 : /// OPEN_FILES starts in uninitialized state, and it's initialized by
82 : /// the virtual_file::init() function. It must be called exactly once at page
83 : /// server startup.
84 : static OPEN_FILES: OnceCell<OpenFiles> = OnceCell::new();
85 :
86 : struct OpenFiles {
87 : slots: &'static [Slot],
88 :
89 : /// clock arm for the clock algorithm
90 : next: AtomicUsize,
91 : }
92 :
93 : struct Slot {
94 : inner: RwLock<SlotInner>,
95 :
96 : /// has this file been used since last clock sweep?
97 : recently_used: AtomicBool,
98 : }
99 :
100 : struct SlotInner {
101 : /// Counter that's incremented every time a different file is stored here.
102 : /// To avoid the ABA problem.
103 : tag: u64,
104 :
105 : /// the underlying file
106 : file: Option<File>,
107 : }
108 :
109 : impl OpenFiles {
110 : /// Find a slot to use, evicting an existing file descriptor if needed.
111 : ///
112 : /// On return, we hold a lock on the slot, and its 'tag' has been updated
113 : /// recently_used has been set. It's all ready for reuse.
114 175895 : fn find_victim_slot(&self) -> (SlotHandle, RwLockWriteGuard<SlotInner>) {
115 175895 : //
116 175895 : // Run the clock algorithm to find a slot to replace.
117 175895 : //
118 175895 : let num_slots = self.slots.len();
119 175895 : let mut retries = 0;
120 : let mut slot;
121 : let mut slot_guard;
122 : let index;
123 355021 : loop {
124 355021 : let next = self.next.fetch_add(1, Ordering::AcqRel) % num_slots;
125 355021 : slot = &self.slots[next];
126 355021 :
127 355021 : // If the recently_used flag on this slot is set, continue the clock
128 355021 : // sweep. Otherwise try to use this slot. If we cannot acquire the
129 355021 : // lock, also continue the clock sweep.
130 355021 : //
131 355021 : // We only continue in this manner for a while, though. If we loop
132 355021 : // through the array twice without finding a victim, just pick the
133 355021 : // next slot and wait until we can reuse it. This way, we avoid
134 355021 : // spinning in the extreme case that all the slots are busy with an
135 355021 : // I/O operation.
136 355021 : if retries < num_slots * 2 {
137 354796 : if !slot.recently_used.swap(false, Ordering::Release) {
138 211771 : if let Ok(guard) = slot.inner.try_write() {
139 175670 : slot_guard = guard;
140 175670 : index = next;
141 175670 : break;
142 36101 : }
143 143025 : }
144 179126 : retries += 1;
145 : } else {
146 225 : slot_guard = slot.inner.write().unwrap();
147 225 : index = next;
148 225 : break;
149 : }
150 : }
151 :
152 : //
153 : // We now have the victim slot locked. If it was in use previously, close the
154 : // old file.
155 : //
156 175895 : if let Some(old_file) = slot_guard.file.take() {
157 120767 : // the normal path of dropping VirtualFile uses "close", use "close-by-replace" here to
158 120767 : // distinguish the two.
159 120767 : STORAGE_IO_TIME_METRIC
160 120767 : .get(StorageIoOperation::CloseByReplace)
161 120767 : .observe_closure_duration(|| drop(old_file));
162 120767 : }
163 :
164 : // Prepare the slot for reuse and return it
165 175895 : slot_guard.tag += 1;
166 175895 : slot.recently_used.store(true, Ordering::Relaxed);
167 175895 : (
168 175895 : SlotHandle {
169 175895 : index,
170 175895 : tag: slot_guard.tag,
171 175895 : },
172 175895 : slot_guard,
173 175895 : )
174 175895 : }
175 : }
176 :
177 : /// Identify error types that should alwways terminate the process. Other
178 : /// error types may be elegible for retry.
179 UBC 0 : pub(crate) fn is_fatal_io_error(e: &std::io::Error) -> bool {
180 0 : use nix::errno::Errno::*;
181 0 : match e.raw_os_error().map(nix::errno::from_i32) {
182 : Some(EIO) => {
183 : // Terminate on EIO because we no longer trust the device to store
184 : // data safely, or to uphold persistence guarantees on fsync.
185 0 : true
186 : }
187 : Some(EROFS) => {
188 : // Terminate on EROFS because a filesystem is usually remounted
189 : // readonly when it has experienced some critical issue, so the same
190 : // logic as EIO applies.
191 0 : true
192 : }
193 : Some(EACCES) => {
194 : // Terminate on EACCESS because we should always have permissions
195 : // for our own data dir: if we don't, then we can't do our job and
196 : // need administrative intervention to fix permissions. Terminating
197 : // is the best way to make sure we stop cleanly rather than going
198 : // into infinite retry loops, and will make it clear to the outside
199 : // world that we need help.
200 0 : true
201 : }
202 : _ => {
203 : // Treat all other local file I/O errors are retryable. This includes:
204 : // - ENOSPC: we stay up and wait for eviction to free some space
205 : // - EINVAL, EBADF, EBADFD: this is a code bug, not a filesystem/hardware issue
206 : // - WriteZero, Interrupted: these are used internally VirtualFile
207 0 : false
208 : }
209 : }
210 0 : }
211 :
212 : /// Call this when the local filesystem gives us an error with an external
213 : /// cause: this includes EIO, EROFS, and EACCESS: all these indicate either
214 : /// bad storage or bad configuration, and we can't fix that from inside
215 : /// a running process.
216 0 : pub(crate) fn on_fatal_io_error(e: &std::io::Error, context: &str) -> ! {
217 0 : tracing::error!("Fatal I/O error: {e}: {context})");
218 0 : std::process::abort();
219 : }
220 :
221 : pub(crate) trait MaybeFatalIo<T> {
222 : fn maybe_fatal_err(self, context: &str) -> std::io::Result<T>;
223 : fn fatal_err(self, context: &str) -> T;
224 : }
225 :
226 : impl<T> MaybeFatalIo<T> for std::io::Result<T> {
227 : /// Terminate the process if the result is an error of a fatal type, else pass it through
228 : ///
229 : /// This is appropriate for writes, where we typically want to die on EIO/ACCES etc, but
230 : /// not on ENOSPC.
231 CBC 106 : fn maybe_fatal_err(self, context: &str) -> std::io::Result<T> {
232 106 : if let Err(e) = &self {
233 UBC 0 : if is_fatal_io_error(e) {
234 0 : on_fatal_io_error(e, context);
235 0 : }
236 CBC 106 : }
237 106 : self
238 106 : }
239 :
240 : /// Terminate the process on any I/O error.
241 : ///
242 : /// This is appropriate for reads on files that we know exist: they should always work.
243 2090 : fn fatal_err(self, context: &str) -> T {
244 2090 : match self {
245 2090 : Ok(v) => v,
246 UBC 0 : Err(e) => {
247 0 : on_fatal_io_error(&e, context);
248 : }
249 : }
250 CBC 2090 : }
251 : }
252 :
253 : impl VirtualFile {
254 : /// Open a file in read-only mode. Like File::open.
255 33301 : pub async fn open(path: &Utf8Path) -> Result<VirtualFile, std::io::Error> {
256 33301 : Self::open_with_options(path, OpenOptions::new().read(true)).await
257 33301 : }
258 :
259 : /// Create a new file for writing. If the file exists, it will be truncated.
260 : /// Like File::create.
261 14827 : pub async fn create(path: &Utf8Path) -> Result<VirtualFile, std::io::Error> {
262 14827 : Self::open_with_options(
263 14827 : path,
264 14827 : OpenOptions::new().write(true).create(true).truncate(true),
265 14827 : )
266 UBC 0 : .await
267 CBC 14827 : }
268 :
269 : /// Open a file with given options.
270 : ///
271 : /// Note: If any custom flags were set in 'open_options' through OpenOptionsExt,
272 : /// they will be applied also when the file is subsequently re-opened, not only
273 : /// on the first time. Make sure that's sane!
274 73736 : pub async fn open_with_options(
275 73736 : path: &Utf8Path,
276 73736 : open_options: &OpenOptions,
277 73736 : ) -> Result<VirtualFile, std::io::Error> {
278 73736 : let path_str = path.to_string();
279 73736 : let parts = path_str.split('/').collect::<Vec<&str>>();
280 73736 : let tenant_id;
281 73736 : let timeline_id;
282 73736 : if parts.len() > 5 && parts[parts.len() - 5] == TENANTS_SEGMENT_NAME {
283 64305 : tenant_id = parts[parts.len() - 4].to_string();
284 64305 : timeline_id = parts[parts.len() - 2].to_string();
285 64305 : } else {
286 9431 : tenant_id = "*".to_string();
287 9431 : timeline_id = "*".to_string();
288 9431 : }
289 73736 : let (handle, mut slot_guard) = get_open_files().find_victim_slot();
290 :
291 : // NB: there is also StorageIoOperation::OpenAfterReplace which is for the case
292 : // where our caller doesn't get to use the returned VirtualFile before its
293 : // slot gets re-used by someone else.
294 73736 : let file = STORAGE_IO_TIME_METRIC
295 73736 : .get(StorageIoOperation::Open)
296 73736 : .observe_closure_duration(|| open_options.open(path))?;
297 :
298 : // Strip all options other than read and write.
299 : //
300 : // It would perhaps be nicer to check just for the read and write flags
301 : // explicitly, but OpenOptions doesn't contain any functions to read flags,
302 : // only to set them.
303 73736 : let mut reopen_options = open_options.clone();
304 73736 : reopen_options.create(false);
305 73736 : reopen_options.create_new(false);
306 73736 : reopen_options.truncate(false);
307 73736 :
308 73736 : let vfile = VirtualFile {
309 73736 : handle: RwLock::new(handle),
310 73736 : pos: 0,
311 73736 : path: path.to_path_buf(),
312 73736 : open_options: reopen_options,
313 73736 : tenant_id,
314 73736 : timeline_id,
315 73736 : };
316 73736 :
317 73736 : // TODO: Under pressure, it's likely the slot will get re-used and
318 73736 : // the underlying file closed before they get around to using it.
319 73736 : // => https://github.com/neondatabase/neon/issues/6065
320 73736 : slot_guard.file.replace(file);
321 73736 :
322 73736 : Ok(vfile)
323 73736 : }
324 :
325 : /// Writes a file to the specified `final_path` in a crash safe fasion
326 : ///
327 : /// The file is first written to the specified tmp_path, and in a second
328 : /// step, the tmp path is renamed to the final path. As renames are
329 : /// atomic, a crash during the write operation will never leave behind a
330 : /// partially written file.
331 7469 : pub async fn crashsafe_overwrite(
332 7469 : final_path: &Utf8Path,
333 7469 : tmp_path: &Utf8Path,
334 7469 : content: &[u8],
335 7469 : ) -> std::io::Result<()> {
336 7469 : let Some(final_path_parent) = final_path.parent() else {
337 UBC 0 : return Err(std::io::Error::from_raw_os_error(
338 0 : nix::errno::Errno::EINVAL as i32,
339 0 : ));
340 : };
341 CBC 7469 : std::fs::remove_file(tmp_path).or_else(fs_ext::ignore_not_found)?;
342 7469 : let mut file = Self::open_with_options(
343 7469 : tmp_path,
344 7469 : OpenOptions::new()
345 7469 : .write(true)
346 7469 : // Use `create_new` so that, if we race with ourselves or something else,
347 7469 : // we bail out instead of causing damage.
348 7469 : .create_new(true),
349 7469 : )
350 UBC 0 : .await?;
351 CBC 7469 : file.write_all(content).await?;
352 7469 : file.sync_all().await?;
353 7469 : drop(file); // before the rename, that's important!
354 7469 : // renames are atomic
355 7469 : std::fs::rename(tmp_path, final_path)?;
356 : // Only open final path parent dirfd now, so that this operation only
357 : // ever holds one VirtualFile fd at a time. That's important because
358 : // the current `find_victim_slot` impl might pick the same slot for both
359 : // VirtualFile., and it eventually does a blocking write lock instead of
360 : // try_lock.
361 7469 : let final_parent_dirfd =
362 7469 : Self::open_with_options(final_path_parent, OpenOptions::new().read(true)).await?;
363 7469 : final_parent_dirfd.sync_all().await?;
364 7469 : Ok(())
365 7469 : }
366 :
367 : /// Call File::sync_all() on the underlying File.
368 35386 : pub async fn sync_all(&self) -> Result<(), Error> {
369 35386 : self.with_file(StorageIoOperation::Fsync, |file| file.sync_all())
370 UBC 0 : .await?
371 CBC 35386 : }
372 :
373 20448 : pub async fn metadata(&self) -> Result<fs::Metadata, Error> {
374 20448 : self.with_file(StorageIoOperation::Metadata, |file| file.metadata())
375 UBC 0 : .await?
376 CBC 20448 : }
377 :
378 : /// Helper function that looks up the underlying File for this VirtualFile,
379 : /// opening it and evicting some other File if necessary. It calls 'func'
380 : /// with the physical File.
381 12659316 : async fn with_file<F, R>(&self, op: StorageIoOperation, mut func: F) -> Result<R, Error>
382 12659316 : where
383 12659316 : F: FnMut(&File) -> R,
384 12659316 : {
385 12659316 : let open_files = get_open_files();
386 :
387 102159 : let mut handle_guard = {
388 : // Read the cached slot handle, and see if the slot that it points to still
389 : // contains our File.
390 : //
391 : // We only need to hold the handle lock while we read the current handle. If
392 : // another thread closes the file and recycles the slot for a different file,
393 : // we will notice that the handle we read is no longer valid and retry.
394 12659316 : let mut handle = *self.handle.read().unwrap();
395 12682940 : loop {
396 12682940 : // Check if the slot contains our File
397 12682940 : {
398 12682940 : let slot = &open_files.slots[handle.index];
399 12682940 : let slot_guard = slot.inner.read().unwrap();
400 12682940 : if slot_guard.tag == handle.tag {
401 12557157 : if let Some(file) = &slot_guard.file {
402 : // Found a cached file descriptor.
403 12557157 : slot.recently_used.store(true, Ordering::Relaxed);
404 12557157 : return Ok(STORAGE_IO_TIME_METRIC
405 12557157 : .get(op)
406 12557157 : .observe_closure_duration(|| func(file)));
407 UBC 0 : }
408 CBC 125783 : }
409 : }
410 :
411 : // The slot didn't contain our File. We will have to open it ourselves,
412 : // but before that, grab a write lock on handle in the VirtualFile, so
413 : // that no other thread will try to concurrently open the same file.
414 125783 : let handle_guard = self.handle.write().unwrap();
415 125783 :
416 125783 : // If another thread changed the handle while we were not holding the lock,
417 125783 : // then the handle might now be valid again. Loop back to retry.
418 125783 : if *handle_guard != handle {
419 23624 : handle = *handle_guard;
420 23624 : continue;
421 102159 : }
422 102159 : break handle_guard;
423 102159 : }
424 102159 : };
425 102159 :
426 102159 : // We need to open the file ourselves. The handle in the VirtualFile is
427 102159 : // now locked in write-mode. Find a free slot to put it in.
428 102159 : let (handle, mut slot_guard) = open_files.find_victim_slot();
429 :
430 : // Re-open the physical file.
431 : // NB: we use StorageIoOperation::OpenAferReplace for this to distinguish this
432 : // case from StorageIoOperation::Open. This helps with identifying thrashing
433 : // of the virtual file descriptor cache.
434 102159 : let file = STORAGE_IO_TIME_METRIC
435 102159 : .get(StorageIoOperation::OpenAfterReplace)
436 102159 : .observe_closure_duration(|| self.open_options.open(&self.path))?;
437 :
438 : // Perform the requested operation on it
439 102159 : let result = STORAGE_IO_TIME_METRIC
440 102159 : .get(op)
441 102159 : .observe_closure_duration(|| func(&file));
442 102159 :
443 102159 : // Store the File in the slot and update the handle in the VirtualFile
444 102159 : // to point to it.
445 102159 : slot_guard.file.replace(file);
446 102159 :
447 102159 : *handle_guard = handle;
448 102159 :
449 102159 : Ok(result)
450 12659316 : }
451 :
452 UBC 0 : pub fn remove(self) {
453 0 : let path = self.path.clone();
454 0 : drop(self);
455 0 : std::fs::remove_file(path).expect("failed to remove the virtual file");
456 0 : }
457 :
458 CBC 61363 : pub async fn seek(&mut self, pos: SeekFrom) -> Result<u64, Error> {
459 61363 : match pos {
460 61358 : SeekFrom::Start(offset) => {
461 61358 : self.pos = offset;
462 61358 : }
463 2 : SeekFrom::End(offset) => {
464 2 : self.pos = self
465 2 : .with_file(StorageIoOperation::Seek, |mut file| {
466 2 : file.seek(SeekFrom::End(offset))
467 2 : })
468 1 : .await??
469 : }
470 3 : SeekFrom::Current(offset) => {
471 3 : let pos = self.pos as i128 + offset as i128;
472 3 : if pos < 0 {
473 1 : return Err(Error::new(
474 1 : ErrorKind::InvalidInput,
475 1 : "offset would be negative",
476 1 : ));
477 2 : }
478 2 : if pos > u64::MAX as i128 {
479 UBC 0 : return Err(Error::new(ErrorKind::InvalidInput, "offset overflow"));
480 CBC 2 : }
481 2 : self.pos = pos as u64;
482 : }
483 : }
484 61361 : Ok(self.pos)
485 61363 : }
486 :
487 : // Copied from https://doc.rust-lang.org/1.72.0/src/std/os/unix/fs.rs.html#117-135
488 4718002 : pub async fn read_exact_at(&self, mut buf: &mut [u8], mut offset: u64) -> Result<(), Error> {
489 9436004 : while !buf.is_empty() {
490 4718002 : match self.read_at(buf, offset).await {
491 : Ok(0) => {
492 UBC 0 : return Err(Error::new(
493 0 : std::io::ErrorKind::UnexpectedEof,
494 0 : "failed to fill whole buffer",
495 0 : ))
496 : }
497 CBC 4718002 : Ok(n) => {
498 4718002 : buf = &mut buf[n..];
499 4718002 : offset += n as u64;
500 4718002 : }
501 UBC 0 : Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {}
502 0 : Err(e) => return Err(e),
503 : }
504 : }
505 CBC 4718002 : Ok(())
506 4718002 : }
507 :
508 : // Copied from https://doc.rust-lang.org/1.72.0/src/std/os/unix/fs.rs.html#219-235
509 3223141 : pub async fn write_all_at(&self, mut buf: &[u8], mut offset: u64) -> Result<(), Error> {
510 6446282 : while !buf.is_empty() {
511 3223141 : match self.write_at(buf, offset).await {
512 : Ok(0) => {
513 UBC 0 : return Err(Error::new(
514 0 : std::io::ErrorKind::WriteZero,
515 0 : "failed to write whole buffer",
516 0 : ));
517 : }
518 CBC 3223141 : Ok(n) => {
519 3223141 : buf = &buf[n..];
520 3223141 : offset += n as u64;
521 3223141 : }
522 UBC 0 : Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {}
523 0 : Err(e) => return Err(e),
524 : }
525 : }
526 CBC 3223141 : Ok(())
527 3223141 : }
528 :
529 4662131 : pub async fn write_all(&mut self, mut buf: &[u8]) -> Result<(), Error> {
530 9324245 : while !buf.is_empty() {
531 4662115 : match self.write(buf).await {
532 : Ok(0) => {
533 UBC 0 : return Err(Error::new(
534 0 : std::io::ErrorKind::WriteZero,
535 0 : "failed to write whole buffer",
536 0 : ));
537 : }
538 CBC 4662114 : Ok(n) => {
539 4662114 : buf = &buf[n..];
540 4662114 : }
541 1 : Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {}
542 1 : Err(e) => return Err(e),
543 : }
544 : }
545 4662130 : Ok(())
546 4662131 : }
547 :
548 4662115 : async fn write(&mut self, buf: &[u8]) -> Result<usize, std::io::Error> {
549 4662115 : let pos = self.pos;
550 4662115 : let n = self.write_at(buf, pos).await?;
551 4662114 : self.pos += n as u64;
552 4662114 : Ok(n)
553 4662115 : }
554 :
555 4718224 : pub async fn read_at(&self, buf: &mut [u8], offset: u64) -> Result<usize, Error> {
556 4718224 : let result = self
557 4718224 : .with_file(StorageIoOperation::Read, |file| file.read_at(buf, offset))
558 UBC 0 : .await?;
559 CBC 4718224 : if let Ok(size) = result {
560 4718223 : STORAGE_IO_SIZE
561 4718223 : .with_label_values(&["read", &self.tenant_id, &self.timeline_id])
562 4718223 : .add(size as i64);
563 4718223 : }
564 4718224 : result
565 4718224 : }
566 :
567 7885256 : async fn write_at(&self, buf: &[u8], offset: u64) -> Result<usize, Error> {
568 7885256 : let result = self
569 7885256 : .with_file(StorageIoOperation::Write, |file| file.write_at(buf, offset))
570 UBC 0 : .await?;
571 CBC 7885256 : if let Ok(size) = result {
572 7885255 : STORAGE_IO_SIZE
573 7885255 : .with_label_values(&["write", &self.tenant_id, &self.timeline_id])
574 7885255 : .add(size as i64);
575 7885255 : }
576 7885256 : result
577 7885256 : }
578 : }
579 :
580 : #[cfg(test)]
581 : impl VirtualFile {
582 10100 : pub(crate) async fn read_blk(
583 10100 : &self,
584 10100 : blknum: u32,
585 10100 : ) -> Result<crate::tenant::block_io::BlockLease<'_>, std::io::Error> {
586 10100 : use crate::page_cache::PAGE_SZ;
587 10100 : let mut buf = [0; PAGE_SZ];
588 10100 : self.read_exact_at(&mut buf, blknum as u64 * (PAGE_SZ as u64))
589 UBC 0 : .await?;
590 CBC 10100 : Ok(std::sync::Arc::new(buf).into())
591 10100 : }
592 :
593 112 : async fn read_to_end(&mut self, buf: &mut Vec<u8>) -> Result<(), Error> {
594 222 : loop {
595 222 : let mut tmp = [0; 128];
596 222 : match self.read_at(&mut tmp, self.pos).await {
597 111 : Ok(0) => return Ok(()),
598 110 : Ok(n) => {
599 110 : self.pos += n as u64;
600 110 : buf.extend_from_slice(&tmp[..n]);
601 110 : }
602 1 : Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {}
603 1 : Err(e) => return Err(e),
604 : }
605 : }
606 112 : }
607 : }
608 :
609 : impl Drop for VirtualFile {
610 : /// If a VirtualFile is dropped, close the underlying file if it was open.
611 59947 : fn drop(&mut self) {
612 59947 : let handle = self.handle.get_mut().unwrap();
613 59947 :
614 59947 : // We could check with a read-lock first, to avoid waiting on an
615 59947 : // unrelated I/O.
616 59947 : let slot = &get_open_files().slots[handle.index];
617 59947 : let mut slot_guard = slot.inner.write().unwrap();
618 59947 : if slot_guard.tag == handle.tag {
619 49816 : slot.recently_used.store(false, Ordering::Relaxed);
620 : // there is also operation "close-by-replace" for closes done on eviction for
621 : // comparison.
622 49816 : if let Some(fd) = slot_guard.file.take() {
623 49816 : STORAGE_IO_TIME_METRIC
624 49816 : .get(StorageIoOperation::Close)
625 49816 : .observe_closure_duration(|| drop(fd));
626 49816 : }
627 10131 : }
628 59947 : }
629 : }
630 :
631 : impl OpenFiles {
632 608 : fn new(num_slots: usize) -> OpenFiles {
633 608 : let mut slots = Box::new(Vec::with_capacity(num_slots));
634 56210 : for _ in 0..num_slots {
635 56210 : let slot = Slot {
636 56210 : recently_used: AtomicBool::new(false),
637 56210 : inner: RwLock::new(SlotInner { tag: 0, file: None }),
638 56210 : };
639 56210 : slots.push(slot);
640 56210 : }
641 :
642 608 : OpenFiles {
643 608 : next: AtomicUsize::new(0),
644 608 : slots: Box::leak(slots),
645 608 : }
646 608 : }
647 : }
648 :
649 : ///
650 : /// Initialize the virtual file module. This must be called once at page
651 : /// server startup.
652 : ///
653 557 : pub fn init(num_slots: usize) {
654 557 : if OPEN_FILES.set(OpenFiles::new(num_slots)).is_err() {
655 UBC 0 : panic!("virtual_file::init called twice");
656 CBC 557 : }
657 557 : crate::metrics::virtual_file_descriptor_cache::SIZE_MAX.set(num_slots as u64);
658 557 : }
659 :
660 : const TEST_MAX_FILE_DESCRIPTORS: usize = 10;
661 :
662 : // Get a handle to the global slots array.
663 12792991 : fn get_open_files() -> &'static OpenFiles {
664 12792991 : //
665 12792991 : // In unit tests, page server startup doesn't happen and no one calls
666 12792991 : // virtual_file::init(). Initialize it here, with a small array.
667 12792991 : //
668 12792991 : // This applies to the virtual file tests below, but all other unit
669 12792991 : // tests too, so the virtual file facility is always usable in
670 12792991 : // unit tests.
671 12792991 : //
672 12792991 : if cfg!(test) {
673 185559 : OPEN_FILES.get_or_init(|| OpenFiles::new(TEST_MAX_FILE_DESCRIPTORS))
674 : } else {
675 12607432 : OPEN_FILES.get().expect("virtual_file::init not called yet")
676 : }
677 12792991 : }
678 :
679 : #[cfg(test)]
680 : mod tests {
681 : use super::*;
682 : use rand::seq::SliceRandom;
683 : use rand::thread_rng;
684 : use rand::Rng;
685 : use std::future::Future;
686 : use std::io::Write;
687 : use std::sync::Arc;
688 :
689 : enum MaybeVirtualFile {
690 : VirtualFile(VirtualFile),
691 : File(File),
692 : }
693 :
694 : impl From<VirtualFile> for MaybeVirtualFile {
695 3 : fn from(vf: VirtualFile) -> Self {
696 3 : MaybeVirtualFile::VirtualFile(vf)
697 3 : }
698 : }
699 :
700 : impl MaybeVirtualFile {
701 202 : async fn read_exact_at(&self, buf: &mut [u8], offset: u64) -> Result<(), Error> {
702 202 : match self {
703 101 : MaybeVirtualFile::VirtualFile(file) => file.read_exact_at(buf, offset).await,
704 101 : MaybeVirtualFile::File(file) => file.read_exact_at(buf, offset),
705 : }
706 202 : }
707 4 : async fn write_all_at(&self, buf: &[u8], offset: u64) -> Result<(), Error> {
708 4 : match self {
709 2 : MaybeVirtualFile::VirtualFile(file) => file.write_all_at(buf, offset).await,
710 2 : MaybeVirtualFile::File(file) => file.write_all_at(buf, offset),
711 : }
712 4 : }
713 18 : async fn seek(&mut self, pos: SeekFrom) -> Result<u64, Error> {
714 18 : match self {
715 9 : MaybeVirtualFile::VirtualFile(file) => file.seek(pos).await,
716 9 : MaybeVirtualFile::File(file) => file.seek(pos),
717 : }
718 18 : }
719 4 : async fn write_all(&mut self, buf: &[u8]) -> Result<(), Error> {
720 4 : match self {
721 2 : MaybeVirtualFile::VirtualFile(file) => file.write_all(buf).await,
722 2 : MaybeVirtualFile::File(file) => file.write_all(buf),
723 : }
724 4 : }
725 :
726 : // Helper function to slurp contents of a file, starting at the current position,
727 : // into a string
728 221 : async fn read_string(&mut self) -> Result<String, Error> {
729 221 : use std::io::Read;
730 221 : let mut buf = String::new();
731 221 : match self {
732 112 : MaybeVirtualFile::VirtualFile(file) => {
733 112 : let mut buf = Vec::new();
734 112 : file.read_to_end(&mut buf).await?;
735 111 : return Ok(String::from_utf8(buf).unwrap());
736 : }
737 109 : MaybeVirtualFile::File(file) => {
738 109 : file.read_to_string(&mut buf)?;
739 : }
740 : }
741 108 : Ok(buf)
742 221 : }
743 :
744 : // Helper function to slurp a portion of a file into a string
745 202 : async fn read_string_at(&mut self, pos: u64, len: usize) -> Result<String, Error> {
746 202 : let mut buf = vec![0; len];
747 202 : self.read_exact_at(&mut buf, pos).await?;
748 202 : Ok(String::from_utf8(buf).unwrap())
749 202 : }
750 : }
751 :
752 1 : #[tokio::test]
753 1 : async fn test_virtual_files() -> Result<(), Error> {
754 1 : // The real work is done in the test_files() helper function. This
755 1 : // allows us to run the same set of tests against a native File, and
756 1 : // VirtualFile. We trust the native Files and wouldn't need to test them,
757 1 : // but this allows us to verify that the operations return the same
758 1 : // results with VirtualFiles as with native Files. (Except that with
759 1 : // native files, you will run out of file descriptors if the ulimit
760 1 : // is low enough.)
761 103 : test_files("virtual_files", |path, open_options| async move {
762 103 : let vf = VirtualFile::open_with_options(&path, &open_options).await?;
763 103 : Ok(MaybeVirtualFile::VirtualFile(vf))
764 103 : })
765 UBC 0 : .await
766 : }
767 :
768 CBC 1 : #[tokio::test]
769 1 : async fn test_physical_files() -> Result<(), Error> {
770 103 : test_files("physical_files", |path, open_options| async move {
771 103 : Ok(MaybeVirtualFile::File(open_options.open(path)?))
772 103 : })
773 UBC 0 : .await
774 : }
775 :
776 CBC 2 : async fn test_files<OF, FT>(testname: &str, openfunc: OF) -> Result<(), Error>
777 2 : where
778 2 : OF: Fn(Utf8PathBuf, OpenOptions) -> FT,
779 2 : FT: Future<Output = Result<MaybeVirtualFile, std::io::Error>>,
780 2 : {
781 2 : let testdir = crate::config::PageServerConf::test_repo_dir(testname);
782 2 : std::fs::create_dir_all(&testdir)?;
783 :
784 2 : let path_a = testdir.join("file_a");
785 2 : let mut file_a = openfunc(
786 2 : path_a.clone(),
787 2 : OpenOptions::new()
788 2 : .write(true)
789 2 : .create(true)
790 2 : .truncate(true)
791 2 : .to_owned(),
792 2 : )
793 UBC 0 : .await?;
794 CBC 2 : file_a.write_all(b"foobar").await?;
795 :
796 : // cannot read from a file opened in write-only mode
797 2 : let _ = file_a.read_string().await.unwrap_err();
798 :
799 : // Close the file and re-open for reading
800 2 : let mut file_a = openfunc(path_a, OpenOptions::new().read(true).to_owned()).await?;
801 :
802 : // cannot write to a file opened in read-only mode
803 2 : let _ = file_a.write_all(b"bar").await.unwrap_err();
804 2 :
805 2 : // Try simple read
806 2 : assert_eq!("foobar", file_a.read_string().await?);
807 :
808 : // It's positioned at the EOF now.
809 2 : assert_eq!("", file_a.read_string().await?);
810 :
811 : // Test seeks.
812 2 : assert_eq!(file_a.seek(SeekFrom::Start(1)).await?, 1);
813 2 : assert_eq!("oobar", file_a.read_string().await?);
814 :
815 2 : assert_eq!(file_a.seek(SeekFrom::End(-2)).await?, 4);
816 2 : assert_eq!("ar", file_a.read_string().await?);
817 :
818 2 : assert_eq!(file_a.seek(SeekFrom::Start(1)).await?, 1);
819 2 : assert_eq!(file_a.seek(SeekFrom::Current(2)).await?, 3);
820 2 : assert_eq!("bar", file_a.read_string().await?);
821 :
822 2 : assert_eq!(file_a.seek(SeekFrom::Current(-5)).await?, 1);
823 2 : assert_eq!("oobar", file_a.read_string().await?);
824 :
825 : // Test erroneous seeks to before byte 0
826 2 : file_a.seek(SeekFrom::End(-7)).await.unwrap_err();
827 2 : assert_eq!(file_a.seek(SeekFrom::Start(1)).await?, 1);
828 2 : file_a.seek(SeekFrom::Current(-2)).await.unwrap_err();
829 2 :
830 2 : // the erroneous seek should have left the position unchanged
831 2 : assert_eq!("oobar", file_a.read_string().await?);
832 :
833 : // Create another test file, and try FileExt functions on it.
834 2 : let path_b = testdir.join("file_b");
835 2 : let mut file_b = openfunc(
836 2 : path_b.clone(),
837 2 : OpenOptions::new()
838 2 : .read(true)
839 2 : .write(true)
840 2 : .create(true)
841 2 : .truncate(true)
842 2 : .to_owned(),
843 2 : )
844 UBC 0 : .await?;
845 CBC 2 : file_b.write_all_at(b"BAR", 3).await?;
846 2 : file_b.write_all_at(b"FOO", 0).await?;
847 :
848 2 : assert_eq!(file_b.read_string_at(2, 3).await?, "OBA");
849 :
850 : // Open a lot of files, enough to cause some evictions. (Or to be precise,
851 : // open the same file many times. The effect is the same.)
852 : //
853 : // leave file_a positioned at offset 1 before we start
854 2 : assert_eq!(file_a.seek(SeekFrom::Start(1)).await?, 1);
855 :
856 2 : let mut vfiles = Vec::new();
857 202 : for _ in 0..100 {
858 200 : let mut vfile =
859 200 : openfunc(path_b.clone(), OpenOptions::new().read(true).to_owned()).await?;
860 200 : assert_eq!("FOOBAR", vfile.read_string().await?);
861 200 : vfiles.push(vfile);
862 : }
863 :
864 : // make sure we opened enough files to definitely cause evictions.
865 2 : assert!(vfiles.len() > TEST_MAX_FILE_DESCRIPTORS * 2);
866 :
867 : // The underlying file descriptor for 'file_a' should be closed now. Try to read
868 : // from it again. We left the file positioned at offset 1 above.
869 2 : assert_eq!("oobar", file_a.read_string().await?);
870 :
871 : // Check that all the other FDs still work too. Use them in random order for
872 : // good measure.
873 2 : vfiles.as_mut_slice().shuffle(&mut thread_rng());
874 200 : for vfile in vfiles.iter_mut() {
875 200 : assert_eq!("OOBAR", vfile.read_string_at(1, 5).await?);
876 : }
877 :
878 2 : Ok(())
879 2 : }
880 :
881 : /// Test using VirtualFiles from many threads concurrently. This tests both using
882 : /// a lot of VirtualFiles concurrently, causing evictions, and also using the same
883 : /// VirtualFile from multiple threads concurrently.
884 1 : #[tokio::test]
885 1 : async fn test_vfile_concurrency() -> Result<(), Error> {
886 1 : const SIZE: usize = 8 * 1024;
887 1 : const VIRTUAL_FILES: usize = 100;
888 1 : const THREADS: usize = 100;
889 1 : const SAMPLE: [u8; SIZE] = [0xADu8; SIZE];
890 1 :
891 1 : let testdir = crate::config::PageServerConf::test_repo_dir("vfile_concurrency");
892 1 : std::fs::create_dir_all(&testdir)?;
893 :
894 : // Create a test file.
895 1 : let test_file_path = testdir.join("concurrency_test_file");
896 : {
897 1 : let file = File::create(&test_file_path)?;
898 1 : file.write_all_at(&SAMPLE, 0)?;
899 : }
900 :
901 : // Open the file many times.
902 1 : let mut files = Vec::new();
903 101 : for _ in 0..VIRTUAL_FILES {
904 100 : let f = VirtualFile::open_with_options(&test_file_path, OpenOptions::new().read(true))
905 UBC 0 : .await?;
906 CBC 100 : files.push(f);
907 : }
908 1 : let files = Arc::new(files);
909 1 :
910 1 : // Launch many threads, and use the virtual files concurrently in random order.
911 1 : let rt = tokio::runtime::Builder::new_multi_thread()
912 1 : .worker_threads(THREADS)
913 1 : .thread_name("test_vfile_concurrency thread")
914 1 : .build()
915 1 : .unwrap();
916 1 : let mut hdls = Vec::new();
917 101 : for _threadno in 0..THREADS {
918 100 : let files = files.clone();
919 100 : let hdl = rt.spawn(async move {
920 100 : let mut buf = [0u8; SIZE];
921 100 : let mut rng = rand::rngs::OsRng;
922 100000 : for _ in 1..1000 {
923 99900 : let f = &files[rng.gen_range(0..files.len())];
924 99900 : f.read_exact_at(&mut buf, 0).await.unwrap();
925 99900 : assert!(buf == SAMPLE);
926 : }
927 100 : });
928 100 : hdls.push(hdl);
929 100 : }
930 101 : for hdl in hdls {
931 100 : hdl.await?;
932 : }
933 1 : std::mem::forget(rt);
934 1 :
935 1 : Ok(())
936 : }
937 :
938 1 : #[tokio::test]
939 1 : async fn test_atomic_overwrite_basic() {
940 1 : let testdir = crate::config::PageServerConf::test_repo_dir("test_atomic_overwrite_basic");
941 1 : std::fs::create_dir_all(&testdir).unwrap();
942 1 :
943 1 : let path = testdir.join("myfile");
944 1 : let tmp_path = testdir.join("myfile.tmp");
945 1 :
946 1 : VirtualFile::crashsafe_overwrite(&path, &tmp_path, b"foo")
947 UBC 0 : .await
948 CBC 1 : .unwrap();
949 1 : let mut file = MaybeVirtualFile::from(VirtualFile::open(&path).await.unwrap());
950 1 : let post = file.read_string().await.unwrap();
951 1 : assert_eq!(post, "foo");
952 1 : assert!(!tmp_path.exists());
953 1 : drop(file);
954 1 :
955 1 : VirtualFile::crashsafe_overwrite(&path, &tmp_path, b"bar")
956 UBC 0 : .await
957 CBC 1 : .unwrap();
958 1 : let mut file = MaybeVirtualFile::from(VirtualFile::open(&path).await.unwrap());
959 1 : let post = file.read_string().await.unwrap();
960 1 : assert_eq!(post, "bar");
961 1 : assert!(!tmp_path.exists());
962 1 : drop(file);
963 : }
964 :
965 1 : #[tokio::test]
966 1 : async fn test_atomic_overwrite_preexisting_tmp() {
967 1 : let testdir =
968 1 : crate::config::PageServerConf::test_repo_dir("test_atomic_overwrite_preexisting_tmp");
969 1 : std::fs::create_dir_all(&testdir).unwrap();
970 1 :
971 1 : let path = testdir.join("myfile");
972 1 : let tmp_path = testdir.join("myfile.tmp");
973 1 :
974 1 : std::fs::write(&tmp_path, "some preexisting junk that should be removed").unwrap();
975 1 : assert!(tmp_path.exists());
976 :
977 1 : VirtualFile::crashsafe_overwrite(&path, &tmp_path, b"foo")
978 UBC 0 : .await
979 CBC 1 : .unwrap();
980 :
981 1 : let mut file = MaybeVirtualFile::from(VirtualFile::open(&path).await.unwrap());
982 1 : let post = file.read_string().await.unwrap();
983 1 : assert_eq!(post, "foo");
984 1 : assert!(!tmp_path.exists());
985 1 : drop(file);
986 : }
987 : }
|
