tyler/proxmox-backup
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

split tape code into new pbs_tape workspace

Browse Source
This commit is contained in:
Dietmar Maurer
2021-09-13 11:54:24 +02:00
parent bfd2b47649
commit 048b43af24
48 changed files with 672 additions and 698 deletions
Download Patch File Download Diff File Expand all files Collapse all files

373
src/tape/file_formats/blocked_reader.rs
View File

@ -1,373 +0,0 @@
use std::io::Read;
use crate::tape::{
TapeRead,
BlockRead,
BlockReadError,
file_formats::{
PROXMOX_TAPE_BLOCK_HEADER_MAGIC_1_0,
BlockHeader,
BlockHeaderFlags,
},
};
/// Read a block stream generated by 'BlockWriter'.
///
/// This class implements 'TapeRead'. It always read whole blocks from
/// the underlying reader, and does additional error checks:
///
/// - check magic number (detect streams not written by 'BlockWriter')
/// - check block size
/// - check block sequence numbers
///
/// The reader consumes the EOF mark after the data stream (if read to
/// the end of the stream).
pub struct BlockedReader<R> {
reader: R,
buffer: Box<BlockHeader>,
seq_nr: u32,
found_end_marker: bool,
incomplete: bool,
got_eod: bool,
read_error: bool,
read_pos: usize,
}
impl <R: BlockRead> BlockedReader<R> {
/// Create a new BlockedReader instance.
///
/// This tries to read the first block. Please inspect the error
/// to detect EOF and EOT.
pub fn open(mut reader: R) -> Result<Self, BlockReadError> {
let mut buffer = BlockHeader::new();
Self::read_block_frame(&mut buffer, &mut reader)?;
let (_size, found_end_marker) = Self::check_buffer(&buffer, 0)?;
let mut incomplete = false;
let mut got_eod = false;
if found_end_marker {
incomplete = buffer.flags.contains(BlockHeaderFlags::INCOMPLETE);
Self::consume_eof_marker(&mut reader)?;
got_eod = true;
}
Ok(Self {
reader,
buffer,
found_end_marker,
incomplete,
got_eod,
seq_nr: 1,
read_error: false,
read_pos: 0,
})
}
fn check_buffer(buffer: &BlockHeader, seq_nr: u32) -> Result<(usize, bool), std::io::Error> {
if buffer.magic != PROXMOX_TAPE_BLOCK_HEADER_MAGIC_1_0 {
proxmox::io_bail!("detected tape block with wrong magic number - not written by proxmox tape");
}
if seq_nr != buffer.seq_nr() {
proxmox::io_bail!(
"detected tape block with wrong sequence number ({} != {})",
seq_nr, buffer.seq_nr())
}
let size = buffer.size();
let found_end_marker = buffer.flags.contains(BlockHeaderFlags::END_OF_STREAM);
if size > buffer.payload.len() {
proxmox::io_bail!("detected tape block with wrong payload size ({} > {}", size, buffer.payload.len());
} else if size == 0 && !found_end_marker {
proxmox::io_bail!("detected tape block with zero payload size");
}
Ok((size, found_end_marker))
}
fn read_block_frame(buffer: &mut BlockHeader, reader: &mut R) -> Result<(), BlockReadError> {
let data = unsafe {
std::slice::from_raw_parts_mut(
(buffer as *mut BlockHeader) as *mut u8,
BlockHeader::SIZE,
)
};
let bytes = reader.read_block(data)?;
if bytes != BlockHeader::SIZE {
return Err(proxmox::io_format_err!("got wrong block size").into());
}
Ok(())
}
fn consume_eof_marker(reader: &mut R) -> Result<(), std::io::Error> {
let mut tmp_buf = [0u8; 512]; // use a small buffer for testing EOF
match reader.read_block(&mut tmp_buf) {
Ok(_) => {
proxmox::io_bail!("detected tape block after block-stream end marker");
}
Err(BlockReadError::EndOfFile) => {
return Ok(());
}
Err(BlockReadError::EndOfStream) => {
proxmox::io_bail!("got unexpected end of tape");
}
Err(BlockReadError::Error(err)) => {
return Err(err);
}
}
}
fn read_block(&mut self, check_end_marker: bool) -> Result<usize, std::io::Error> {
match Self::read_block_frame(&mut self.buffer, &mut self.reader) {
Ok(()) => { /* ok */ }
Err(BlockReadError::EndOfFile) => {
self.got_eod = true;
self.read_pos = self.buffer.payload.len();
if !self.found_end_marker && check_end_marker {
proxmox::io_bail!("detected tape stream without end marker");
}
return Ok(0); // EOD
}
Err(BlockReadError::EndOfStream) => {
proxmox::io_bail!("got unexpected end of tape");
}
Err(BlockReadError::Error(err)) => {
return Err(err);
}
}
let (size, found_end_marker) = Self::check_buffer(&self.buffer, self.seq_nr)?;
self.seq_nr += 1;
if found_end_marker { // consume EOF mark
self.found_end_marker = true;
self.incomplete = self.buffer.flags.contains(BlockHeaderFlags::INCOMPLETE);
Self::consume_eof_marker(&mut self.reader)?;
self.got_eod = true;
}
self.read_pos = 0;
Ok(size)
}
}
impl <R: BlockRead> TapeRead for BlockedReader<R> {
fn is_incomplete(&self) -> Result<bool, std::io::Error> {
if !self.got_eod {
proxmox::io_bail!("is_incomplete failed: EOD not reached");
}
if !self.found_end_marker {
proxmox::io_bail!("is_incomplete failed: no end marker found");
}
Ok(self.incomplete)
}
fn has_end_marker(&self) -> Result<bool, std::io::Error> {
if !self.got_eod {
proxmox::io_bail!("has_end_marker failed: EOD not reached");
}
Ok(self.found_end_marker)
}
// like ReadExt::skip_to_end(), but does not raise an error if the
// stream has no end marker.
fn skip_data(&mut self) -> Result<usize, std::io::Error> {
let mut bytes = 0;
let buffer_size = self.buffer.size();
let rest = (buffer_size as isize) - (self.read_pos as isize);
if rest > 0 {
bytes = rest as usize;
}
loop {
if self.got_eod {
return Ok(bytes);
}
bytes += self.read_block(false)?;
}
}
}
impl <R: BlockRead> Read for BlockedReader<R> {
fn read(&mut self, buffer: &mut [u8]) -> Result<usize, std::io::Error> {
if self.read_error {
proxmox::io_bail!("detected read after error - internal error");
}
let mut buffer_size = self.buffer.size();
let mut rest = (buffer_size as isize) - (self.read_pos as isize);
if rest <= 0 && !self.got_eod { // try to refill buffer
buffer_size = match self.read_block(true) {
Ok(len) => len,
err => {
self.read_error = true;
return err;
}
};
rest = buffer_size as isize;
}
if rest <= 0 {
Ok(0)
} else {
let copy_len = if (buffer.len() as isize) < rest {
buffer.len()
} else {
rest as usize
};
buffer[..copy_len].copy_from_slice(
&self.buffer.payload[self.read_pos..(self.read_pos + copy_len)]);
self.read_pos += copy_len;
Ok(copy_len)
}
}
}
#[cfg(test)]
mod test {
use std::io::Read;
use anyhow::{bail, Error};
use crate::tape::{
TapeWrite,
BlockReadError,
helpers::{EmulateTapeReader, EmulateTapeWriter},
file_formats::{
PROXMOX_TAPE_BLOCK_SIZE,
BlockedReader,
BlockedWriter,
},
};
fn write_and_verify(data: &[u8]) -> Result<(), Error> {
let mut tape_data = Vec::new();
{
let writer = EmulateTapeWriter::new(&mut tape_data, 1024*1024*10);
let mut writer = BlockedWriter::new(writer);
writer.write_all(data)?;
writer.finish(false)?;
}
assert_eq!(
tape_data.len(),
((data.len() + PROXMOX_TAPE_BLOCK_SIZE)/PROXMOX_TAPE_BLOCK_SIZE)
*PROXMOX_TAPE_BLOCK_SIZE
);
let reader = &mut &tape_data[..];
let reader = EmulateTapeReader::new(reader);
let mut reader = BlockedReader::open(reader)?;
let mut read_data = Vec::with_capacity(PROXMOX_TAPE_BLOCK_SIZE);
reader.read_to_end(&mut read_data)?;
assert_eq!(data.len(), read_data.len());
assert_eq!(data, &read_data[..]);
Ok(())
}
#[test]
fn empty_stream() -> Result<(), Error> {
write_and_verify(b"")
}
#[test]
fn small_data() -> Result<(), Error> {
write_and_verify(b"ABC")
}
#[test]
fn large_data() -> Result<(), Error> {
let data = proxmox::sys::linux::random_data(1024*1024*5)?;
write_and_verify(&data)
}
#[test]
fn no_data() -> Result<(), Error> {
let tape_data = Vec::new();
let reader = &mut &tape_data[..];
let reader = EmulateTapeReader::new(reader);
match BlockedReader::open(reader) {
Err(BlockReadError::EndOfFile) => { /* OK */ },
_ => bail!("expected EOF"),
}
Ok(())
}
#[test]
fn no_end_marker() -> Result<(), Error> {
let mut tape_data = Vec::new();
{
let writer = EmulateTapeWriter::new(&mut tape_data, 1024*1024);
let mut writer = BlockedWriter::new(writer);
// write at least one block
let data = proxmox::sys::linux::random_data(PROXMOX_TAPE_BLOCK_SIZE)?;
writer.write_all(&data)?;
// but do not call finish here
}
let reader = &mut &tape_data[..];
let reader = EmulateTapeReader::new(reader);
let mut reader = BlockedReader::open(reader)?;
let mut data = Vec::with_capacity(PROXMOX_TAPE_BLOCK_SIZE);
assert!(reader.read_to_end(&mut data).is_err());
Ok(())
}
#[test]
fn small_read_buffer() -> Result<(), Error> {
let mut tape_data = Vec::new();
{
let writer = EmulateTapeWriter::new(&mut tape_data, 1024*1024);
let mut writer = BlockedWriter::new(writer);
writer.write_all(b"ABC")?;
writer.finish(false)?;
}
let reader = &mut &tape_data[..];
let reader = EmulateTapeReader::new(reader);
let mut reader = BlockedReader::open(reader)?;
let mut buf = [0u8; 1];
assert_eq!(reader.read(&mut buf)?, 1, "wrong byte count");
assert_eq!(&buf, b"A");
assert_eq!(reader.read(&mut buf)?, 1, "wrong byte count");
assert_eq!(&buf, b"B");
assert_eq!(reader.read(&mut buf)?, 1, "wrong byte count");
assert_eq!(&buf, b"C");
assert_eq!(reader.read(&mut buf)?, 0, "wrong byte count");
assert_eq!(reader.read(&mut buf)?, 0, "wrong byte count");
Ok(())
}
}

145
src/tape/file_formats/blocked_writer.rs
View File

@ -1,145 +0,0 @@
use proxmox::tools::vec;
use crate::tape::{
TapeWrite,
BlockWrite,
file_formats::{
BlockHeader,
BlockHeaderFlags,
},
};
/// Assemble and write blocks of data
///
/// This type implement 'TapeWrite'. Data written is assembled to
/// equally sized blocks (see 'BlockHeader'), which are then written
/// to the underlying writer.
pub struct BlockedWriter<W: BlockWrite> {
writer: W,
buffer: Box<BlockHeader>,
buffer_pos: usize,
seq_nr: u32,
logical_end_of_media: bool,
bytes_written: usize,
wrote_eof: bool,
}
impl <W: BlockWrite> Drop for BlockedWriter<W> {
// Try to make sure to end the file with a filemark
fn drop(&mut self) {
if !self.wrote_eof {
let _ = self.writer.write_filemark();
}
}
}
impl <W: BlockWrite> BlockedWriter<W> {
/// Allow access to underlying writer
pub fn writer_ref_mut(&mut self) -> &mut W {
&mut self.writer
}
/// Creates a new instance.
pub fn new(writer: W) -> Self {
Self {
writer,
buffer: BlockHeader::new(),
buffer_pos: 0,
seq_nr: 0,
logical_end_of_media: false,
bytes_written: 0,
wrote_eof: false,
}
}
fn write_block(buffer: &BlockHeader, writer: &mut W) -> Result<bool, std::io::Error> {
let data = unsafe {
std::slice::from_raw_parts(
(buffer as *const BlockHeader) as *const u8,
BlockHeader::SIZE,
)
};
writer.write_block(data)
}
fn write_eof(&mut self) -> Result<(), std::io::Error> {
if self.wrote_eof {
proxmox::io_bail!("BlockedWriter: detected multiple EOF writes");
}
self.wrote_eof = true;
self.writer.write_filemark()
}
fn write(&mut self, data: &[u8]) -> Result<usize, std::io::Error> {
if data.is_empty() { return Ok(0); }
let rest = self.buffer.payload.len() - self.buffer_pos;
let bytes = if data.len() < rest { data.len() } else { rest };
self.buffer.payload[self.buffer_pos..(self.buffer_pos+bytes)]
.copy_from_slice(&data[..bytes]);
let rest = rest - bytes;
if rest == 0 {
self.buffer.flags = BlockHeaderFlags::empty();
self.buffer.set_size(self.buffer.payload.len());
self.buffer.set_seq_nr(self.seq_nr);
self.seq_nr += 1;
let leom = Self::write_block(&self.buffer, &mut self.writer)?;
if leom { self.logical_end_of_media = true; }
self.buffer_pos = 0;
self.bytes_written += BlockHeader::SIZE;
} else {
self.buffer_pos += bytes;
}
Ok(bytes)
}
}
impl <W: BlockWrite> TapeWrite for BlockedWriter<W> {
fn write_all(&mut self, mut data: &[u8]) -> Result<bool, std::io::Error> {
while !data.is_empty() {
match self.write(data) {
Ok(n) => data = &data[n..],
Err(e) => return Err(e),
}
}
Ok(self.logical_end_of_media)
}
fn bytes_written(&self) -> usize {
self.bytes_written
}
/// flush last block, set END_OF_STREAM flag
///
/// Note: This may write an empty block just including the
/// END_OF_STREAM flag.
fn finish(&mut self, incomplete: bool) -> Result<bool, std::io::Error> {
vec::clear(&mut self.buffer.payload[self.buffer_pos..]);
self.buffer.flags = BlockHeaderFlags::END_OF_STREAM;
if incomplete { self.buffer.flags |= BlockHeaderFlags::INCOMPLETE; }
self.buffer.set_size(self.buffer_pos);
self.buffer.set_seq_nr(self.seq_nr);
self.seq_nr += 1;
self.bytes_written += BlockHeader::SIZE;
let leom = Self::write_block(&self.buffer, &mut self.writer)?;
self.write_eof()?;
Ok(leom)
}
/// Returns if the writer already detected the logical end of media
fn logical_end_of_media(&self) -> bool {
self.logical_end_of_media
}
}

8
src/tape/file_formats/catalog_archive.rs
View File

@ -6,13 +6,15 @@ use proxmox::{
tools::Uuid,
};
use pbs_tape::{
PROXMOX_TAPE_BLOCK_SIZE,
TapeWrite, MediaContentHeader,
};
use crate::{
tape::{
TapeWrite,
file_formats::{
PROXMOX_TAPE_BLOCK_SIZE,
PROXMOX_BACKUP_CATALOG_ARCHIVE_MAGIC_1_0,
MediaContentHeader,
CatalogArchiveHeader,
},
},

19
src/tape/file_formats/chunk_archive.rs
View File

@ -9,17 +9,16 @@ use proxmox::tools::{
};
use pbs_datastore::DataBlob;
use pbs_tape::{
PROXMOX_TAPE_BLOCK_SIZE,
TapeWrite, MediaContentHeader,
};
use crate::tape::{
TapeWrite,
file_formats::{
PROXMOX_TAPE_BLOCK_SIZE,
PROXMOX_BACKUP_CHUNK_ARCHIVE_MAGIC_1_1,
PROXMOX_BACKUP_CHUNK_ARCHIVE_ENTRY_MAGIC_1_0,
MediaContentHeader,
ChunkArchiveHeader,
ChunkArchiveEntryHeader,
},
use crate::tape::file_formats::{
PROXMOX_BACKUP_CHUNK_ARCHIVE_MAGIC_1_1,
PROXMOX_BACKUP_CHUNK_ARCHIVE_ENTRY_MAGIC_1_0,
ChunkArchiveHeader,
ChunkArchiveEntryHeader,
};
/// Writes chunk archives to tape.

165
src/tape/file_formats/mod.rs
View File

@ -3,8 +3,6 @@
use std::collections::HashMap;
use anyhow::{bail, Error};
use bitflags::bitflags;
use endian_trait::Endian;
use serde::{Deserialize, Serialize};
@ -12,12 +10,6 @@ use proxmox::tools::Uuid;
use pbs_api_types::Fingerprint;
mod blocked_reader;
pub use blocked_reader::*;
mod blocked_writer;
pub use blocked_writer::*;
mod chunk_archive;
pub use chunk_archive::*;
@ -33,14 +25,6 @@ pub use multi_volume_writer::*;
mod multi_volume_reader;
pub use multi_volume_reader::*;
/// We use 256KB blocksize (always)
pub const PROXMOX_TAPE_BLOCK_SIZE: usize = 256*1024;
// openssl::sha::sha256(b"Proxmox Tape Block Header v1.0")[0..8]
pub const PROXMOX_TAPE_BLOCK_HEADER_MAGIC_1_0: [u8; 8] = [220, 189, 175, 202, 235, 160, 165, 40];
// openssl::sha::sha256(b"Proxmox Backup Content Header v1.0")[0..8];
pub const PROXMOX_BACKUP_CONTENT_HEADER_MAGIC_1_0: [u8; 8] = [99, 238, 20, 159, 205, 242, 155, 12];
// openssl::sha::sha256(b"Proxmox Backup Tape Label v1.0")[0..8];
pub const PROXMOX_BACKUP_MEDIA_LABEL_MAGIC_1_0: [u8; 8] = [42, 5, 191, 60, 176, 48, 170, 57];
// openssl::sha::sha256(b"Proxmox Backup MediaSet Label v1.0")
@ -84,109 +68,6 @@ pub fn proxmox_tape_magic_to_text(magic: &[u8; 8]) -> Option<String> {
PROXMOX_TAPE_CONTENT_NAME.get(magic).map(|s| String::from(*s))
}
/// Tape Block Header with data payload
///
/// All tape files are written as sequence of blocks.
///
/// Note: this struct is large, never put this on the stack!
/// so we use an unsized type to avoid that.
///
/// Tape data block are always read/written with a fixed size
/// (`PROXMOX_TAPE_BLOCK_SIZE`). But they may contain less data, so the
/// header has an additional size field. For streams of blocks, there
/// is a sequence number (`seq_nr`) which may be use for additional
/// error checking.
#[repr(C,packed)]
pub struct BlockHeader {
/// fixed value `PROXMOX_TAPE_BLOCK_HEADER_MAGIC_1_0`
pub magic: [u8; 8],
pub flags: BlockHeaderFlags,
/// size as 3 bytes unsigned, little endian
pub size: [u8; 3],
/// block sequence number
pub seq_nr: u32,
pub payload: [u8],
}
bitflags! {
/// Header flags (e.g. `END_OF_STREAM` or `INCOMPLETE`)
pub struct BlockHeaderFlags: u8 {
/// Marks the last block in a stream.
const END_OF_STREAM = 0b00000001;
/// Mark multivolume streams (when set in the last block)
const INCOMPLETE = 0b00000010;
}
}
#[derive(Endian, Copy, Clone, Debug)]
#[repr(C,packed)]
/// Media Content Header
///
/// All tape files start with this header. The header may contain some
/// informational data indicated by `size`.
///
/// `| MediaContentHeader | header data (size) | stream data |`
///
/// Note: The stream data following may be of any size.
pub struct MediaContentHeader {
/// fixed value `PROXMOX_BACKUP_CONTENT_HEADER_MAGIC_1_0`
pub magic: [u8; 8],
/// magic number for the content following
pub content_magic: [u8; 8],
/// unique ID to identify this data stream
pub uuid: [u8; 16],
/// stream creation time
pub ctime: i64,
/// Size of header data
pub size: u32,
/// Part number for multipart archives.
pub part_number: u8,
/// Reserved for future use
pub reserved_0: u8,
/// Reserved for future use
pub reserved_1: u8,
/// Reserved for future use
pub reserved_2: u8,
}
impl MediaContentHeader {
/// Create a new instance with autogenerated Uuid
pub fn new(content_magic: [u8; 8], size: u32) -> Self {
let uuid = *proxmox::tools::uuid::Uuid::generate()
.into_inner();
Self {
magic: PROXMOX_BACKUP_CONTENT_HEADER_MAGIC_1_0,
content_magic,
uuid,
ctime: proxmox::tools::time::epoch_i64(),
size,
part_number: 0,
reserved_0: 0,
reserved_1: 0,
reserved_2: 0,
}
}
/// Helper to check magic numbers and size constraints
pub fn check(&self, content_magic: [u8; 8], min_size: u32, max_size: u32) -> Result<(), Error> {
if self.magic != PROXMOX_BACKUP_CONTENT_HEADER_MAGIC_1_0 {
bail!("MediaContentHeader: wrong magic");
}
if self.content_magic != content_magic {
bail!("MediaContentHeader: wrong content magic");
}
if self.size < min_size || self.size > max_size {
bail!("MediaContentHeader: got unexpected size");
}
Ok(())
}
/// Returns the content Uuid
pub fn content_uuid(&self) -> Uuid {
Uuid::from(self.uuid)
}
}
#[derive(Deserialize, Serialize)]
/// Header for chunk archives
@ -280,49 +161,3 @@ impl MediaSetLabel {
}
}
impl BlockHeader {
pub const SIZE: usize = PROXMOX_TAPE_BLOCK_SIZE;
/// Allocates a new instance on the heap
pub fn new() -> Box<Self> {
use std::alloc::{alloc_zeroed, Layout};
// align to PAGESIZE, so that we can use it with SG_IO
let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) } as usize;
let mut buffer = unsafe {
let ptr = alloc_zeroed(
Layout::from_size_align(Self::SIZE, page_size)
.unwrap(),
);
Box::from_raw(
std::slice::from_raw_parts_mut(ptr, Self::SIZE - 16)
as *mut [u8] as *mut Self
)
};
buffer.magic = PROXMOX_TAPE_BLOCK_HEADER_MAGIC_1_0;
buffer
}
/// Set the `size` field
pub fn set_size(&mut self, size: usize) {
let size = size.to_le_bytes();
self.size.copy_from_slice(&size[..3]);
}
/// Returns the `size` field
pub fn size(&self) -> usize {
(self.size[0] as usize) + ((self.size[1] as usize)<<8) + ((self.size[2] as usize)<<16)
}
/// Set the `seq_nr` field
pub fn set_seq_nr(&mut self, seq_nr: u32) {
self.seq_nr = seq_nr.to_le();
}
/// Returns the `seq_nr` field
pub fn seq_nr(&self) -> u32 {
u32::from_le(self.seq_nr)
}
}

5
src/tape/file_formats/multi_volume_reader.rs
View File

@ -4,10 +4,7 @@ use anyhow::{bail, Error};
use proxmox::tools::io::ReadExt;
use crate::tape::{
TapeRead,
file_formats::MediaContentHeader,
};
use pbs_tape::{TapeRead, MediaContentHeader};
/// Read multi volume data streams written by `MultiVolumeWriter`
///

5
src/tape/file_formats/multi_volume_writer.rs
View File

@ -2,10 +2,7 @@ use anyhow::Error;
use proxmox::tools::Uuid;
use crate::tape::{
TapeWrite,
file_formats::MediaContentHeader,
};
use pbs_tape::{TapeWrite, MediaContentHeader};
/// Writes data streams using multiple volumes
///

8
src/tape/file_formats/snapshot_archive.rs
View File

@ -7,13 +7,15 @@ use proxmox::{
tools::Uuid,
};
use pbs_tape::{
PROXMOX_TAPE_BLOCK_SIZE,
TapeWrite, MediaContentHeader,
};
use crate::tape::{
TapeWrite,
SnapshotReader,
file_formats::{
PROXMOX_TAPE_BLOCK_SIZE,
PROXMOX_BACKUP_SNAPSHOT_ARCHIVE_MAGIC_1_1,
MediaContentHeader,
SnapshotArchiveHeader,
},
};