rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-07-06 05:30:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

todo!() for a world where the blob writer struct is gone and we use the TBD EphemeralFile superclass instead

Browse Source
This commit is contained in:
Christian Schwarz
2023-08-31 08:46:23 +00:00
parent a66990883c
commit 946bebc05a
3 changed files with 15 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
pageserver/src/tenant/blob_io.rs
View File

@@ -91,71 +91,3 @@ pub trait BlobWriter {
/// which can be used to retrieve the data later.
fn write_blob(&mut self, srcbuf: &[u8]) -> Result<u64, Error>;
}
///
/// An implementation of BlobWriter to write blobs to anything that
/// implements std::io::Write.
///
pub struct WriteBlobWriter<W>
where
W: std::io::Write,
{
inner: W,
offset: u64,
}
impl<W> WriteBlobWriter<W>
where
W: std::io::Write,
{
pub fn new(inner: W, start_offset: u64) -> Self {
WriteBlobWriter {
inner,
offset: start_offset,
}
}
pub fn size(&self) -> u64 {
self.offset
}
/// Access the underlying Write object.
///
/// NOTE: WriteBlobWriter keeps track of the current write offset. If
/// you write something directly to the inner Write object, it makes the
/// internally tracked 'offset' to go out of sync. So don't do that.
pub fn into_inner(self) -> W {
self.inner
}
}
impl<W> BlobWriter for WriteBlobWriter<W>
where
W: std::io::Write,
{
fn write_blob(&mut self, srcbuf: &[u8]) -> Result<u64, Error> {
let offset = self.offset;
if srcbuf.len() < 128 {
// Short blob. Write a 1-byte length header
let len_buf = srcbuf.len() as u8;
self.inner.write_all(&[len_buf])?;
self.offset += 1;
} else {
// Write a 4-byte length header
if srcbuf.len() > 0x7fff_ffff {
return Err(Error::new(
ErrorKind::Other,
format!("blob too large ({} bytes)", srcbuf.len()),
));
}
let mut len_buf = ((srcbuf.len()) as u32).to_be_bytes();
len_buf[0] |= 0x80;
self.inner.write_all(&len_buf)?;
self.offset += 4;
}
self.inner.write_all(srcbuf)?;
self.offset += srcbuf.len() as u64;
Ok(offset)
}
}

45
pageserver/src/tenant/storage_layer/delta_layer.rs
View File

@@ -31,7 +31,7 @@ use crate::config::PageServerConf;
use crate::context::RequestContext;
use crate::page_cache::PAGE_SZ;
use crate::repository::{Key, Value, KEY_SIZE};
use crate::tenant::blob_io::{BlobWriter, WriteBlobWriter};
use crate::tenant::blob_io::BlobWriter;
use crate::tenant::block_io::{BlockBuf, BlockCursor, BlockLease, BlockReader, FileBlockReader};
use crate::tenant::disk_btree::{DiskBtreeBuilder, DiskBtreeReader, VisitDirection};
use crate::tenant::storage_layer::{
@@ -582,8 +582,6 @@ struct DeltaLayerWriterInner {
lsn_range: Range<Lsn>,
tree: DiskBtreeBuilder<BlockBuf, DELTA_KEY_SIZE>,
blob_writer: WriteBlobWriter<BufWriter<VirtualFile>>,
}
impl DeltaLayerWriterInner {
@@ -605,12 +603,6 @@ impl DeltaLayerWriterInner {
// FIXME: throw an error instead?
let path = DeltaLayer::temp_path_for(conf, &tenant_id, &timeline_id, key_start, &lsn_range);
let mut file = VirtualFile::create(&path)?;
// make room for the header block
file.seek(SeekFrom::Start(PAGE_SZ as u64))?;
let buf_writer = BufWriter::new(file);
let blob_writer = WriteBlobWriter::new(buf_writer, PAGE_SZ as u64);
// Initialize the b-tree index builder
let block_buf = BlockBuf::new();
let tree_builder = DiskBtreeBuilder::new(block_buf);
@@ -623,7 +615,6 @@ impl DeltaLayerWriterInner {
key_start,
lsn_range,
tree: tree_builder,
blob_writer,
})
}
@@ -633,7 +624,8 @@ impl DeltaLayerWriterInner {
/// The values must be appended in key, lsn order.
///
async fn put_value(&mut self, key: Key, lsn: Lsn, val: Value) -> anyhow::Result<()> {
self.put_value_bytes(key, lsn, &Value::ser(&val)?, val.will_init()).await
self.put_value_bytes(key, lsn, &Value::ser(&val)?, val.will_init())
.await
}
async fn put_value_bytes(
@@ -644,30 +636,20 @@ impl DeltaLayerWriterInner {
will_init: bool,
) -> anyhow::Result<()> {
assert!(self.lsn_range.start <= lsn);
let off = self.blob_writer.write_blob(val)?; // TOOD: do what we discussed
let blob_ref = BlobRef::new(off, will_init);
let delta_key = DeltaKey::from_key_lsn(&key, lsn);
self.tree.append(&delta_key.0, blob_ref.0)?;
Ok(())
todo!("use TBD EphemeralFile superclass");
}
fn size(&self) -> u64 {
self.blob_writer.size() + self.tree.borrow_writer().size()
todo!()
}
///
/// Finish writing the delta layer.
///
fn finish(self, key_end: Key) -> anyhow::Result<DeltaLayer> {
let index_start_blk =
((self.blob_writer.size() + PAGE_SZ as u64 - 1) / PAGE_SZ as u64) as u32;
let index_start_blk: u32 = todo!();
let buf_writer = self.blob_writer.into_inner();
let mut file = buf_writer.into_inner()?;
let mut file: VirtualFile = todo!("EphemeralFile superclass needs into_inner() api");
// Write out the index
let (index_root_blk, block_buf) = self.tree.finish()?;
@@ -811,7 +793,8 @@ impl DeltaLayerWriter {
self.inner
.as_mut()
.unwrap()
.put_value_bytes(key, lsn, val, will_init).await
.put_value_bytes(key, lsn, val, will_init)
.await
}
pub fn size(&self) -> u64 {
@@ -828,15 +811,7 @@ impl DeltaLayerWriter {
impl Drop for DeltaLayerWriter {
fn drop(&mut self) {
if let Some(inner) = self.inner.take() {
match inner.blob_writer.into_inner().into_inner() {
Ok(vfile) => vfile.remove(),
Err(err) => warn!(
"error while flushing buffer of image layer temporary file: {}",
err
),
}
}
todo!("TBD EpheemralFile superclass into_inner(); => VirtualFile => remove()");
}
}

26
pageserver/src/tenant/storage_layer/image_layer.rs
View File

@@ -27,7 +27,7 @@ use crate::config::PageServerConf;
use crate::context::RequestContext;
use crate::page_cache::PAGE_SZ;
use crate::repository::{Key, KEY_SIZE};
use crate::tenant::blob_io::{BlobWriter, WriteBlobWriter};
use crate::tenant::blob_io::BlobWriter;
use crate::tenant::block_io::{BlockBuf, BlockReader, FileBlockReader};
use crate::tenant::disk_btree::{DiskBtreeBuilder, DiskBtreeReader, VisitDirection};
use crate::tenant::storage_layer::{
@@ -511,7 +511,6 @@ struct ImageLayerWriterInner {
key_range: Range<Key>,
lsn: Lsn,
blob_writer: WriteBlobWriter<VirtualFile>,
tree: DiskBtreeBuilder<BlockBuf, KEY_SIZE>,
}
@@ -538,13 +537,6 @@ impl ImageLayerWriterInner {
},
);
info!("new image layer {}", path.display());
let mut file = VirtualFile::open_with_options(
&path,
std::fs::OpenOptions::new().write(true).create_new(true),
)?;
// make room for the header block
file.seek(SeekFrom::Start(PAGE_SZ as u64))?;
let blob_writer = WriteBlobWriter::new(file, PAGE_SZ as u64);
// Initialize the b-tree index builder
let block_buf = BlockBuf::new();
@@ -558,7 +550,6 @@ impl ImageLayerWriterInner {
key_range: key_range.clone(),
lsn,
tree: tree_builder,
blob_writer,
};
Ok(writer)
@@ -570,13 +561,7 @@ impl ImageLayerWriterInner {
/// The page versions must be appended in blknum order.
///
fn put_image(&mut self, key: Key, img: &[u8]) -> anyhow::Result<()> {
ensure!(self.key_range.contains(&key));
let off = self.blob_writer.write_blob(img)?;
let mut keybuf: [u8; KEY_SIZE] = [0u8; KEY_SIZE];
key.write_to_byte_slice(&mut keybuf);
self.tree.append(&keybuf, off)?;
todo!("use TBD EphemeralFile superclass that skips copying into mutable_tail");
Ok(())
}
@@ -584,10 +569,9 @@ impl ImageLayerWriterInner {
/// Finish writing the image layer.
///
fn finish(self) -> anyhow::Result<ImageLayer> {
let index_start_blk =
((self.blob_writer.size() + PAGE_SZ as u64 - 1) / PAGE_SZ as u64) as u32;
let index_start_blk: u32 = todo!();
let mut file = self.blob_writer.into_inner();
let mut file: VirtualFile = todo!("EphemeralFile superclass needs into_inner() api");
// Write out the index
file.seek(SeekFrom::Start(index_start_blk as u64 * PAGE_SZ as u64))?;
@@ -725,7 +709,7 @@ impl ImageLayerWriter {
impl Drop for ImageLayerWriter {
fn drop(&mut self) {
if let Some(inner) = self.inner.take() {
inner.blob_writer.into_inner().remove();
todo!("TBD EpheemralFile superclass into_inner(); => VirtualFile => remove()");
}
}
}