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