Add a blobstore benchmark tool

Summary:
We want to be able to benchmark blobstore stacks to allow us to compare blobstore stacks, and ensure that we're not regressing as we improve SQLblob to production state.

Use Criterion to benchmark a few basic workflows - starting with writes, but reads can be added later.

Reviewed By: ahornby

Differential Revision: D20720789

fbshipit-source-id: e8b10664a9d08a1aa7e646e1ebde251bec0db991

eden/mononoke/cmds/benchmark_storage_config/main.rs

@@ -0,0 +1,78 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * This software may be used and distributed according to the terms of the
+ * GNU General Public License version 2.
+ */
+
+#![deny(warnings)]
+
+use std::time::Duration;
+
+use clap::Arg;
+use criterion::Criterion;
+
+use blobstore_factory::make_blobstore;
+use cmdlib::args;
+use context::CoreContext;
+
+mod parallel_puts;
+mod single_puts;
+
+pub const KB: usize = 1024;
+pub const MB: usize = KB * 1024;
+const ARG_STORAGE_CONFIG_NAME: &'static str = "storage-config-name";
+
+#[fbinit::main]
+fn main(fb: fbinit::FacebookInit) {
+    let app = args::MononokeApp::new("benchmark_storage_config")
+        .with_advanced_args_hidden()
+        .with_all_repos()
+        .build()
+        .arg(
+            Arg::with_name(ARG_STORAGE_CONFIG_NAME)
+                .long(ARG_STORAGE_CONFIG_NAME)
+                .takes_value(true)
+                .required(true)
+                .help("the name of the storage config to benchmark"),
+        );
+    let matches = app.get_matches();
+    let logger = args::init_logging(fb, &matches);
+    args::init_cachelib(fb, &matches, None);
+
+    let storage_config = args::read_storage_configs(fb, &matches)
+        .expect("Could not read storage configs")
+        .remove(
+            matches
+                .value_of(ARG_STORAGE_CONFIG_NAME)
+                .expect("No storage config name"),
+        )
+        .expect("Storage config not found");
+    let mysql_options = args::parse_mysql_options(&matches);
+    let blobstore_options = args::parse_blobstore_options(&matches);
+    let mut runtime = args::init_runtime(&matches).expect("Cannot start tokio");
+    let ctx = CoreContext::new_with_logger(fb, logger.clone());
+
+    let blobstore = runtime
+        .block_on(make_blobstore(
+            fb,
+            storage_config.blobstore,
+            mysql_options,
+            blobstore_factory::ReadOnlyStorage(false),
+            blobstore_options,
+            logger,
+        ))
+        .expect("Could not make blobstore");
+
+    let mut criterion = Criterion::default()
+        .measurement_time(Duration::from_secs(60))
+        .sample_size(10)
+        .warm_up_time(Duration::from_secs(60));
+
+    // Tests are run from here
+    single_puts::benchmark(&mut criterion, ctx.clone(), blobstore.clone(), &mut runtime);
+    parallel_puts::benchmark(&mut criterion, ctx.clone(), blobstore.clone(), &mut runtime);
+
+    runtime.shutdown_on_idle();
+    criterion.final_summary();
+}

eden/mononoke/cmds/benchmark_storage_config/parallel_puts.rs

@@ -0,0 +1,73 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * This software may be used and distributed according to the terms of the
+ * GNU General Public License version 2.
+ */
+
+use std::sync::Arc;
+
+use criterion::{BenchmarkId, Criterion, Throughput};
+use futures::{
+    compat::Future01CompatExt,
+    stream::{FuturesUnordered, TryStreamExt},
+};
+use rand::{thread_rng, Rng, RngCore};
+use tokio_compat::runtime::Runtime;
+
+use blobstore::{Blobstore, BlobstoreBytes};
+use context::CoreContext;
+
+use crate::{KB, MB};
+
+pub fn benchmark(
+    c: &mut Criterion,
+    ctx: CoreContext,
+    blobstore: Arc<dyn Blobstore>,
+    runtime: &mut Runtime,
+) {
+    let mut group = c.benchmark_group("parallel_puts");
+
+    for size in [128, 16 * KB, 512 * KB, 8 * MB].iter() {
+        for concurrency in [4, 16, 256].iter() {
+            group.throughput(Throughput::Bytes(*size as u64 * *concurrency as u64));
+            group.bench_with_input(
+                BenchmarkId::from_parameter(format!("{} x{}", size, concurrency)),
+                size,
+                |b, &size| {
+                    let blocks: Vec<_> = std::iter::repeat(())
+                        .take(*concurrency)
+                        .map(|()| {
+                            let mut block = Vec::with_capacity(size);
+                            block.resize(size, 0u8);
+                            thread_rng().fill(&mut block as &mut [u8]);
+
+                            BlobstoreBytes::from_bytes(block)
+                        })
+                        .collect();
+                    let test = |ctx: CoreContext, blobstore: Arc<dyn Blobstore>| {
+                        let blocks = blocks.clone();
+                        async move {
+                            let futs: FuturesUnordered<_> = blocks
+                                .into_iter()
+                                .map(|block| {
+                                    let key = format!("benchmark.{:x}", thread_rng().next_u64());
+                                    blobstore.put(ctx.clone(), key, block).compat()
+                                })
+                                .collect();
+
+                            futs.try_for_each(|_| async move { Ok(()) })
+                                .await
+                                .expect("Puts failed");
+                        }
+                    };
+                    b.iter(|| {
+                        runtime
+                            .block_on_std(async { test(ctx.clone(), Arc::clone(&blobstore)).await })
+                    });
+                },
+            );
+        }
+    }
+    group.finish();
+}

eden/mononoke/cmds/benchmark_storage_config/single_puts.rs

@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * This software may be used and distributed according to the terms of the
+ * GNU General Public License version 2.
+ */
+
+use std::sync::Arc;
+
+use criterion::{BenchmarkId, Criterion, Throughput};
+use futures::compat::Future01CompatExt;
+use rand::{thread_rng, Rng, RngCore};
+use tokio_compat::runtime::Runtime;
+
+use blobstore::{Blobstore, BlobstoreBytes};
+use context::CoreContext;
+
+use crate::{KB, MB};
+
+pub fn benchmark(
+    c: &mut Criterion,
+    ctx: CoreContext,
+    blobstore: Arc<dyn Blobstore>,
+    runtime: &mut Runtime,
+) {
+    let mut group = c.benchmark_group("single_puts");
+
+    for size in [128, 16 * KB, 512 * KB, 8 * MB].iter() {
+        group.throughput(Throughput::Bytes(*size as u64));
+        group.bench_with_input(BenchmarkId::from_parameter(size), size, |b, &size| {
+            let mut block = Vec::with_capacity(size);
+            block.resize(size, 0u8);
+            thread_rng().fill(&mut block as &mut [u8]);
+
+            let block = BlobstoreBytes::from_bytes(block);
+            let test = |ctx, blobstore: Arc<dyn Blobstore>| {
+                let block = block.clone();
+                async move {
+                    let key = format!("benchmark.{:x}", thread_rng().next_u64());
+                    blobstore
+                        .put(ctx, key, block)
+                        .compat()
+                        .await
+                        .expect("Put failed");
+                }
+            };
+            b.iter(|| {
+                runtime.block_on_std(async { test(ctx.clone(), Arc::clone(&blobstore)).await })
+            });
+        });
+    }
+    group.finish();
+}