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         if let Err(err) =
  94             nix::fcntl::flock(file.as_raw_fd(), nix::fcntl::FlockArg::LockSharedNonblock)
  95         {
  96             bail!("unable to get shared lock - {}", err);
  97         }
  98
  99         // FIXME: This is NOT OUR job! Check the callers of this method and remove this!
 100         file.seek(SeekFrom::Start(0))?;
 101
 102         let header_size = std::mem::size_of::<DynamicIndexHeader>();
 103
 104         let header: Box<DynamicIndexHeader> = unsafe { file.read_host_value_boxed()? };
 105
 106         if header.magic != super::DYNAMIC_SIZED_CHUNK_INDEX_1_0 {
 107             bail!("got unknown magic number");
 108         }
 109
 110         let ctime = proxmox::tools::time::epoch_i64();
 111
 112         let rawfd = file.as_raw_fd();
 113
 114         let stat = nix::sys::stat::fstat(rawfd)?;
 115
 116         let size = stat.st_size as usize;
 117
 118         let index_size = size - 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     #[allow(clippy::cast_ptr_alignment)]
 223     fn chunk_info(&self, pos: usize) -> Option<ChunkReadInfo> {
 224         if pos >= self.index.len() {
 225             return None;
 226         }
 227         let start = if pos == 0 { 0 } else { self.index[pos - 1].end() };
 228
 229         let end = self.index[pos].end();
 230
 231         Some(ChunkReadInfo {
 232             range: start..end,
 233             digest: self.index[pos].digest.clone(),
 234         })
 235     }
 236
 237     fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
 238         let end_idx = self.index.len() - 1;
 239         let end = self.chunk_end(end_idx);
 240         let found_idx = self.binary_search(0, 0, end_idx, end, offset);
 241         let found_idx = match found_idx {
 242             Ok(i) => i,
 243             Err(_) => return None
 244         };
 245
 246         let found_start = if found_idx == 0 {
 247             0
 248         } else {
 249             self.chunk_end(found_idx - 1)
 250         };
 251
 252         Some((found_idx, offset - found_start))
 253     }
 254 }
 255
 256 struct CachedChunk {
 257     range: Range<u64>,
 258     data: Vec<u8>,
 259 }
 260
 261 impl CachedChunk {
 262     /// Perform sanity checks on the range and data size:
 263     pub fn new(range: Range<u64>, data: Vec<u8>) -> Result<Self, Error> {
 264         if data.len() as u64 != range.end - range.start {
 265             bail!(
 266                 "read chunk with wrong size ({} != {})",
 267                 data.len(),
 268                 range.end - range.start,
 269             );
 270         }
 271         Ok(Self { range, data })
 272     }
 273 }
 274
 275 pub struct BufferedDynamicReader<S> {
 276     store: S,
 277     index: DynamicIndexReader,
 278     archive_size: u64,
 279     read_buffer: Vec<u8>,
 280     buffered_chunk_idx: usize,
 281     buffered_chunk_start: u64,
 282     read_offset: u64,
 283     lru_cache: crate::tools::lru_cache::LruCache<usize, CachedChunk>,
 284 }
 285
 286 struct ChunkCacher<'a, S> {
 287     store: &'a mut S,
 288     index: &'a DynamicIndexReader,
 289 }
 290
 291 impl<'a, S: ReadChunk> crate::tools::lru_cache::Cacher<usize, CachedChunk> for ChunkCacher<'a, S> {
 292     fn fetch(&mut self, index: usize) -> Result<Option<CachedChunk>, Error> {
 293         let info = match self.index.chunk_info(index) {
 294             Some(info) => info,
 295             None => bail!("chunk index out of range"),
 296         };
 297         let range = info.range;
 298         let data = self.store.read_chunk(&info.digest)?;
 299         CachedChunk::new(range, data).map(Some)
 300     }
 301 }
 302
 303 impl<S: ReadChunk> BufferedDynamicReader<S> {
 304     pub fn new(index: DynamicIndexReader, store: S) -> Self {
 305         let archive_size = index.index_bytes();
 306         Self {
 307             store,
 308             index,
 309             archive_size,
 310             read_buffer: Vec::with_capacity(1024 * 1024),
 311             buffered_chunk_idx: 0,
 312             buffered_chunk_start: 0,
 313             read_offset: 0,
 314             lru_cache: crate::tools::lru_cache::LruCache::new(32),
 315         }
 316     }
 317
 318     pub fn archive_size(&self) -> u64 {
 319         self.archive_size
 320     }
 321
 322     fn buffer_chunk(&mut self, idx: usize) -> Result<(), Error> {
 323         //let (start, end, data) = self.lru_cache.access(
 324         let cached_chunk = self.lru_cache.access(
 325             idx,
 326             &mut ChunkCacher {
 327                 store: &mut self.store,
 328                 index: &self.index,
 329             },
 330         )?.ok_or_else(|| format_err!("chunk not found by cacher"))?;
 331
 332         // fixme: avoid copy
 333         self.read_buffer.clear();
 334         self.read_buffer.extend_from_slice(&cached_chunk.data);
 335
 336         self.buffered_chunk_idx = idx;
 337
 338         self.buffered_chunk_start = cached_chunk.range.start;
 339         //println!("BUFFER {} {}",  self.buffered_chunk_start, end);
 340         Ok(())
 341     }
 342 }
 343
 344 impl<S: ReadChunk> crate::tools::BufferedRead for BufferedDynamicReader<S> {
 345     fn buffered_read(&mut self, offset: u64) -> Result<&[u8], Error> {
 346         if offset == self.archive_size {
 347             return Ok(&self.read_buffer[0..0]);
 348         }
 349
 350         let buffer_len = self.read_buffer.len();
 351         let index = &self.index;
 352
 353         // optimization for sequential read
 354         if buffer_len > 0
 355             && ((self.buffered_chunk_idx + 1) < index.index.len())
 356             && (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
 357         {
 358             let next_idx = self.buffered_chunk_idx + 1;
 359             let next_end = index.chunk_end(next_idx);
 360             if offset < next_end {
 361                 self.buffer_chunk(next_idx)?;
 362                 let buffer_offset = (offset - self.buffered_chunk_start) as usize;
 363                 return Ok(&self.read_buffer[buffer_offset..]);
 364             }
 365         }
 366
 367         if (buffer_len == 0)
 368             || (offset < self.buffered_chunk_start)
 369             || (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
 370         {
 371             let end_idx = index.index.len() - 1;
 372             let end = index.chunk_end(end_idx);
 373             let idx = index.binary_search(0, 0, end_idx, end, offset)?;
 374             self.buffer_chunk(idx)?;
 375         }
 376
 377         let buffer_offset = (offset - self.buffered_chunk_start) as usize;
 378         Ok(&self.read_buffer[buffer_offset..])
 379     }
 380 }
 381
 382 impl<S: ReadChunk> std::io::Read for BufferedDynamicReader<S> {
 383     fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
 384         use crate::tools::BufferedRead;
 385         use std::io::{Error, ErrorKind};
 386
 387         let data = match self.buffered_read(self.read_offset) {
 388             Ok(v) => v,
 389             Err(err) => return Err(Error::new(ErrorKind::Other, err.to_string())),
 390         };
 391
 392         let n = if data.len() > buf.len() {
 393             buf.len()
 394         } else {
 395             data.len()
 396         };
 397
 398         buf[0..n].copy_from_slice(&data[0..n]);
 399
 400         self.read_offset += n as u64;
 401
 402         Ok(n)
 403     }
 404 }
 405
 406 impl<S: ReadChunk> std::io::Seek for BufferedDynamicReader<S> {
 407     fn seek(&mut self, pos: SeekFrom) -> Result<u64, std::io::Error> {
 408         let new_offset = match pos {
 409             SeekFrom::Start(start_offset) => start_offset as i64,
 410             SeekFrom::End(end_offset) => (self.archive_size as i64) + end_offset,
 411             SeekFrom::Current(offset) => (self.read_offset as i64) + offset,
 412         };
 413
 414         use std::io::{Error, ErrorKind};
 415         if (new_offset < 0) || (new_offset > (self.archive_size as i64)) {
 416             return Err(Error::new(
 417                 ErrorKind::Other,
 418                 format!(
 419                     "seek is out of range {} ([0..{}])",
 420                     new_offset, self.archive_size
 421                 ),
 422             ));
 423         }
 424         self.read_offset = new_offset as u64;
 425
 426         Ok(self.read_offset)
 427     }
 428 }
 429
 430 /// This is a workaround until we have cleaned up the chunk/reader/... infrastructure for better
 431 /// async use!
 432 ///
 433 /// Ideally BufferedDynamicReader gets replaced so the LruCache maps to `BroadcastFuture<Chunk>`,
 434 /// so that we can properly access it from multiple threads simultaneously while not issuing
 435 /// duplicate simultaneous reads over http.
 436 #[derive(Clone)]
 437 pub struct LocalDynamicReadAt<R: ReadChunk> {
 438     inner: Arc<Mutex<BufferedDynamicReader<R>>>,
 439 }
 440
 441 impl<R: ReadChunk> LocalDynamicReadAt<R> {
 442     pub fn new(inner: BufferedDynamicReader<R>) -> Self {
 443         Self {
 444             inner: Arc::new(Mutex::new(inner)),
 445         }
 446     }
 447 }
 448
 449 impl<R: ReadChunk> ReadAt for LocalDynamicReadAt<R> {
 450     fn start_read_at<'a>(
 451         self: Pin<&'a Self>,
 452         _cx: &mut Context,
 453         buf: &'a mut [u8],
 454         offset: u64,
 455     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
 456         use std::io::Read;
 457         MaybeReady::Ready(tokio::task::block_in_place(move || {
 458             let mut reader = self.inner.lock().unwrap();
 459             reader.seek(SeekFrom::Start(offset))?;
 460             Ok(reader.read(buf)?)
 461         }))
 462     }
 463
 464     fn poll_complete<'a>(
 465         self: Pin<&'a Self>,
 466         _op: ReadAtOperation<'a>,
 467     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
 468         panic!("LocalDynamicReadAt::start_read_at returned Pending");
 469     }
 470 }
 471
 472
 473 /// Create dynamic index files (`.dixd`)
 474 pub struct DynamicIndexWriter {
 475     store: Arc<ChunkStore>,
 476     _lock: tools::ProcessLockSharedGuard,
 477     writer: BufWriter<File>,
 478     closed: bool,
 479     filename: PathBuf,
 480     tmp_filename: PathBuf,
 481     csum: Option<openssl::sha::Sha256>,
 482     pub uuid: [u8; 16],
 483     pub ctime: i64,
 484 }
 485
 486 impl Drop for DynamicIndexWriter {
 487     fn drop(&mut self) {
 488         let _ = std::fs::remove_file(&self.tmp_filename); // ignore errors
 489     }
 490 }
 491
 492 impl DynamicIndexWriter {
 493     pub fn create(store: Arc<ChunkStore>, path: &Path) -> Result<Self, Error> {
 494         let shared_lock = store.try_shared_lock()?;
 495
 496         let full_path = store.relative_path(path);
 497         let mut tmp_path = full_path.clone();
 498         tmp_path.set_extension("tmp_didx");
 499
 500         let file = std::fs::OpenOptions::new()
 501             .create(true)
 502             .truncate(true)
 503             .read(true)
 504             .write(true)
 505             .open(&tmp_path)?;
 506
 507         let mut writer = BufWriter::with_capacity(1024 * 1024, file);
 508
 509         let ctime = proxmox::tools::time::epoch_i64();
 510
 511         let uuid = Uuid::generate();
 512
 513         let mut header = DynamicIndexHeader::zeroed();
 514         header.magic = super::DYNAMIC_SIZED_CHUNK_INDEX_1_0;
 515         header.ctime = i64::to_le(ctime);
 516         header.uuid = *uuid.as_bytes();
 517         // header.index_csum = [0u8; 32];
 518         writer.write_all(header.as_bytes())?;
 519
 520         let csum = Some(openssl::sha::Sha256::new());
 521
 522         Ok(Self {
 523             store,
 524             _lock: shared_lock,
 525             writer,
 526             closed: false,
 527             filename: full_path,
 528             tmp_filename: tmp_path,
 529             ctime,
 530             uuid: *uuid.as_bytes(),
 531             csum,
 532         })
 533     }
 534
 535     // fixme: use add_chunk instead?
 536     pub fn insert_chunk(&self, chunk: &DataBlob, digest: &[u8; 32]) -> Result<(bool, u64), Error> {
 537         self.store.insert_chunk(chunk, digest)
 538     }
 539
 540     pub fn close(&mut self) -> Result<[u8; 32], Error> {
 541         if self.closed {
 542             bail!(
 543                 "cannot close already closed archive index file {:?}",
 544                 self.filename
 545             );
 546         }
 547
 548         self.closed = true;
 549
 550         self.writer.flush()?;
 551
 552         let csum_offset = proxmox::offsetof!(DynamicIndexHeader, index_csum);
 553         self.writer.seek(SeekFrom::Start(csum_offset as u64))?;
 554
 555         let csum = self.csum.take().unwrap();
 556         let index_csum = csum.finish();
 557
 558         self.writer.write_all(&index_csum)?;
 559         self.writer.flush()?;
 560
 561         if let Err(err) = std::fs::rename(&self.tmp_filename, &self.filename) {
 562             bail!("Atomic rename file {:?} failed - {}", self.filename, err);
 563         }
 564
 565         Ok(index_csum)
 566     }
 567
 568     // fixme: rename to add_digest
 569     pub fn add_chunk(&mut self, offset: u64, digest: &[u8; 32]) -> Result<(), Error> {
 570         if self.closed {
 571             bail!(
 572                 "cannot write to closed dynamic index file {:?}",
 573                 self.filename
 574             );
 575         }
 576
 577         let offset_le: &[u8; 8] = unsafe { &std::mem::transmute::<u64, [u8; 8]>(offset.to_le()) };
 578
 579         if let Some(ref mut csum) = self.csum {
 580             csum.update(offset_le);
 581             csum.update(digest);
 582         }
 583
 584         self.writer.write_all(offset_le)?;
 585         self.writer.write_all(digest)?;
 586         Ok(())
 587     }
 588 }
 589
 590 /// Writer which splits a binary stream into dynamic sized chunks
 591 ///
 592 /// And store the resulting chunk list into the index file.
 593 pub struct DynamicChunkWriter {
 594     index: DynamicIndexWriter,
 595     closed: bool,
 596     chunker: Chunker,
 597     stat: ChunkStat,
 598     chunk_offset: usize,
 599     last_chunk: usize,
 600     chunk_buffer: Vec<u8>,
 601 }
 602
 603 impl DynamicChunkWriter {
 604     pub fn new(index: DynamicIndexWriter, chunk_size: usize) -> Self {
 605         Self {
 606             index,
 607             closed: false,
 608             chunker: Chunker::new(chunk_size),
 609             stat: ChunkStat::new(0),
 610             chunk_offset: 0,
 611             last_chunk: 0,
 612             chunk_buffer: Vec::with_capacity(chunk_size * 4),
 613         }
 614     }
 615
 616     pub fn stat(&self) -> &ChunkStat {
 617         &self.stat
 618     }
 619
 620     pub fn close(&mut self) -> Result<(), Error> {
 621         if self.closed {
 622             return Ok(());
 623         }
 624
 625         self.closed = true;
 626
 627         self.write_chunk_buffer()?;
 628
 629         self.index.close()?;
 630
 631         self.stat.size = self.chunk_offset as u64;
 632
 633         // add size of index file
 634         self.stat.size +=
 635             (self.stat.chunk_count * 40 + std::mem::size_of::<DynamicIndexHeader>()) as u64;
 636
 637         Ok(())
 638     }
 639
 640     fn write_chunk_buffer(&mut self) -> Result<(), Error> {
 641         let chunk_size = self.chunk_buffer.len();
 642
 643         if chunk_size == 0 {
 644             return Ok(());
 645         }
 646
 647         let expected_chunk_size = self.chunk_offset - self.last_chunk;
 648         if expected_chunk_size != self.chunk_buffer.len() {
 649             bail!("wrong chunk size {} != {}", expected_chunk_size, chunk_size);
 650         }
 651
 652         self.stat.chunk_count += 1;
 653
 654         self.last_chunk = self.chunk_offset;
 655
 656         let (chunk, digest) = DataChunkBuilder::new(&self.chunk_buffer)
 657             .compress(true)
 658             .build()?;
 659
 660         match self.index.insert_chunk(&chunk, &digest) {
 661             Ok((is_duplicate, compressed_size)) => {
 662                 self.stat.compressed_size += compressed_size;
 663                 if is_duplicate {
 664                     self.stat.duplicate_chunks += 1;
 665                 } else {
 666                     self.stat.disk_size += compressed_size;
 667                 }
 668
 669                 println!(
 670                     "ADD CHUNK {:016x} {} {}% {} {}",
 671                     self.chunk_offset,
 672                     chunk_size,
 673                     (compressed_size * 100) / (chunk_size as u64),
 674                     is_duplicate,
 675                     proxmox::tools::digest_to_hex(&digest)
 676                 );
 677                 self.index.add_chunk(self.chunk_offset as u64, &digest)?;
 678                 self.chunk_buffer.truncate(0);
 679                 Ok(())
 680             }
 681             Err(err) => {
 682                 self.chunk_buffer.truncate(0);
 683                 Err(err)
 684             }
 685         }
 686     }
 687 }
 688
 689 impl Write for DynamicChunkWriter {
 690     fn write(&mut self, data: &[u8]) -> std::result::Result<usize, std::io::Error> {
 691         let chunker = &mut self.chunker;
 692
 693         let pos = chunker.scan(data);
 694
 695         if pos > 0 {
 696             self.chunk_buffer.extend_from_slice(&data[0..pos]);
 697             self.chunk_offset += pos;
 698
 699             if let Err(err) = self.write_chunk_buffer() {
 700                 return Err(std::io::Error::new(
 701                     std::io::ErrorKind::Other,
 702                     err.to_string(),
 703                 ));
 704             }
 705             Ok(pos)
 706         } else {
 707             self.chunk_offset += data.len();
 708             self.chunk_buffer.extend_from_slice(data);
 709             Ok(data.len())
 710         }
 711     }
 712
 713     fn flush(&mut self) -> std::result::Result<(), std::io::Error> {
 714         Err(std::io::Error::new(
 715             std::io::ErrorKind::Other,
 716             "please use close() instead of flush()",
 717         ))
 718     }
 719 }