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