src/backup/dynamic_index.rs

   1 use std::fs::File;
   2 use std::io::{self, BufWriter, Seek, SeekFrom, Write};
   3 use std::ops::Range;
   4 use std::os::unix::io::AsRawFd;
   5 use std::path::{Path, PathBuf};
   6 use std::sync::{Arc, Mutex};
   7 use std::task::Context;
   8 use std::pin::Pin;
   9
  10 use anyhow::{bail, format_err, Error};
  11
  12 use proxmox::tools::io::ReadExt;
  13 use proxmox::tools::uuid::Uuid;
  14 use proxmox::tools::mmap::Mmap;
  15 use pxar::accessor::{MaybeReady, ReadAt, ReadAtOperation};
  16
  17 use super::chunk_stat::ChunkStat;
  18 use super::chunk_store::ChunkStore;
  19 use super::index::ChunkReadInfo;
  20 use super::read_chunk::ReadChunk;
  21 use super::Chunker;
  22 use super::IndexFile;
  23 use super::{DataBlob, DataChunkBuilder};
  24 use crate::tools;
  25
  26 /// Header format definition for dynamic index files (`.dixd`)
  27 #[repr(C)]
  28 pub struct DynamicIndexHeader {
  29     pub magic: [u8; 8],
  30     pub uuid: [u8; 16],
  31     pub ctime: i64,
  32     /// Sha256 over the index ``SHA256(offset1||digest1||offset2||digest2||...)``
  33     pub index_csum: [u8; 32],
  34     reserved: [u8; 4032], // overall size is one page (4096 bytes)
  35 }
  36 proxmox::static_assert_size!(DynamicIndexHeader, 4096);
  37 // TODO: Once non-Copy unions are stabilized, use:
  38 // union DynamicIndexHeader {
  39 //     reserved: [u8; 4096],
  40 //     pub data: DynamicIndexHeaderData,
  41 // }
  42
  43 impl DynamicIndexHeader {
  44     /// Convenience method to allocate a zero-initialized header struct.
  45     pub fn zeroed() -> Box<Self> {
  46         unsafe {
  47             Box::from_raw(std::alloc::alloc_zeroed(std::alloc::Layout::new::<Self>()) as *mut Self)
  48         }
  49     }
  50
  51     pub fn as_bytes(&self) -> &[u8] {
  52         unsafe {
  53             std::slice::from_raw_parts(
  54                 self as *const Self as *const u8,
  55                 std::mem::size_of::<Self>(),
  56             )
  57         }
  58     }
  59 }
  60
  61 #[derive(Clone, Debug)]
  62 #[repr(C)]
  63 pub struct DynamicEntry {
  64     end_le: u64,
  65     digest: [u8; 32],
  66 }
  67
  68 impl DynamicEntry {
  69     #[inline]
  70     pub fn end(&self) -> u64 {
  71         u64::from_le(self.end_le)
  72     }
  73 }
  74
  75 pub struct DynamicIndexReader {
  76     _file: File,
  77     pub size: usize,
  78     index: Mmap<DynamicEntry>,
  79     pub uuid: [u8; 16],
  80     pub ctime: i64,
  81     pub index_csum: [u8; 32],
  82 }
  83
  84 impl DynamicIndexReader {
  85     pub fn open(path: &Path) -> Result<Self, Error> {
  86         File::open(path)
  87             .map_err(Error::from)
  88             .and_then(Self::new)
  89             .map_err(|err| format_err!("Unable to open dynamic index {:?} - {}", path, err))
  90     }
  91
  92     pub fn new(mut file: std::fs::File) -> Result<Self, Error> {
  93         // FIXME: This is NOT OUR job! Check the callers of this method and remove this!
  94         file.seek(SeekFrom::Start(0))?;
  95
  96         let header_size = std::mem::size_of::<DynamicIndexHeader>();
  97
  98         let rawfd = file.as_raw_fd();
  99         let stat = match nix::sys::stat::fstat(rawfd) {
 100             Ok(stat) => stat,
 101             Err(err) => bail!("fstat failed - {}", err),
 102         };
 103
 104         let size = stat.st_size as usize;
 105
 106         if size < header_size {
 107             bail!("index too small ({})", stat.st_size);
 108         }
 109
 110         let header: Box<DynamicIndexHeader> = unsafe { file.read_host_value_boxed()? };
 111
 112         if header.magic != super::DYNAMIC_SIZED_CHUNK_INDEX_1_0 {
 113             bail!("got unknown magic number");
 114         }
 115
 116         let ctime = proxmox::tools::time::epoch_i64();
 117
 118         let index_size = stat.st_size as usize - header_size;
 119         let index_count = index_size / 40;
 120         if index_count * 40 != index_size {
 121             bail!("got unexpected file size");
 122         }
 123
 124         let index = unsafe {
 125             Mmap::map_fd(
 126                 rawfd,
 127                 header_size as u64,
 128                 index_count,
 129                 nix::sys::mman::ProtFlags::PROT_READ,
 130                 nix::sys::mman::MapFlags::MAP_PRIVATE,
 131             )?
 132         };
 133
 134         Ok(Self {
 135             _file: file,
 136             size,
 137             index,
 138             ctime,
 139             uuid: header.uuid,
 140             index_csum: header.index_csum,
 141         })
 142     }
 143
 144     #[inline]
 145     #[allow(clippy::cast_ptr_alignment)]
 146     fn chunk_end(&self, pos: usize) -> u64 {
 147         if pos >= self.index.len() {
 148             panic!("chunk index out of range");
 149         }
 150         self.index[pos].end()
 151     }
 152
 153     #[inline]
 154     fn chunk_digest(&self, pos: usize) -> &[u8; 32] {
 155         if pos >= self.index.len() {
 156             panic!("chunk index out of range");
 157         }
 158         &self.index[pos].digest
 159     }
 160
 161     // TODO: can we use std::slice::binary_search with Mmap now?
 162     fn binary_search(
 163         &self,
 164         start_idx: usize,
 165         start: u64,
 166         end_idx: usize,
 167         end: u64,
 168         offset: u64,
 169     ) -> Result<usize, Error> {
 170         if (offset >= end) || (offset < start) {
 171             bail!("offset out of range");
 172         }
 173
 174         if end_idx == start_idx {
 175             return Ok(start_idx); // found
 176         }
 177         let middle_idx = (start_idx + end_idx) / 2;
 178         let middle_end = self.chunk_end(middle_idx);
 179
 180         if offset < middle_end {
 181             self.binary_search(start_idx, start, middle_idx, middle_end, offset)
 182         } else {
 183             self.binary_search(middle_idx + 1, middle_end, end_idx, end, offset)
 184         }
 185     }
 186 }
 187
 188 impl IndexFile for DynamicIndexReader {
 189     fn index_count(&self) -> usize {
 190         self.index.len()
 191     }
 192
 193     fn index_digest(&self, pos: usize) -> Option<&[u8; 32]> {
 194         if pos >= self.index.len() {
 195             None
 196         } else {
 197             Some(unsafe { std::mem::transmute(self.chunk_digest(pos).as_ptr()) })
 198         }
 199     }
 200
 201     fn index_bytes(&self) -> u64 {
 202         if self.index.is_empty() {
 203             0
 204         } else {
 205             self.chunk_end(self.index.len() - 1)
 206         }
 207     }
 208
 209     fn compute_csum(&self) -> ([u8; 32], u64) {
 210         let mut csum = openssl::sha::Sha256::new();
 211         let mut chunk_end = 0;
 212         for pos in 0..self.index_count() {
 213             let info = self.chunk_info(pos).unwrap();
 214             chunk_end = info.range.end;
 215             csum.update(&chunk_end.to_le_bytes());
 216             csum.update(&info.digest);
 217         }
 218         let csum = csum.finish();
 219         (csum, chunk_end)
 220     }
 221
 222     fn chunk_info(&self, pos: usize) -> Option<ChunkReadInfo> {
 223         if pos >= self.index.len() {
 224             return None;
 225         }
 226         let start = if pos == 0 { 0 } else { self.index[pos - 1].end() };
 227
 228         let end = self.index[pos].end();
 229
 230         Some(ChunkReadInfo {
 231             range: start..end,
 232             digest: self.index[pos].digest,
 233         })
 234     }
 235
 236     fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
 237         let end_idx = self.index.len() - 1;
 238         let end = self.chunk_end(end_idx);
 239         let found_idx = self.binary_search(0, 0, end_idx, end, offset);
 240         let found_idx = match found_idx {
 241             Ok(i) => i,
 242             Err(_) => return None
 243         };
 244
 245         let found_start = if found_idx == 0 {
 246             0
 247         } else {
 248             self.chunk_end(found_idx - 1)
 249         };
 250
 251         Some((found_idx, offset - found_start))
 252     }
 253 }
 254
 255 struct CachedChunk {
 256     range: Range<u64>,
 257     data: Vec<u8>,
 258 }
 259
 260 impl CachedChunk {
 261     /// Perform sanity checks on the range and data size:
 262     pub fn new(range: Range<u64>, data: Vec<u8>) -> Result<Self, Error> {
 263         if data.len() as u64 != range.end - range.start {
 264             bail!(
 265                 "read chunk with wrong size ({} != {})",
 266                 data.len(),
 267                 range.end - range.start,
 268             );
 269         }
 270         Ok(Self { range, data })
 271     }
 272 }
 273
 274 pub struct BufferedDynamicReader<S> {
 275     store: S,
 276     index: DynamicIndexReader,
 277     archive_size: u64,
 278     read_buffer: Vec<u8>,
 279     buffered_chunk_idx: usize,
 280     buffered_chunk_start: u64,
 281     read_offset: u64,
 282     lru_cache: crate::tools::lru_cache::LruCache<usize, CachedChunk>,
 283 }
 284
 285 struct ChunkCacher<'a, S> {
 286     store: &'a mut S,
 287     index: &'a DynamicIndexReader,
 288 }
 289
 290 impl<'a, S: ReadChunk> crate::tools::lru_cache::Cacher<usize, CachedChunk> for ChunkCacher<'a, S> {
 291     fn fetch(&mut self, index: usize) -> Result<Option<CachedChunk>, Error> {
 292         let info = match self.index.chunk_info(index) {
 293             Some(info) => info,
 294             None => bail!("chunk index out of range"),
 295         };
 296         let range = info.range;
 297         let data = self.store.read_chunk(&info.digest)?;
 298         CachedChunk::new(range, data).map(Some)
 299     }
 300 }
 301
 302 impl<S: ReadChunk> BufferedDynamicReader<S> {
 303     pub fn new(index: DynamicIndexReader, store: S) -> Self {
 304         let archive_size = index.index_bytes();
 305         Self {
 306             store,
 307             index,
 308             archive_size,
 309             read_buffer: Vec::with_capacity(1024 * 1024),
 310             buffered_chunk_idx: 0,
 311             buffered_chunk_start: 0,
 312             read_offset: 0,
 313             lru_cache: crate::tools::lru_cache::LruCache::new(32),
 314         }
 315     }
 316
 317     pub fn archive_size(&self) -> u64 {
 318         self.archive_size
 319     }
 320
 321     fn buffer_chunk(&mut self, idx: usize) -> Result<(), Error> {
 322         //let (start, end, data) = self.lru_cache.access(
 323         let cached_chunk = self.lru_cache.access(
 324             idx,
 325             &mut ChunkCacher {
 326                 store: &mut self.store,
 327                 index: &self.index,
 328             },
 329         )?.ok_or_else(|| format_err!("chunk not found by cacher"))?;
 330
 331         // fixme: avoid copy
 332         self.read_buffer.clear();
 333         self.read_buffer.extend_from_slice(&cached_chunk.data);
 334
 335         self.buffered_chunk_idx = idx;
 336
 337         self.buffered_chunk_start = cached_chunk.range.start;
 338         //println!("BUFFER {} {}",  self.buffered_chunk_start, end);
 339         Ok(())
 340     }
 341 }
 342
 343 impl<S: ReadChunk> crate::tools::BufferedRead for BufferedDynamicReader<S> {
 344     fn buffered_read(&mut self, offset: u64) -> Result<&[u8], Error> {
 345         if offset == self.archive_size {
 346             return Ok(&self.read_buffer[0..0]);
 347         }
 348
 349         let buffer_len = self.read_buffer.len();
 350         let index = &self.index;
 351
 352         // optimization for sequential read
 353         if buffer_len > 0
 354             && ((self.buffered_chunk_idx + 1) < index.index.len())
 355             && (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
 356         {
 357             let next_idx = self.buffered_chunk_idx + 1;
 358             let next_end = index.chunk_end(next_idx);
 359             if offset < next_end {
 360                 self.buffer_chunk(next_idx)?;
 361                 let buffer_offset = (offset - self.buffered_chunk_start) as usize;
 362                 return Ok(&self.read_buffer[buffer_offset..]);
 363             }
 364         }
 365
 366         if (buffer_len == 0)
 367             || (offset < self.buffered_chunk_start)
 368             || (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
 369         {
 370             let end_idx = index.index.len() - 1;
 371             let end = index.chunk_end(end_idx);
 372             let idx = index.binary_search(0, 0, end_idx, end, offset)?;
 373             self.buffer_chunk(idx)?;
 374         }
 375
 376         let buffer_offset = (offset - self.buffered_chunk_start) as usize;
 377         Ok(&self.read_buffer[buffer_offset..])
 378     }
 379 }
 380
 381 impl<S: ReadChunk> std::io::Read for BufferedDynamicReader<S> {
 382     fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
 383         use crate::tools::BufferedRead;
 384         use std::io::{Error, ErrorKind};
 385
 386         let data = match self.buffered_read(self.read_offset) {
 387             Ok(v) => v,
 388             Err(err) => return Err(Error::new(ErrorKind::Other, err.to_string())),
 389         };
 390
 391         let n = if data.len() > buf.len() {
 392             buf.len()
 393         } else {
 394             data.len()
 395         };
 396
 397         buf[0..n].copy_from_slice(&data[0..n]);
 398
 399         self.read_offset += n as u64;
 400
 401         Ok(n)
 402     }
 403 }
 404
 405 impl<S: ReadChunk> std::io::Seek for BufferedDynamicReader<S> {
 406     fn seek(&mut self, pos: SeekFrom) -> Result<u64, std::io::Error> {
 407         let new_offset = match pos {
 408             SeekFrom::Start(start_offset) => start_offset as i64,
 409             SeekFrom::End(end_offset) => (self.archive_size as i64) + end_offset,
 410             SeekFrom::Current(offset) => (self.read_offset as i64) + offset,
 411         };
 412
 413         use std::io::{Error, ErrorKind};
 414         if (new_offset < 0) || (new_offset > (self.archive_size as i64)) {
 415             return Err(Error::new(
 416                 ErrorKind::Other,
 417                 format!(
 418                     "seek is out of range {} ([0..{}])",
 419                     new_offset, self.archive_size
 420                 ),
 421             ));
 422         }
 423         self.read_offset = new_offset as u64;
 424
 425         Ok(self.read_offset)
 426     }
 427 }
 428
 429 /// This is a workaround until we have cleaned up the chunk/reader/... infrastructure for better
 430 /// async use!
 431 ///
 432 /// Ideally BufferedDynamicReader gets replaced so the LruCache maps to `BroadcastFuture<Chunk>`,
 433 /// so that we can properly access it from multiple threads simultaneously while not issuing
 434 /// duplicate simultaneous reads over http.
 435 #[derive(Clone)]
 436 pub struct LocalDynamicReadAt<R: ReadChunk> {
 437     inner: Arc<Mutex<BufferedDynamicReader<R>>>,
 438 }
 439
 440 impl<R: ReadChunk> LocalDynamicReadAt<R> {
 441     pub fn new(inner: BufferedDynamicReader<R>) -> Self {
 442         Self {
 443             inner: Arc::new(Mutex::new(inner)),
 444         }
 445     }
 446 }
 447
 448 impl<R: ReadChunk> ReadAt for LocalDynamicReadAt<R> {
 449     fn start_read_at<'a>(
 450         self: Pin<&'a Self>,
 451         _cx: &mut Context,
 452         buf: &'a mut [u8],
 453         offset: u64,
 454     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
 455         use std::io::Read;
 456         MaybeReady::Ready(tokio::task::block_in_place(move || {
 457             let mut reader = self.inner.lock().unwrap();
 458             reader.seek(SeekFrom::Start(offset))?;
 459             Ok(reader.read(buf)?)
 460         }))
 461     }
 462
 463     fn poll_complete<'a>(
 464         self: Pin<&'a Self>,
 465         _op: ReadAtOperation<'a>,
 466     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
 467         panic!("LocalDynamicReadAt::start_read_at returned Pending");
 468     }
 469 }
 470
 471
 472 /// Create dynamic index files (`.dixd`)
 473 pub struct DynamicIndexWriter {
 474     store: Arc<ChunkStore>,
 475     _lock: tools::ProcessLockSharedGuard,
 476     writer: BufWriter<File>,
 477     closed: bool,
 478     filename: PathBuf,
 479     tmp_filename: PathBuf,
 480     csum: Option<openssl::sha::Sha256>,
 481     pub uuid: [u8; 16],
 482     pub ctime: i64,
 483 }
 484
 485 impl Drop for DynamicIndexWriter {
 486     fn drop(&mut self) {
 487         let _ = std::fs::remove_file(&self.tmp_filename); // ignore errors
 488     }
 489 }
 490
 491 impl DynamicIndexWriter {
 492     pub fn create(store: Arc<ChunkStore>, path: &Path) -> Result<Self, Error> {
 493         let shared_lock = store.try_shared_lock()?;
 494
 495         let full_path = store.relative_path(path);
 496         let mut tmp_path = full_path.clone();
 497         tmp_path.set_extension("tmp_didx");
 498
 499         let file = std::fs::OpenOptions::new()
 500             .create(true)
 501             .truncate(true)
 502             .read(true)
 503             .write(true)
 504             .open(&tmp_path)?;
 505
 506         let mut writer = BufWriter::with_capacity(1024 * 1024, file);
 507
 508         let ctime = proxmox::tools::time::epoch_i64();
 509
 510         let uuid = Uuid::generate();
 511
 512         let mut header = DynamicIndexHeader::zeroed();
 513         header.magic = super::DYNAMIC_SIZED_CHUNK_INDEX_1_0;
 514         header.ctime = i64::to_le(ctime);
 515         header.uuid = *uuid.as_bytes();
 516         // header.index_csum = [0u8; 32];
 517         writer.write_all(header.as_bytes())?;
 518
 519         let csum = Some(openssl::sha::Sha256::new());
 520
 521         Ok(Self {
 522             store,
 523             _lock: shared_lock,
 524             writer,
 525             closed: false,
 526             filename: full_path,
 527             tmp_filename: tmp_path,
 528             ctime,
 529             uuid: *uuid.as_bytes(),
 530             csum,
 531         })
 532     }
 533
 534     // fixme: use add_chunk instead?
 535     pub fn insert_chunk(&self, chunk: &DataBlob, digest: &[u8; 32]) -> Result<(bool, u64), Error> {
 536         self.store.insert_chunk(chunk, digest)
 537     }
 538
 539     pub fn close(&mut self) -> Result<[u8; 32], Error> {
 540         if self.closed {
 541             bail!(
 542                 "cannot close already closed archive index file {:?}",
 543                 self.filename
 544             );
 545         }
 546
 547         self.closed = true;
 548
 549         self.writer.flush()?;
 550
 551         let csum_offset = proxmox::offsetof!(DynamicIndexHeader, index_csum);
 552         self.writer.seek(SeekFrom::Start(csum_offset as u64))?;
 553
 554         let csum = self.csum.take().unwrap();
 555         let index_csum = csum.finish();
 556
 557         self.writer.write_all(&index_csum)?;
 558         self.writer.flush()?;
 559
 560         if let Err(err) = std::fs::rename(&self.tmp_filename, &self.filename) {
 561             bail!("Atomic rename file {:?} failed - {}", self.filename, err);
 562         }
 563
 564         Ok(index_csum)
 565     }
 566
 567     // fixme: rename to add_digest
 568     pub fn add_chunk(&mut self, offset: u64, digest: &[u8; 32]) -> Result<(), Error> {
 569         if self.closed {
 570             bail!(
 571                 "cannot write to closed dynamic index file {:?}",
 572                 self.filename
 573             );
 574         }
 575
 576         let offset_le: &[u8; 8] = unsafe { &std::mem::transmute::<u64, [u8; 8]>(offset.to_le()) };
 577
 578         if let Some(ref mut csum) = self.csum {
 579             csum.update(offset_le);
 580             csum.update(digest);
 581         }
 582
 583         self.writer.write_all(offset_le)?;
 584         self.writer.write_all(digest)?;
 585         Ok(())
 586     }
 587 }
 588
 589 /// Writer which splits a binary stream into dynamic sized chunks
 590 ///
 591 /// And store the resulting chunk list into the index file.
 592 pub struct DynamicChunkWriter {
 593     index: DynamicIndexWriter,
 594     closed: bool,
 595     chunker: Chunker,
 596     stat: ChunkStat,
 597     chunk_offset: usize,
 598     last_chunk: usize,
 599     chunk_buffer: Vec<u8>,
 600 }
 601
 602 impl DynamicChunkWriter {
 603     pub fn new(index: DynamicIndexWriter, chunk_size: usize) -> Self {
 604         Self {
 605             index,
 606             closed: false,
 607             chunker: Chunker::new(chunk_size),
 608             stat: ChunkStat::new(0),
 609             chunk_offset: 0,
 610             last_chunk: 0,
 611             chunk_buffer: Vec::with_capacity(chunk_size * 4),
 612         }
 613     }
 614
 615     pub fn stat(&self) -> &ChunkStat {
 616         &self.stat
 617     }
 618
 619     pub fn close(&mut self) -> Result<(), Error> {
 620         if self.closed {
 621             return Ok(());
 622         }
 623
 624         self.closed = true;
 625
 626         self.write_chunk_buffer()?;
 627
 628         self.index.close()?;
 629
 630         self.stat.size = self.chunk_offset as u64;
 631
 632         // add size of index file
 633         self.stat.size +=
 634             (self.stat.chunk_count * 40 + std::mem::size_of::<DynamicIndexHeader>()) as u64;
 635
 636         Ok(())
 637     }
 638
 639     fn write_chunk_buffer(&mut self) -> Result<(), Error> {
 640         let chunk_size = self.chunk_buffer.len();
 641
 642         if chunk_size == 0 {
 643             return Ok(());
 644         }
 645
 646         let expected_chunk_size = self.chunk_offset - self.last_chunk;
 647         if expected_chunk_size != self.chunk_buffer.len() {
 648             bail!("wrong chunk size {} != {}", expected_chunk_size, chunk_size);
 649         }
 650
 651         self.stat.chunk_count += 1;
 652
 653         self.last_chunk = self.chunk_offset;
 654
 655         let (chunk, digest) = DataChunkBuilder::new(&self.chunk_buffer)
 656             .compress(true)
 657             .build()?;
 658
 659         match self.index.insert_chunk(&chunk, &digest) {
 660             Ok((is_duplicate, compressed_size)) => {
 661                 self.stat.compressed_size += compressed_size;
 662                 if is_duplicate {
 663                     self.stat.duplicate_chunks += 1;
 664                 } else {
 665                     self.stat.disk_size += compressed_size;
 666                 }
 667
 668                 println!(
 669                     "ADD CHUNK {:016x} {} {}% {} {}",
 670                     self.chunk_offset,
 671                     chunk_size,
 672                     (compressed_size * 100) / (chunk_size as u64),
 673                     is_duplicate,
 674                     proxmox::tools::digest_to_hex(&digest)
 675                 );
 676                 self.index.add_chunk(self.chunk_offset as u64, &digest)?;
 677                 self.chunk_buffer.truncate(0);
 678                 Ok(())
 679             }
 680             Err(err) => {
 681                 self.chunk_buffer.truncate(0);
 682                 Err(err)
 683             }
 684         }
 685     }
 686 }
 687
 688 impl Write for DynamicChunkWriter {
 689     fn write(&mut self, data: &[u8]) -> std::result::Result<usize, std::io::Error> {
 690         let chunker = &mut self.chunker;
 691
 692         let pos = chunker.scan(data);
 693
 694         if pos > 0 {
 695             self.chunk_buffer.extend_from_slice(&data[0..pos]);
 696             self.chunk_offset += pos;
 697
 698             if let Err(err) = self.write_chunk_buffer() {
 699                 return Err(std::io::Error::new(
 700                     std::io::ErrorKind::Other,
 701                     err.to_string(),
 702                 ));
 703             }
 704             Ok(pos)
 705         } else {
 706             self.chunk_offset += data.len();
 707             self.chunk_buffer.extend_from_slice(data);
 708             Ok(data.len())
 709         }
 710     }
 711
 712     fn flush(&mut self) -> std::result::Result<(), std::io::Error> {
 713         Err(std::io::Error::new(
 714             std::io::ErrorKind::Other,
 715             "please use close() instead of flush()",
 716         ))
 717     }
 718 }