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::vec;
  15 use proxmox::tools::mmap::Mmap;
  16 use pxar::accessor::{MaybeReady, ReadAt, ReadAtOperation};
  17
  18 use super::chunk_stat::ChunkStat;
  19 use super::chunk_store::ChunkStore;
  20 use super::index::ChunkReadInfo;
  21 use super::read_chunk::ReadChunk;
  22 use super::Chunker;
  23 use super::IndexFile;
  24 use super::{DataBlob, DataChunkBuilder};
  25 use crate::tools::{self, epoch_now_u64};
  26
  27 /// Header format definition for dynamic index files (`.dixd`)
  28 #[repr(C)]
  29 pub struct DynamicIndexHeader {
  30     pub magic: [u8; 8],
  31     pub uuid: [u8; 16],
  32     pub ctime: u64,
  33     /// Sha256 over the index ``SHA256(offset1||digest1||offset2||digest2||...)``
  34     pub index_csum: [u8; 32],
  35     reserved: [u8; 4032], // overall size is one page (4096 bytes)
  36 }
  37 proxmox::static_assert_size!(DynamicIndexHeader, 4096);
  38 // TODO: Once non-Copy unions are stabilized, use:
  39 // union DynamicIndexHeader {
  40 //     reserved: [u8; 4096],
  41 //     pub data: DynamicIndexHeaderData,
  42 // }
  43
  44 #[derive(Clone, Debug)]
  45 #[repr(C)]
  46 pub struct DynamicEntry {
  47     end_le: u64,
  48     digest: [u8; 32],
  49 }
  50
  51 impl DynamicEntry {
  52     #[inline]
  53     pub fn end(&self) -> u64 {
  54         u64::from_le(self.end_le)
  55     }
  56 }
  57
  58 pub struct DynamicIndexReader {
  59     _file: File,
  60     pub size: usize,
  61     index: Mmap<DynamicEntry>,
  62     pub uuid: [u8; 16],
  63     pub ctime: u64,
  64     pub index_csum: [u8; 32],
  65 }
  66
  67 impl DynamicIndexReader {
  68     pub fn open(path: &Path) -> Result<Self, Error> {
  69         File::open(path)
  70             .map_err(Error::from)
  71             .and_then(Self::new)
  72             .map_err(|err| format_err!("Unable to open dynamic index {:?} - {}", path, err))
  73     }
  74
  75     pub fn new(mut file: std::fs::File) -> Result<Self, Error> {
  76         if let Err(err) =
  77             nix::fcntl::flock(file.as_raw_fd(), nix::fcntl::FlockArg::LockSharedNonblock)
  78         {
  79             bail!("unable to get shared lock - {}", err);
  80         }
  81
  82         // FIXME: This is NOT OUR job! Check the callers of this method and remove this!
  83         file.seek(SeekFrom::Start(0))?;
  84
  85         let header_size = std::mem::size_of::<DynamicIndexHeader>();
  86
  87         let header: Box<DynamicIndexHeader> = unsafe { file.read_host_value_boxed()? };
  88
  89         if header.magic != super::DYNAMIC_SIZED_CHUNK_INDEX_1_0 {
  90             bail!("got unknown magic number");
  91         }
  92
  93         let ctime = u64::from_le(header.ctime);
  94
  95         let rawfd = file.as_raw_fd();
  96
  97         let stat = nix::sys::stat::fstat(rawfd)?;
  98
  99         let size = stat.st_size as usize;
 100
 101         let index_size = size - header_size;
 102         let index_count = index_size / 40;
 103         if index_count * 40 != index_size {
 104             bail!("got unexpected file size");
 105         }
 106
 107         let index = unsafe {
 108             Mmap::map_fd(
 109                 rawfd,
 110                 header_size as u64,
 111                 index_count,
 112                 nix::sys::mman::ProtFlags::PROT_READ,
 113                 nix::sys::mman::MapFlags::MAP_PRIVATE,
 114             )?
 115         };
 116
 117         Ok(Self {
 118             _file: file,
 119             size,
 120             index,
 121             ctime,
 122             uuid: header.uuid,
 123             index_csum: header.index_csum,
 124         })
 125     }
 126
 127     #[inline]
 128     #[allow(clippy::cast_ptr_alignment)]
 129     fn chunk_end(&self, pos: usize) -> u64 {
 130         if pos >= self.index.len() {
 131             panic!("chunk index out of range");
 132         }
 133         self.index[pos].end()
 134     }
 135
 136     #[inline]
 137     fn chunk_digest(&self, pos: usize) -> &[u8; 32] {
 138         if pos >= self.index.len() {
 139             panic!("chunk index out of range");
 140         }
 141         &self.index[pos].digest
 142     }
 143
 144     // TODO: can we use std::slice::binary_search with Mmap now?
 145     fn binary_search(
 146         &self,
 147         start_idx: usize,
 148         start: u64,
 149         end_idx: usize,
 150         end: u64,
 151         offset: u64,
 152     ) -> Result<usize, Error> {
 153         if (offset >= end) || (offset < start) {
 154             bail!("offset out of range");
 155         }
 156
 157         if end_idx == start_idx {
 158             return Ok(start_idx); // found
 159         }
 160         let middle_idx = (start_idx + end_idx) / 2;
 161         let middle_end = self.chunk_end(middle_idx);
 162
 163         if offset < middle_end {
 164             self.binary_search(start_idx, start, middle_idx, middle_end, offset)
 165         } else {
 166             self.binary_search(middle_idx + 1, middle_end, end_idx, end, offset)
 167         }
 168     }
 169 }
 170
 171 impl IndexFile for DynamicIndexReader {
 172     fn index_count(&self) -> usize {
 173         self.index.len()
 174     }
 175
 176     fn index_digest(&self, pos: usize) -> Option<&[u8; 32]> {
 177         if pos >= self.index.len() {
 178             None
 179         } else {
 180             Some(unsafe { std::mem::transmute(self.chunk_digest(pos).as_ptr()) })
 181         }
 182     }
 183
 184     fn index_bytes(&self) -> u64 {
 185         if self.index.is_empty() {
 186             0
 187         } else {
 188             self.chunk_end(self.index.len() - 1)
 189         }
 190     }
 191
 192     fn compute_csum(&self) -> ([u8; 32], u64) {
 193         let mut csum = openssl::sha::Sha256::new();
 194         let mut chunk_end = 0;
 195         for pos in 0..self.index_count() {
 196             let info = self.chunk_info(pos).unwrap();
 197             chunk_end = info.range.end;
 198             csum.update(&chunk_end.to_le_bytes());
 199             csum.update(&info.digest);
 200         }
 201         let csum = csum.finish();
 202         (csum, chunk_end)
 203     }
 204
 205     #[allow(clippy::cast_ptr_alignment)]
 206     fn chunk_info(&self, pos: usize) -> Option<ChunkReadInfo> {
 207         if pos >= self.index.len() {
 208             return None;
 209         }
 210         let start = if pos == 0 { 0 } else { self.index[pos - 1].end() };
 211
 212         let end = self.index[pos].end();
 213
 214         Some(ChunkReadInfo {
 215             range: start..end,
 216             digest: self.index[pos].digest.clone(),
 217         })
 218     }
 219
 220     fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
 221         let end_idx = self.index.len() - 1;
 222         let end = self.chunk_end(end_idx);
 223         let found_idx = self.binary_search(0, 0, end_idx, end, offset);
 224         let found_idx = match found_idx {
 225             Ok(i) => i,
 226             Err(_) => return None
 227         };
 228
 229         let found_start = if found_idx == 0 {
 230             0
 231         } else {
 232             self.chunk_end(found_idx - 1)
 233         };
 234
 235         Some((found_idx, offset - found_start))
 236     }
 237 }
 238
 239 struct CachedChunk {
 240     range: Range<u64>,
 241     data: Vec<u8>,
 242 }
 243
 244 impl CachedChunk {
 245     /// Perform sanity checks on the range and data size:
 246     pub fn new(range: Range<u64>, data: Vec<u8>) -> Result<Self, Error> {
 247         if data.len() as u64 != range.end - range.start {
 248             bail!(
 249                 "read chunk with wrong size ({} != {})",
 250                 data.len(),
 251                 range.end - range.start,
 252             );
 253         }
 254         Ok(Self { range, data })
 255     }
 256 }
 257
 258 pub struct BufferedDynamicReader<S> {
 259     store: S,
 260     index: DynamicIndexReader,
 261     archive_size: u64,
 262     read_buffer: Vec<u8>,
 263     buffered_chunk_idx: usize,
 264     buffered_chunk_start: u64,
 265     read_offset: u64,
 266     lru_cache: crate::tools::lru_cache::LruCache<usize, CachedChunk>,
 267 }
 268
 269 struct ChunkCacher<'a, S> {
 270     store: &'a mut S,
 271     index: &'a DynamicIndexReader,
 272 }
 273
 274 impl<'a, S: ReadChunk> crate::tools::lru_cache::Cacher<usize, CachedChunk> for ChunkCacher<'a, S> {
 275     fn fetch(&mut self, index: usize) -> Result<Option<CachedChunk>, Error> {
 276         let info = match self.index.chunk_info(index) {
 277             Some(info) => info,
 278             None => bail!("chunk index out of range"),
 279         };
 280         let range = info.range;
 281         let data = self.store.read_chunk(&info.digest)?;
 282         CachedChunk::new(range, data).map(Some)
 283     }
 284 }
 285
 286 impl<S: ReadChunk> BufferedDynamicReader<S> {
 287     pub fn new(index: DynamicIndexReader, store: S) -> Self {
 288         let archive_size = index.index_bytes();
 289         Self {
 290             store,
 291             index,
 292             archive_size,
 293             read_buffer: Vec::with_capacity(1024 * 1024),
 294             buffered_chunk_idx: 0,
 295             buffered_chunk_start: 0,
 296             read_offset: 0,
 297             lru_cache: crate::tools::lru_cache::LruCache::new(32),
 298         }
 299     }
 300
 301     pub fn archive_size(&self) -> u64 {
 302         self.archive_size
 303     }
 304
 305     fn buffer_chunk(&mut self, idx: usize) -> Result<(), Error> {
 306         //let (start, end, data) = self.lru_cache.access(
 307         let cached_chunk = self.lru_cache.access(
 308             idx,
 309             &mut ChunkCacher {
 310                 store: &mut self.store,
 311                 index: &self.index,
 312             },
 313         )?.ok_or_else(|| format_err!("chunk not found by cacher"))?;
 314
 315         // fixme: avoid copy
 316         self.read_buffer.clear();
 317         self.read_buffer.extend_from_slice(&cached_chunk.data);
 318
 319         self.buffered_chunk_idx = idx;
 320
 321         self.buffered_chunk_start = cached_chunk.range.start;
 322         //println!("BUFFER {} {}",  self.buffered_chunk_start, end);
 323         Ok(())
 324     }
 325 }
 326
 327 impl<S: ReadChunk> crate::tools::BufferedRead for BufferedDynamicReader<S> {
 328     fn buffered_read(&mut self, offset: u64) -> Result<&[u8], Error> {
 329         if offset == self.archive_size {
 330             return Ok(&self.read_buffer[0..0]);
 331         }
 332
 333         let buffer_len = self.read_buffer.len();
 334         let index = &self.index;
 335
 336         // optimization for sequential read
 337         if buffer_len > 0
 338             && ((self.buffered_chunk_idx + 1) < index.index.len())
 339             && (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
 340         {
 341             let next_idx = self.buffered_chunk_idx + 1;
 342             let next_end = index.chunk_end(next_idx);
 343             if offset < next_end {
 344                 self.buffer_chunk(next_idx)?;
 345                 let buffer_offset = (offset - self.buffered_chunk_start) as usize;
 346                 return Ok(&self.read_buffer[buffer_offset..]);
 347             }
 348         }
 349
 350         if (buffer_len == 0)
 351             || (offset < self.buffered_chunk_start)
 352             || (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
 353         {
 354             let end_idx = index.index.len() - 1;
 355             let end = index.chunk_end(end_idx);
 356             let idx = index.binary_search(0, 0, end_idx, end, offset)?;
 357             self.buffer_chunk(idx)?;
 358         }
 359
 360         let buffer_offset = (offset - self.buffered_chunk_start) as usize;
 361         Ok(&self.read_buffer[buffer_offset..])
 362     }
 363 }
 364
 365 impl<S: ReadChunk> std::io::Read for BufferedDynamicReader<S> {
 366     fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
 367         use crate::tools::BufferedRead;
 368         use std::io::{Error, ErrorKind};
 369
 370         let data = match self.buffered_read(self.read_offset) {
 371             Ok(v) => v,
 372             Err(err) => return Err(Error::new(ErrorKind::Other, err.to_string())),
 373         };
 374
 375         let n = if data.len() > buf.len() {
 376             buf.len()
 377         } else {
 378             data.len()
 379         };
 380
 381         buf[0..n].copy_from_slice(&data[0..n]);
 382
 383         self.read_offset += n as u64;
 384
 385         Ok(n)
 386     }
 387 }
 388
 389 impl<S: ReadChunk> std::io::Seek for BufferedDynamicReader<S> {
 390     fn seek(&mut self, pos: SeekFrom) -> Result<u64, std::io::Error> {
 391         let new_offset = match pos {
 392             SeekFrom::Start(start_offset) => start_offset as i64,
 393             SeekFrom::End(end_offset) => (self.archive_size as i64) + end_offset,
 394             SeekFrom::Current(offset) => (self.read_offset as i64) + offset,
 395         };
 396
 397         use std::io::{Error, ErrorKind};
 398         if (new_offset < 0) || (new_offset > (self.archive_size as i64)) {
 399             return Err(Error::new(
 400                 ErrorKind::Other,
 401                 format!(
 402                     "seek is out of range {} ([0..{}])",
 403                     new_offset, self.archive_size
 404                 ),
 405             ));
 406         }
 407         self.read_offset = new_offset as u64;
 408
 409         Ok(self.read_offset)
 410     }
 411 }
 412
 413 /// This is a workaround until we have cleaned up the chunk/reader/... infrastructure for better
 414 /// async use!
 415 ///
 416 /// Ideally BufferedDynamicReader gets replaced so the LruCache maps to `BroadcastFuture<Chunk>`,
 417 /// so that we can properly access it from multiple threads simultaneously while not issuing
 418 /// duplicate simultaneous reads over http.
 419 #[derive(Clone)]
 420 pub struct LocalDynamicReadAt<R: ReadChunk> {
 421     inner: Arc<Mutex<BufferedDynamicReader<R>>>,
 422 }
 423
 424 impl<R: ReadChunk> LocalDynamicReadAt<R> {
 425     pub fn new(inner: BufferedDynamicReader<R>) -> Self {
 426         Self {
 427             inner: Arc::new(Mutex::new(inner)),
 428         }
 429     }
 430 }
 431
 432 impl<R: ReadChunk> ReadAt for LocalDynamicReadAt<R> {
 433     fn start_read_at<'a>(
 434         self: Pin<&'a Self>,
 435         _cx: &mut Context,
 436         buf: &'a mut [u8],
 437         offset: u64,
 438     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
 439         use std::io::Read;
 440         MaybeReady::Ready(tokio::task::block_in_place(move || {
 441             let mut reader = self.inner.lock().unwrap();
 442             reader.seek(SeekFrom::Start(offset))?;
 443             Ok(reader.read(buf)?)
 444         }))
 445     }
 446
 447     fn poll_complete<'a>(
 448         self: Pin<&'a Self>,
 449         _op: ReadAtOperation<'a>,
 450     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
 451         panic!("LocalDynamicReadAt::start_read_at returned Pending");
 452     }
 453 }
 454
 455
 456 /// Create dynamic index files (`.dixd`)
 457 pub struct DynamicIndexWriter {
 458     store: Arc<ChunkStore>,
 459     _lock: tools::ProcessLockSharedGuard,
 460     writer: BufWriter<File>,
 461     closed: bool,
 462     filename: PathBuf,
 463     tmp_filename: PathBuf,
 464     csum: Option<openssl::sha::Sha256>,
 465     pub uuid: [u8; 16],
 466     pub ctime: u64,
 467 }
 468
 469 impl Drop for DynamicIndexWriter {
 470     fn drop(&mut self) {
 471         let _ = std::fs::remove_file(&self.tmp_filename); // ignore errors
 472     }
 473 }
 474
 475 impl DynamicIndexWriter {
 476     pub fn create(store: Arc<ChunkStore>, path: &Path) -> Result<Self, Error> {
 477         let shared_lock = store.try_shared_lock()?;
 478
 479         let full_path = store.relative_path(path);
 480         let mut tmp_path = full_path.clone();
 481         tmp_path.set_extension("tmp_didx");
 482
 483         let file = std::fs::OpenOptions::new()
 484             .create(true)
 485             .truncate(true)
 486             .read(true)
 487             .write(true)
 488             .open(&tmp_path)?;
 489
 490         let mut writer = BufWriter::with_capacity(1024 * 1024, file);
 491
 492         let header_size = std::mem::size_of::<DynamicIndexHeader>();
 493
 494         // todo: use static assertion when available in rust
 495         if header_size != 4096 {
 496             panic!("got unexpected header size");
 497         }
 498
 499         let ctime = epoch_now_u64()?;
 500
 501         let uuid = Uuid::generate();
 502
 503         let mut buffer = vec::zeroed(header_size);
 504         let header = crate::tools::map_struct_mut::<DynamicIndexHeader>(&mut buffer)?;
 505
 506         header.magic = super::DYNAMIC_SIZED_CHUNK_INDEX_1_0;
 507         header.ctime = u64::to_le(ctime);
 508         header.uuid = *uuid.as_bytes();
 509
 510         header.index_csum = [0u8; 32];
 511
 512         writer.write_all(&buffer)?;
 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 }