pbs-client/src/backup_writer.rs

   1 use std::collections::HashSet;
   2 use std::future::Future;
   3 use std::os::unix::fs::OpenOptionsExt;
   4 use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
   5 use std::sync::{Arc, Mutex};
   6
   7 use anyhow::{bail, format_err, Error};
   8 use futures::future::{self, AbortHandle, Either, FutureExt, TryFutureExt};
   9 use futures::stream::{Stream, StreamExt, TryStreamExt};
  10 use serde_json::{json, Value};
  11 use tokio::io::AsyncReadExt;
  12 use tokio::sync::{mpsc, oneshot};
  13 use tokio_stream::wrappers::ReceiverStream;
  14
  15 use proxmox::tools::digest_to_hex;
  16
  17 use pbs_api_types::HumanByte;
  18 use pbs_tools::crypt_config::CryptConfig;
  19 use pbs_datastore::{CATALOG_NAME, PROXMOX_BACKUP_PROTOCOL_ID_V1};
  20 use pbs_datastore::data_blob::{ChunkInfo, DataBlob, DataChunkBuilder};
  21 use pbs_datastore::dynamic_index::DynamicIndexReader;
  22 use pbs_datastore::fixed_index::FixedIndexReader;
  23 use pbs_datastore::index::IndexFile;
  24 use pbs_datastore::manifest::{ArchiveType, BackupManifest, MANIFEST_BLOB_NAME};
  25
  26 use super::merge_known_chunks::{MergeKnownChunks, MergedChunkInfo};
  27
  28 use super::{H2Client, HttpClient};
  29
  30 pub struct BackupWriter {
  31     h2: H2Client,
  32     abort: AbortHandle,
  33     verbose: bool,
  34     crypt_config: Option<Arc<CryptConfig>>,
  35 }
  36
  37 impl Drop for BackupWriter {
  38     fn drop(&mut self) {
  39         self.abort.abort();
  40     }
  41 }
  42
  43 pub struct BackupStats {
  44     pub size: u64,
  45     pub csum: [u8; 32],
  46 }
  47
  48 /// Options for uploading blobs/streams to the server
  49 #[derive(Default, Clone)]
  50 pub struct UploadOptions {
  51     pub previous_manifest: Option<Arc<BackupManifest>>,
  52     pub compress: bool,
  53     pub encrypt: bool,
  54     pub fixed_size: Option<u64>,
  55 }
  56
  57 struct UploadStats {
  58     chunk_count: usize,
  59     chunk_reused: usize,
  60     size: usize,
  61     size_reused: usize,
  62     size_compressed: usize,
  63     duration: std::time::Duration,
  64     csum: [u8; 32],
  65 }
  66
  67 type UploadQueueSender = mpsc::Sender<(MergedChunkInfo, Option<h2::client::ResponseFuture>)>;
  68 type UploadResultReceiver = oneshot::Receiver<Result<(), Error>>;
  69
  70 impl BackupWriter {
  71     fn new(
  72         h2: H2Client,
  73         abort: AbortHandle,
  74         crypt_config: Option<Arc<CryptConfig>>,
  75         verbose: bool,
  76     ) -> Arc<Self> {
  77         Arc::new(Self {
  78             h2,
  79             abort,
  80             crypt_config,
  81             verbose,
  82         })
  83     }
  84
  85     // FIXME: extract into (flattened) parameter struct?
  86     #[allow(clippy::too_many_arguments)]
  87     pub async fn start(
  88         client: HttpClient,
  89         crypt_config: Option<Arc<CryptConfig>>,
  90         datastore: &str,
  91         backup_type: &str,
  92         backup_id: &str,
  93         backup_time: i64,
  94         debug: bool,
  95         benchmark: bool,
  96     ) -> Result<Arc<BackupWriter>, Error> {
  97         let param = json!({
  98             "backup-type": backup_type,
  99             "backup-id": backup_id,
 100             "backup-time": backup_time,
 101             "store": datastore,
 102             "debug": debug,
 103             "benchmark": benchmark
 104         });
 105
 106         let req = HttpClient::request_builder(
 107             client.server(),
 108             client.port(),
 109             "GET",
 110             "/api2/json/backup",
 111             Some(param),
 112         )
 113         .unwrap();
 114
 115         let (h2, abort) = client
 116             .start_h2_connection(req, String::from(PROXMOX_BACKUP_PROTOCOL_ID_V1!()))
 117             .await?;
 118
 119         Ok(BackupWriter::new(h2, abort, crypt_config, debug))
 120     }
 121
 122     pub async fn get(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
 123         self.h2.get(path, param).await
 124     }
 125
 126     pub async fn put(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
 127         self.h2.put(path, param).await
 128     }
 129
 130     pub async fn post(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
 131         self.h2.post(path, param).await
 132     }
 133
 134     pub async fn upload_post(
 135         &self,
 136         path: &str,
 137         param: Option<Value>,
 138         content_type: &str,
 139         data: Vec<u8>,
 140     ) -> Result<Value, Error> {
 141         self.h2
 142             .upload("POST", path, param, content_type, data)
 143             .await
 144     }
 145
 146     pub async fn send_upload_request(
 147         &self,
 148         method: &str,
 149         path: &str,
 150         param: Option<Value>,
 151         content_type: &str,
 152         data: Vec<u8>,
 153     ) -> Result<h2::client::ResponseFuture, Error> {
 154         let request =
 155             H2Client::request_builder("localhost", method, path, param, Some(content_type))
 156                 .unwrap();
 157         let response_future = self
 158             .h2
 159             .send_request(request, Some(bytes::Bytes::from(data.clone())))
 160             .await?;
 161         Ok(response_future)
 162     }
 163
 164     pub async fn upload_put(
 165         &self,
 166         path: &str,
 167         param: Option<Value>,
 168         content_type: &str,
 169         data: Vec<u8>,
 170     ) -> Result<Value, Error> {
 171         self.h2.upload("PUT", path, param, content_type, data).await
 172     }
 173
 174     pub async fn finish(self: Arc<Self>) -> Result<(), Error> {
 175         let h2 = self.h2.clone();
 176
 177         h2.post("finish", None)
 178             .map_ok(move |_| {
 179                 self.abort.abort();
 180             })
 181             .await
 182     }
 183
 184     pub fn cancel(&self) {
 185         self.abort.abort();
 186     }
 187
 188     pub async fn upload_blob<R: std::io::Read>(
 189         &self,
 190         mut reader: R,
 191         file_name: &str,
 192     ) -> Result<BackupStats, Error> {
 193         let mut raw_data = Vec::new();
 194         // fixme: avoid loading into memory
 195         reader.read_to_end(&mut raw_data)?;
 196
 197         let csum = openssl::sha::sha256(&raw_data);
 198         let param = json!({"encoded-size": raw_data.len(), "file-name": file_name });
 199         let size = raw_data.len() as u64;
 200         let _value = self
 201             .h2
 202             .upload(
 203                 "POST",
 204                 "blob",
 205                 Some(param),
 206                 "application/octet-stream",
 207                 raw_data,
 208             )
 209             .await?;
 210         Ok(BackupStats { size, csum })
 211     }
 212
 213     pub async fn upload_blob_from_data(
 214         &self,
 215         data: Vec<u8>,
 216         file_name: &str,
 217         options: UploadOptions,
 218     ) -> Result<BackupStats, Error> {
 219         let blob = match (options.encrypt, &self.crypt_config) {
 220             (false, _) => DataBlob::encode(&data, None, options.compress)?,
 221             (true, None) => bail!("requested encryption without a crypt config"),
 222             (true, Some(crypt_config)) => {
 223                 DataBlob::encode(&data, Some(crypt_config), options.compress)?
 224             }
 225         };
 226
 227         let raw_data = blob.into_inner();
 228         let size = raw_data.len() as u64;
 229
 230         let csum = openssl::sha::sha256(&raw_data);
 231         let param = json!({"encoded-size": size, "file-name": file_name });
 232         let _value = self
 233             .h2
 234             .upload(
 235                 "POST",
 236                 "blob",
 237                 Some(param),
 238                 "application/octet-stream",
 239                 raw_data,
 240             )
 241             .await?;
 242         Ok(BackupStats { size, csum })
 243     }
 244
 245     pub async fn upload_blob_from_file<P: AsRef<std::path::Path>>(
 246         &self,
 247         src_path: P,
 248         file_name: &str,
 249         options: UploadOptions,
 250     ) -> Result<BackupStats, Error> {
 251         let src_path = src_path.as_ref();
 252
 253         let mut file = tokio::fs::File::open(src_path)
 254             .await
 255             .map_err(|err| format_err!("unable to open file {:?} - {}", src_path, err))?;
 256
 257         let mut contents = Vec::new();
 258
 259         file.read_to_end(&mut contents)
 260             .await
 261             .map_err(|err| format_err!("unable to read file {:?} - {}", src_path, err))?;
 262
 263         self.upload_blob_from_data(contents, file_name, options)
 264             .await
 265     }
 266
 267     pub async fn upload_stream(
 268         &self,
 269         archive_name: &str,
 270         stream: impl Stream<Item = Result<bytes::BytesMut, Error>>,
 271         options: UploadOptions,
 272     ) -> Result<BackupStats, Error> {
 273         let known_chunks = Arc::new(Mutex::new(HashSet::new()));
 274
 275         let mut param = json!({ "archive-name": archive_name });
 276         let prefix = if let Some(size) = options.fixed_size {
 277             param["size"] = size.into();
 278             "fixed"
 279         } else {
 280             "dynamic"
 281         };
 282
 283         if options.encrypt && self.crypt_config.is_none() {
 284             bail!("requested encryption without a crypt config");
 285         }
 286
 287         let index_path = format!("{}_index", prefix);
 288         let close_path = format!("{}_close", prefix);
 289
 290         if let Some(manifest) = options.previous_manifest {
 291             // try, but ignore errors
 292             match ArchiveType::from_path(archive_name) {
 293                 Ok(ArchiveType::FixedIndex) => {
 294                     let _ = self
 295                         .download_previous_fixed_index(
 296                             archive_name,
 297                             &manifest,
 298                             known_chunks.clone(),
 299                         )
 300                         .await;
 301                 }
 302                 Ok(ArchiveType::DynamicIndex) => {
 303                     let _ = self
 304                         .download_previous_dynamic_index(
 305                             archive_name,
 306                             &manifest,
 307                             known_chunks.clone(),
 308                         )
 309                         .await;
 310                 }
 311                 _ => { /* do nothing */ }
 312             }
 313         }
 314
 315         let wid = self
 316             .h2
 317             .post(&index_path, Some(param))
 318             .await?
 319             .as_u64()
 320             .unwrap();
 321
 322         let upload_stats = Self::upload_chunk_info_stream(
 323             self.h2.clone(),
 324             wid,
 325             stream,
 326             &prefix,
 327             known_chunks.clone(),
 328             if options.encrypt {
 329                 self.crypt_config.clone()
 330             } else {
 331                 None
 332             },
 333             options.compress,
 334             self.verbose,
 335         )
 336         .await?;
 337
 338         let size_dirty = upload_stats.size - upload_stats.size_reused;
 339         let size: HumanByte = upload_stats.size.into();
 340         let archive = if self.verbose {
 341             archive_name
 342         } else {
 343             pbs_tools::format::strip_server_file_extension(archive_name)
 344         };
 345         if archive_name != CATALOG_NAME {
 346             let speed: HumanByte =
 347                 ((size_dirty * 1_000_000) / (upload_stats.duration.as_micros() as usize)).into();
 348             let size_dirty: HumanByte = size_dirty.into();
 349             let size_compressed: HumanByte = upload_stats.size_compressed.into();
 350             println!(
 351                 "{}: had to backup {} of {} (compressed {}) in {:.2}s",
 352                 archive,
 353                 size_dirty,
 354                 size,
 355                 size_compressed,
 356                 upload_stats.duration.as_secs_f64()
 357             );
 358             println!("{}: average backup speed: {}/s", archive, speed);
 359         } else {
 360             println!("Uploaded backup catalog ({})", size);
 361         }
 362
 363         if upload_stats.size_reused > 0 && upload_stats.size > 1024 * 1024 {
 364             let reused_percent = upload_stats.size_reused as f64 * 100. / upload_stats.size as f64;
 365             let reused: HumanByte = upload_stats.size_reused.into();
 366             println!(
 367                 "{}: backup was done incrementally, reused {} ({:.1}%)",
 368                 archive, reused, reused_percent
 369             );
 370         }
 371         if self.verbose && upload_stats.chunk_count > 0 {
 372             println!(
 373                 "{}: Reused {} from {} chunks.",
 374                 archive, upload_stats.chunk_reused, upload_stats.chunk_count
 375             );
 376             println!(
 377                 "{}: Average chunk size was {}.",
 378                 archive,
 379                 HumanByte::from(upload_stats.size / upload_stats.chunk_count)
 380             );
 381             println!(
 382                 "{}: Average time per request: {} microseconds.",
 383                 archive,
 384                 (upload_stats.duration.as_micros()) / (upload_stats.chunk_count as u128)
 385             );
 386         }
 387
 388         let param = json!({
 389             "wid": wid ,
 390             "chunk-count": upload_stats.chunk_count,
 391             "size": upload_stats.size,
 392             "csum": proxmox::tools::digest_to_hex(&upload_stats.csum),
 393         });
 394         let _value = self.h2.post(&close_path, Some(param)).await?;
 395         Ok(BackupStats {
 396             size: upload_stats.size as u64,
 397             csum: upload_stats.csum,
 398         })
 399     }
 400
 401     fn response_queue(
 402         verbose: bool,
 403     ) -> (
 404         mpsc::Sender<h2::client::ResponseFuture>,
 405         oneshot::Receiver<Result<(), Error>>,
 406     ) {
 407         let (verify_queue_tx, verify_queue_rx) = mpsc::channel(100);
 408         let (verify_result_tx, verify_result_rx) = oneshot::channel();
 409
 410         // FIXME: check if this works as expected as replacement for the combinator below?
 411         // tokio::spawn(async move {
 412         //     let result: Result<(), Error> = (async move {
 413         //         while let Some(response) = verify_queue_rx.recv().await {
 414         //             match H2Client::h2api_response(response.await?).await {
 415         //                 Ok(result) => println!("RESPONSE: {:?}", result),
 416         //                 Err(err) => bail!("pipelined request failed: {}", err),
 417         //             }
 418         //         }
 419         //         Ok(())
 420         //     }).await;
 421         //     let _ignore_closed_channel = verify_result_tx.send(result);
 422         // });
 423         // old code for reference?
 424         tokio::spawn(
 425             ReceiverStream::new(verify_queue_rx)
 426                 .map(Ok::<_, Error>)
 427                 .try_for_each(move |response: h2::client::ResponseFuture| {
 428                     response
 429                         .map_err(Error::from)
 430                         .and_then(H2Client::h2api_response)
 431                         .map_ok(move |result| {
 432                             if verbose {
 433                                 println!("RESPONSE: {:?}", result)
 434                             }
 435                         })
 436                         .map_err(|err| format_err!("pipelined request failed: {}", err))
 437                 })
 438                 .map(|result| {
 439                     let _ignore_closed_channel = verify_result_tx.send(result);
 440                 }),
 441         );
 442
 443         (verify_queue_tx, verify_result_rx)
 444     }
 445
 446     fn append_chunk_queue(
 447         h2: H2Client,
 448         wid: u64,
 449         path: String,
 450         verbose: bool,
 451     ) -> (UploadQueueSender, UploadResultReceiver) {
 452         let (verify_queue_tx, verify_queue_rx) = mpsc::channel(64);
 453         let (verify_result_tx, verify_result_rx) = oneshot::channel();
 454
 455         // FIXME: async-block-ify this code!
 456         tokio::spawn(
 457             ReceiverStream::new(verify_queue_rx)
 458                 .map(Ok::<_, Error>)
 459                 .and_then(move |(merged_chunk_info, response): (MergedChunkInfo, Option<h2::client::ResponseFuture>)| {
 460                     match (response, merged_chunk_info) {
 461                         (Some(response), MergedChunkInfo::Known(list)) => {
 462                             Either::Left(
 463                                 response
 464                                     .map_err(Error::from)
 465                                     .and_then(H2Client::h2api_response)
 466                                     .and_then(move |_result| {
 467                                         future::ok(MergedChunkInfo::Known(list))
 468                                     })
 469                             )
 470                         }
 471                         (None, MergedChunkInfo::Known(list)) => {
 472                             Either::Right(future::ok(MergedChunkInfo::Known(list)))
 473                         }
 474                         _ => unreachable!(),
 475                     }
 476                 })
 477                 .merge_known_chunks()
 478                 .and_then(move |merged_chunk_info| {
 479                     match merged_chunk_info {
 480                         MergedChunkInfo::Known(chunk_list) => {
 481                             let mut digest_list = vec![];
 482                             let mut offset_list = vec![];
 483                             for (offset, digest) in chunk_list {
 484                                 digest_list.push(digest_to_hex(&digest));
 485                                 offset_list.push(offset);
 486                             }
 487                             if verbose { println!("append chunks list len ({})", digest_list.len()); }
 488                             let param = json!({ "wid": wid, "digest-list": digest_list, "offset-list": offset_list });
 489                             let request = H2Client::request_builder("localhost", "PUT", &path, None, Some("application/json")).unwrap();
 490                             let param_data = bytes::Bytes::from(param.to_string().into_bytes());
 491                             let upload_data = Some(param_data);
 492                             h2.send_request(request, upload_data)
 493                                 .and_then(move |response| {
 494                                     response
 495                                         .map_err(Error::from)
 496                                         .and_then(H2Client::h2api_response)
 497                                         .map_ok(|_| ())
 498                                 })
 499                                 .map_err(|err| format_err!("pipelined request failed: {}", err))
 500                         }
 501                         _ => unreachable!(),
 502                     }
 503                 })
 504                 .try_for_each(|_| future::ok(()))
 505                 .map(|result| {
 506                       let _ignore_closed_channel = verify_result_tx.send(result);
 507                 })
 508         );
 509
 510         (verify_queue_tx, verify_result_rx)
 511     }
 512
 513     pub async fn download_previous_fixed_index(
 514         &self,
 515         archive_name: &str,
 516         manifest: &BackupManifest,
 517         known_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
 518     ) -> Result<FixedIndexReader, Error> {
 519         let mut tmpfile = std::fs::OpenOptions::new()
 520             .write(true)
 521             .read(true)
 522             .custom_flags(libc::O_TMPFILE)
 523             .open("/tmp")?;
 524
 525         let param = json!({ "archive-name": archive_name });
 526         self.h2
 527             .download("previous", Some(param), &mut tmpfile)
 528             .await?;
 529
 530         let index = FixedIndexReader::new(tmpfile).map_err(|err| {
 531             format_err!("unable to read fixed index '{}' - {}", archive_name, err)
 532         })?;
 533         // Note: do not use values stored in index (not trusted) - instead, computed them again
 534         let (csum, size) = index.compute_csum();
 535         manifest.verify_file(archive_name, &csum, size)?;
 536
 537         // add index chunks to known chunks
 538         let mut known_chunks = known_chunks.lock().unwrap();
 539         for i in 0..index.index_count() {
 540             known_chunks.insert(*index.index_digest(i).unwrap());
 541         }
 542
 543         if self.verbose {
 544             println!(
 545                 "{}: known chunks list length is {}",
 546                 archive_name,
 547                 index.index_count()
 548             );
 549         }
 550
 551         Ok(index)
 552     }
 553
 554     pub async fn download_previous_dynamic_index(
 555         &self,
 556         archive_name: &str,
 557         manifest: &BackupManifest,
 558         known_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
 559     ) -> Result<DynamicIndexReader, Error> {
 560         let mut tmpfile = std::fs::OpenOptions::new()
 561             .write(true)
 562             .read(true)
 563             .custom_flags(libc::O_TMPFILE)
 564             .open("/tmp")?;
 565
 566         let param = json!({ "archive-name": archive_name });
 567         self.h2
 568             .download("previous", Some(param), &mut tmpfile)
 569             .await?;
 570
 571         let index = DynamicIndexReader::new(tmpfile).map_err(|err| {
 572             format_err!("unable to read dynmamic index '{}' - {}", archive_name, err)
 573         })?;
 574         // Note: do not use values stored in index (not trusted) - instead, computed them again
 575         let (csum, size) = index.compute_csum();
 576         manifest.verify_file(archive_name, &csum, size)?;
 577
 578         // add index chunks to known chunks
 579         let mut known_chunks = known_chunks.lock().unwrap();
 580         for i in 0..index.index_count() {
 581             known_chunks.insert(*index.index_digest(i).unwrap());
 582         }
 583
 584         if self.verbose {
 585             println!(
 586                 "{}: known chunks list length is {}",
 587                 archive_name,
 588                 index.index_count()
 589             );
 590         }
 591
 592         Ok(index)
 593     }
 594
 595     /// Retrieve backup time of last backup
 596     pub async fn previous_backup_time(&self) -> Result<Option<i64>, Error> {
 597         let data = self.h2.get("previous_backup_time", None).await?;
 598         serde_json::from_value(data).map_err(|err| {
 599             format_err!(
 600                 "Failed to parse backup time value returned by server - {}",
 601                 err
 602             )
 603         })
 604     }
 605
 606     /// Download backup manifest (index.json) of last backup
 607     pub async fn download_previous_manifest(&self) -> Result<BackupManifest, Error> {
 608         let mut raw_data = Vec::with_capacity(64 * 1024);
 609
 610         let param = json!({ "archive-name": MANIFEST_BLOB_NAME });
 611         self.h2
 612             .download("previous", Some(param), &mut raw_data)
 613             .await?;
 614
 615         let blob = DataBlob::load_from_reader(&mut &raw_data[..])?;
 616         // no expected digest available
 617         let data = blob.decode(self.crypt_config.as_ref().map(Arc::as_ref), None)?;
 618
 619         let manifest =
 620             BackupManifest::from_data(&data[..], self.crypt_config.as_ref().map(Arc::as_ref))?;
 621
 622         Ok(manifest)
 623     }
 624
 625     // We have no `self` here for `h2` and `verbose`, the only other arg "common" with 1 other
 626     // function in the same path is `wid`, so those 3 could be in a struct, but there's no real use
 627     // since this is a private method.
 628     #[allow(clippy::too_many_arguments)]
 629     fn upload_chunk_info_stream(
 630         h2: H2Client,
 631         wid: u64,
 632         stream: impl Stream<Item = Result<bytes::BytesMut, Error>>,
 633         prefix: &str,
 634         known_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
 635         crypt_config: Option<Arc<CryptConfig>>,
 636         compress: bool,
 637         verbose: bool,
 638     ) -> impl Future<Output = Result<UploadStats, Error>> {
 639         let total_chunks = Arc::new(AtomicUsize::new(0));
 640         let total_chunks2 = total_chunks.clone();
 641         let known_chunk_count = Arc::new(AtomicUsize::new(0));
 642         let known_chunk_count2 = known_chunk_count.clone();
 643
 644         let stream_len = Arc::new(AtomicUsize::new(0));
 645         let stream_len2 = stream_len.clone();
 646         let compressed_stream_len = Arc::new(AtomicU64::new(0));
 647         let compressed_stream_len2 = compressed_stream_len.clone();
 648         let reused_len = Arc::new(AtomicUsize::new(0));
 649         let reused_len2 = reused_len.clone();
 650
 651         let append_chunk_path = format!("{}_index", prefix);
 652         let upload_chunk_path = format!("{}_chunk", prefix);
 653         let is_fixed_chunk_size = prefix == "fixed";
 654
 655         let (upload_queue, upload_result) =
 656             Self::append_chunk_queue(h2.clone(), wid, append_chunk_path, verbose);
 657
 658         let start_time = std::time::Instant::now();
 659
 660         let index_csum = Arc::new(Mutex::new(Some(openssl::sha::Sha256::new())));
 661         let index_csum_2 = index_csum.clone();
 662
 663         stream
 664             .and_then(move |data| {
 665                 let chunk_len = data.len();
 666
 667                 total_chunks.fetch_add(1, Ordering::SeqCst);
 668                 let offset = stream_len.fetch_add(chunk_len, Ordering::SeqCst) as u64;
 669
 670                 let mut chunk_builder = DataChunkBuilder::new(data.as_ref()).compress(compress);
 671
 672                 if let Some(ref crypt_config) = crypt_config {
 673                     chunk_builder = chunk_builder.crypt_config(crypt_config);
 674                 }
 675
 676                 let mut known_chunks = known_chunks.lock().unwrap();
 677                 let digest = chunk_builder.digest();
 678
 679                 let mut guard = index_csum.lock().unwrap();
 680                 let csum = guard.as_mut().unwrap();
 681
 682                 let chunk_end = offset + chunk_len as u64;
 683
 684                 if !is_fixed_chunk_size {
 685                     csum.update(&chunk_end.to_le_bytes());
 686                 }
 687                 csum.update(digest);
 688
 689                 let chunk_is_known = known_chunks.contains(digest);
 690                 if chunk_is_known {
 691                     known_chunk_count.fetch_add(1, Ordering::SeqCst);
 692                     reused_len.fetch_add(chunk_len, Ordering::SeqCst);
 693                     future::ok(MergedChunkInfo::Known(vec![(offset, *digest)]))
 694                 } else {
 695                     let compressed_stream_len2 = compressed_stream_len.clone();
 696                     known_chunks.insert(*digest);
 697                     future::ready(chunk_builder.build().map(move |(chunk, digest)| {
 698                         compressed_stream_len2.fetch_add(chunk.raw_size(), Ordering::SeqCst);
 699                         MergedChunkInfo::New(ChunkInfo {
 700                             chunk,
 701                             digest,
 702                             chunk_len: chunk_len as u64,
 703                             offset,
 704                         })
 705                     }))
 706                 }
 707             })
 708             .merge_known_chunks()
 709             .try_for_each(move |merged_chunk_info| {
 710                 let upload_queue = upload_queue.clone();
 711
 712                 if let MergedChunkInfo::New(chunk_info) = merged_chunk_info {
 713                     let offset = chunk_info.offset;
 714                     let digest = chunk_info.digest;
 715                     let digest_str = digest_to_hex(&digest);
 716
 717                     /* too verbose, needs finer verbosity setting granularity
 718                     if verbose {
 719                         println!("upload new chunk {} ({} bytes, offset {})", digest_str,
 720                                  chunk_info.chunk_len, offset);
 721                     }
 722                     */
 723
 724                     let chunk_data = chunk_info.chunk.into_inner();
 725                     let param = json!({
 726                         "wid": wid,
 727                         "digest": digest_str,
 728                         "size": chunk_info.chunk_len,
 729                         "encoded-size": chunk_data.len(),
 730                     });
 731
 732                     let ct = "application/octet-stream";
 733                     let request = H2Client::request_builder(
 734                         "localhost",
 735                         "POST",
 736                         &upload_chunk_path,
 737                         Some(param),
 738                         Some(ct),
 739                     )
 740                     .unwrap();
 741                     let upload_data = Some(bytes::Bytes::from(chunk_data));
 742
 743                     let new_info = MergedChunkInfo::Known(vec![(offset, digest)]);
 744
 745                     Either::Left(h2.send_request(request, upload_data).and_then(
 746                         move |response| async move {
 747                             upload_queue
 748                                 .send((new_info, Some(response)))
 749                                 .await
 750                                 .map_err(|err| {
 751                                     format_err!("failed to send to upload queue: {}", err)
 752                                 })
 753                         },
 754                     ))
 755                 } else {
 756                     Either::Right(async move {
 757                         upload_queue
 758                             .send((merged_chunk_info, None))
 759                             .await
 760                             .map_err(|err| format_err!("failed to send to upload queue: {}", err))
 761                     })
 762                 }
 763             })
 764             .then(move |result| async move { upload_result.await?.and(result) }.boxed())
 765             .and_then(move |_| {
 766                 let duration = start_time.elapsed();
 767                 let chunk_count = total_chunks2.load(Ordering::SeqCst);
 768                 let chunk_reused = known_chunk_count2.load(Ordering::SeqCst);
 769                 let size = stream_len2.load(Ordering::SeqCst);
 770                 let size_reused = reused_len2.load(Ordering::SeqCst);
 771                 let size_compressed = compressed_stream_len2.load(Ordering::SeqCst) as usize;
 772
 773                 let mut guard = index_csum_2.lock().unwrap();
 774                 let csum = guard.take().unwrap().finish();
 775
 776                 futures::future::ok(UploadStats {
 777                     chunk_count,
 778                     chunk_reused,
 779                     size,
 780                     size_reused,
 781                     size_compressed,
 782                     duration,
 783                     csum,
 784                 })
 785             })
 786     }
 787
 788     /// Upload speed test - prints result to stderr
 789     pub async fn upload_speedtest(&self, verbose: bool) -> Result<f64, Error> {
 790         let mut data = vec![];
 791         // generate pseudo random byte sequence
 792         for i in 0..1024 * 1024 {
 793             for j in 0..4 {
 794                 let byte = ((i >> (j << 3)) & 0xff) as u8;
 795                 data.push(byte);
 796             }
 797         }
 798
 799         let item_len = data.len();
 800
 801         let mut repeat = 0;
 802
 803         let (upload_queue, upload_result) = Self::response_queue(verbose);
 804
 805         let start_time = std::time::Instant::now();
 806
 807         loop {
 808             repeat += 1;
 809             if start_time.elapsed().as_secs() >= 5 {
 810                 break;
 811             }
 812
 813             if verbose {
 814                 eprintln!("send test data ({} bytes)", data.len());
 815             }
 816             let request =
 817                 H2Client::request_builder("localhost", "POST", "speedtest", None, None).unwrap();
 818             let request_future = self
 819                 .h2
 820                 .send_request(request, Some(bytes::Bytes::from(data.clone())))
 821                 .await?;
 822
 823             upload_queue.send(request_future).await?;
 824         }
 825
 826         drop(upload_queue); // close queue
 827
 828         let _ = upload_result.await?;
 829
 830         eprintln!(
 831             "Uploaded {} chunks in {} seconds.",
 832             repeat,
 833             start_time.elapsed().as_secs()
 834         );
 835         let speed = ((item_len * (repeat as usize)) as f64) / start_time.elapsed().as_secs_f64();
 836         eprintln!(
 837             "Time per request: {} microseconds.",
 838             (start_time.elapsed().as_micros()) / (repeat as u128)
 839         );
 840
 841         Ok(speed)
 842     }
 843 }