Line data Source code
1 : //!
2 : //! Low-level Block-oriented I/O functions
3 : //!
4 :
5 : use super::ephemeral_file::EphemeralFile;
6 : use super::storage_layer::delta_layer::{Adapter, DeltaLayerInner};
7 : use crate::context::RequestContext;
8 : use crate::page_cache::{self, FileId, PageReadGuard, PageWriteGuard, ReadBufResult, PAGE_SZ};
9 : use crate::virtual_file::VirtualFile;
10 : use bytes::Bytes;
11 : use std::ops::Deref;
12 :
13 : /// This is implemented by anything that can read 8 kB (PAGE_SZ)
14 : /// blocks, using the page cache
15 : ///
16 : /// There are currently two implementations: EphemeralFile, and FileBlockReader
17 : /// below.
18 : pub trait BlockReader {
19 : ///
20 : /// Create a new "cursor" for reading from this reader.
21 : ///
22 : /// A cursor caches the last accessed page, allowing for faster
23 : /// access if the same block is accessed repeatedly.
24 : fn block_cursor(&self) -> BlockCursor<'_>;
25 : }
26 :
27 : impl<B> BlockReader for &B
28 : where
29 : B: BlockReader,
30 : {
31 211493 : fn block_cursor(&self) -> BlockCursor<'_> {
32 211493 : (*self).block_cursor()
33 211493 : }
34 : }
35 :
36 : /// Reference to an in-memory copy of an immutable on-disk block.
37 : pub enum BlockLease<'a> {
38 : PageReadGuard(PageReadGuard<'static>),
39 : EphemeralFileMutableTail(&'a [u8; PAGE_SZ]),
40 : #[cfg(test)]
41 : Arc(std::sync::Arc<[u8; PAGE_SZ]>),
42 : #[cfg(test)]
43 : Vec(Vec<u8>),
44 : }
45 :
46 : impl From<PageReadGuard<'static>> for BlockLease<'static> {
47 2740532 : fn from(value: PageReadGuard<'static>) -> BlockLease<'static> {
48 2740532 : BlockLease::PageReadGuard(value)
49 2740532 : }
50 : }
51 :
52 : #[cfg(test)]
53 : impl<'a> From<std::sync::Arc<[u8; PAGE_SZ]>> for BlockLease<'a> {
54 1016998 : fn from(value: std::sync::Arc<[u8; PAGE_SZ]>) -> Self {
55 1016998 : BlockLease::Arc(value)
56 1016998 : }
57 : }
58 :
59 : impl<'a> Deref for BlockLease<'a> {
60 : type Target = [u8; PAGE_SZ];
61 :
62 15853781 : fn deref(&self) -> &Self::Target {
63 15853781 : match self {
64 13499674 : BlockLease::PageReadGuard(v) => v.deref(),
65 1302601 : BlockLease::EphemeralFileMutableTail(v) => v,
66 : #[cfg(test)]
67 1016998 : BlockLease::Arc(v) => v.deref(),
68 : #[cfg(test)]
69 34508 : BlockLease::Vec(v) => {
70 34508 : TryFrom::try_from(&v[..]).expect("caller must ensure that v has PAGE_SZ")
71 : }
72 : }
73 15853781 : }
74 : }
75 :
76 : /// Provides the ability to read blocks from different sources,
77 : /// similar to using traits for this purpose.
78 : ///
79 : /// Unlike traits, we also support the read function to be async though.
80 : pub(crate) enum BlockReaderRef<'a> {
81 : FileBlockReader(&'a FileBlockReader<'a>),
82 : EphemeralFile(&'a EphemeralFile),
83 : Adapter(Adapter<&'a DeltaLayerInner>),
84 : #[cfg(test)]
85 : TestDisk(&'a super::disk_btree::tests::TestDisk),
86 : #[cfg(test)]
87 : VirtualFile(&'a VirtualFile),
88 : }
89 :
90 : impl<'a> BlockReaderRef<'a> {
91 : #[inline(always)]
92 8732551 : async fn read_blk(
93 8732551 : &self,
94 8732551 : blknum: u32,
95 8732551 : ctx: &RequestContext,
96 8732551 : ) -> Result<BlockLease, std::io::Error> {
97 8732551 : use BlockReaderRef::*;
98 8732551 : match self {
99 656442 : FileBlockReader(r) => r.read_blk(blknum, ctx).await,
100 4955909 : EphemeralFile(r) => r.read_blk(blknum, ctx).await,
101 2083002 : Adapter(r) => r.read_blk(blknum, ctx).await,
102 : #[cfg(test)]
103 1016998 : TestDisk(r) => r.read_blk(blknum),
104 : #[cfg(test)]
105 20200 : VirtualFile(r) => r.read_blk(blknum, ctx).await,
106 : }
107 8732551 : }
108 : }
109 :
110 : ///
111 : /// A "cursor" for efficiently reading multiple pages from a BlockReader
112 : ///
113 : /// You can access the last page with `*cursor`. 'read_blk' returns 'self', so
114 : /// that in many cases you can use a BlockCursor as a drop-in replacement for
115 : /// the underlying BlockReader. For example:
116 : ///
117 : /// ```no_run
118 : /// # use pageserver::tenant::block_io::{BlockReader, FileBlockReader};
119 : /// # use pageserver::context::RequestContext;
120 : /// # let reader: FileBlockReader = unimplemented!("stub");
121 : /// # let ctx: RequestContext = unimplemented!("stub");
122 : /// let cursor = reader.block_cursor();
123 : /// let buf = cursor.read_blk(1, &ctx);
124 : /// // do stuff with 'buf'
125 : /// let buf = cursor.read_blk(2, &ctx);
126 : /// // do stuff with 'buf'
127 : /// ```
128 : ///
129 : pub struct BlockCursor<'a> {
130 : reader: BlockReaderRef<'a>,
131 : }
132 :
133 : impl<'a> BlockCursor<'a> {
134 3570422 : pub(crate) fn new(reader: BlockReaderRef<'a>) -> Self {
135 3570422 : BlockCursor { reader }
136 3570422 : }
137 : // Needed by cli
138 0 : pub fn new_fileblockreader(reader: &'a FileBlockReader) -> Self {
139 0 : BlockCursor {
140 0 : reader: BlockReaderRef::FileBlockReader(reader),
141 0 : }
142 0 : }
143 :
144 : /// Read a block.
145 : ///
146 : /// Returns a "lease" object that can be used to
147 : /// access to the contents of the page. (For the page cache, the
148 : /// lease object represents a lock on the buffer.)
149 : #[inline(always)]
150 8732551 : pub async fn read_blk(
151 8732551 : &self,
152 8732551 : blknum: u32,
153 8732551 : ctx: &RequestContext,
154 8732551 : ) -> Result<BlockLease, std::io::Error> {
155 8732551 : self.reader.read_blk(blknum, ctx).await
156 8732551 : }
157 : }
158 :
159 : /// An adapter for reading a (virtual) file using the page cache.
160 : ///
161 : /// The file is assumed to be immutable. This doesn't provide any functions
162 : /// for modifying the file, nor for invalidating the cache if it is modified.
163 : #[derive(Clone)]
164 : pub struct FileBlockReader<'a> {
165 : pub file: &'a VirtualFile,
166 :
167 : /// Unique ID of this file, used as key in the page cache.
168 : file_id: page_cache::FileId,
169 : }
170 :
171 : impl<'a> FileBlockReader<'a> {
172 2296519 : pub fn new(file: &'a VirtualFile, file_id: FileId) -> Self {
173 2296519 : FileBlockReader { file_id, file }
174 2296519 : }
175 :
176 : /// Read a page from the underlying file into given buffer.
177 60136 : async fn fill_buffer(
178 60136 : &self,
179 60136 : buf: PageWriteGuard<'static>,
180 60136 : blkno: u32,
181 60136 : ctx: &RequestContext,
182 60136 : ) -> Result<PageWriteGuard<'static>, std::io::Error> {
183 60136 : assert!(buf.len() == PAGE_SZ);
184 60136 : self.file
185 60136 : .read_exact_at_page(buf, blkno as u64 * PAGE_SZ as u64, ctx)
186 36421 : .await
187 60136 : }
188 : /// Read a block.
189 : ///
190 : /// Returns a "lease" object that can be used to
191 : /// access to the contents of the page. (For the page cache, the
192 : /// lease object represents a lock on the buffer.)
193 2740532 : pub async fn read_blk<'b>(
194 2740532 : &self,
195 2740532 : blknum: u32,
196 2740532 : ctx: &RequestContext,
197 2740532 : ) -> Result<BlockLease<'b>, std::io::Error> {
198 2740532 : let cache = page_cache::get();
199 2740532 : match cache
200 2740532 : .read_immutable_buf(self.file_id, blknum, ctx)
201 34030 : .await
202 2740532 : .map_err(|e| {
203 0 : std::io::Error::new(
204 0 : std::io::ErrorKind::Other,
205 0 : format!("Failed to read immutable buf: {e:#}"),
206 0 : )
207 2740532 : })? {
208 2680396 : ReadBufResult::Found(guard) => Ok(guard.into()),
209 60136 : ReadBufResult::NotFound(write_guard) => {
210 : // Read the page from disk into the buffer
211 60136 : let write_guard = self.fill_buffer(write_guard, blknum, ctx).await?;
212 60136 : Ok(write_guard.mark_valid().into())
213 : }
214 : }
215 2740532 : }
216 : }
217 :
218 : impl BlockReader for FileBlockReader<'_> {
219 487796 : fn block_cursor(&self) -> BlockCursor<'_> {
220 487796 : BlockCursor::new(BlockReaderRef::FileBlockReader(self))
221 487796 : }
222 : }
223 :
224 : ///
225 : /// Trait for block-oriented output
226 : ///
227 : pub trait BlockWriter {
228 : ///
229 : /// Write a page to the underlying storage.
230 : ///
231 : /// 'buf' must be of size PAGE_SZ. Returns the block number the page was
232 : /// written to.
233 : ///
234 : fn write_blk(&mut self, buf: Bytes) -> Result<u32, std::io::Error>;
235 : }
236 :
237 : ///
238 : /// A simple in-memory buffer of blocks.
239 : ///
240 : pub struct BlockBuf {
241 : pub blocks: Vec<Bytes>,
242 : }
243 : impl BlockWriter for BlockBuf {
244 14223 : fn write_blk(&mut self, buf: Bytes) -> Result<u32, std::io::Error> {
245 14223 : assert!(buf.len() == PAGE_SZ);
246 14223 : let blknum = self.blocks.len();
247 14223 : self.blocks.push(buf);
248 14223 : Ok(blknum as u32)
249 14223 : }
250 : }
251 :
252 : impl BlockBuf {
253 1568 : pub fn new() -> Self {
254 1568 : BlockBuf { blocks: Vec::new() }
255 1568 : }
256 :
257 2023972 : pub fn size(&self) -> u64 {
258 2023972 : (self.blocks.len() * PAGE_SZ) as u64
259 2023972 : }
260 : }
261 : impl Default for BlockBuf {
262 0 : fn default() -> Self {
263 0 : Self::new()
264 0 : }
265 : }
|
