Add timing, rename some spans

Author: gefjon

modules/tpcc/src/load.rs

@@ -372,7 +372,7 @@ fn run_chunk(ctx: &ReducerContext, config: &TpccLoadConfig, job: &TpccLoadJob) -
 }
 
 fn load_item_chunk(ctx: &ReducerContext, config: &TpccLoadConfig, job: &TpccLoadJob) -> Result<ChunkAdvance, String> {
-    let _timer = LogStopwatch::new("load_items");
+    let _timer = LogStopwatch::new("load_item_chunk");
     if job.next_item_id == 0 || job.next_item_id > ITEMS {
         return Err(format!("invalid item cursor {}", job.next_item_id));
     }
@@ -402,7 +402,7 @@ fn load_warehouse_district_chunk(
     config: &TpccLoadConfig,
     job: &TpccLoadJob,
 ) -> Result<ChunkAdvance, String> {
-    let _timer = LogStopwatch::new("load_warehouses_districts");
+    let _timer = LogStopwatch::new("load_warehouses_district_chunk");
     let end_warehouse = warehouse_end(config.database_number, config.warehouses_per_database);
     if job.next_warehouse_id < warehouse_start(config.database_number, config.warehouses_per_database)
         || job.next_warehouse_id > end_warehouse
@@ -439,7 +439,7 @@ fn load_warehouse_district_chunk(
 }
 
 fn load_stock_chunk(ctx: &ReducerContext, config: &TpccLoadConfig, job: &TpccLoadJob) -> Result<ChunkAdvance, String> {
-    let _timer = LogStopwatch::new("load_stocks");
+    let _timer = LogStopwatch::new("load_stock_chunk");
     let start_warehouse = warehouse_start(config.database_number, config.warehouses_per_database);
     let end_warehouse = warehouse_end(config.database_number, config.warehouses_per_database);
     if job.next_warehouse_id < start_warehouse || job.next_warehouse_id > end_warehouse {
@@ -480,7 +480,7 @@ fn load_customer_history_chunk(
     config: &TpccLoadConfig,
     job: &TpccLoadJob,
 ) -> Result<ChunkAdvance, String> {
-    let _timer = LogStopwatch::new("load_customers_history");
+    let _timer = LogStopwatch::new("load_customer_history_chunk");
     let start_warehouse = warehouse_start(config.database_number, config.warehouses_per_database);
     let end_warehouse = warehouse_end(config.database_number, config.warehouses_per_database);
     if job.next_warehouse_id < start_warehouse || job.next_warehouse_id > end_warehouse {
@@ -530,7 +530,7 @@ fn load_customer_history_chunk(
 }
 
 fn load_order_chunk(ctx: &ReducerContext, config: &TpccLoadConfig, job: &TpccLoadJob) -> Result<ChunkAdvance, String> {
-    let _timer = LogStopwatch::new("load_orders");
+    let _timer = LogStopwatch::new("load_order_chunk");
     let start_warehouse = warehouse_start(config.database_number, config.warehouses_per_database);
     let end_warehouse = warehouse_end(config.database_number, config.warehouses_per_database);
     if job.next_warehouse_id < start_warehouse || job.next_warehouse_id > end_warehouse {

tools/tpcc-runner/src/loader.rs

@@ -60,33 +60,58 @@ fn database_number_chunks(num_databases: u16, parallelism: usize) -> Vec<Vec<u16
         .collect()
 }
 
+macro_rules! time {
+    ($span_name:literal { $($body:tt)*}) => {{
+        let before = std::time::Instant::now();
+        log::info!("Span {} starting at {:?}", $span_name, before);
+        let res = (|| Ok::<_, anyhow::Error>({ $($body)* }))();
+        let elapsed = before.elapsed();
+        log::info!("Span {} ended after {:?}", $span_name, elapsed);
+        res?
+    }}
+}
+
 fn run_one_database(config: &LoadConfig, database_number: u16, topology: &DatabaseTopology) -> Result<()> {
-    let database_identity = topology.identity_for_database_number(database_number)?;
-    log::info!(
-        "starting tpcc load into {} / {} with {} warehouse(s)",
-        config.connection.uri,
-        database_identity,
-        config.warehouses_per_database
-    );
+    time!("run_one_database" {
+        let database_identity = topology.identity_for_database_number(database_number)?;
+        log::info!(
+            "starting tpcc load into {} / {} with {} warehouse(s)",
+            config.connection.uri,
+            database_identity,
+            config.warehouses_per_database
+        );
+
+        let mut client = ModuleClient::connect(&config.connection, database_identity)?;
+        client.subscribe_load_state()?;
+
+        time!("reset" {
+            if config.reset {
+                client.reset_tpcc().context("failed to reset tpcc data")?;
+            }
+        });
 
-    let mut client = ModuleClient::connect(&config.connection, database_identity)?;
-    client.subscribe_load_state()?;
+        let request = time!("build_load_request" {
+            build_load_request(config, database_number, topology)?
+        });
+        time!("configure_tpcc_load" {client
+                                     .configure_tpcc_load(request)
+                                     .context("failed to configure tpcc load")})?;
 
-    if config.reset {
-        client.reset_tpcc().context("failed to reset tpcc data")?;
-    }
+        time!("start_tpcc_load" {
+            client.start_tpcc_load().context("failed to start tpcc load")?
+        });
 
-    let request = build_load_request(config, database_number, topology)?;
-    client
-        .configure_tpcc_load(request)
-        .context("failed to configure tpcc load")?;
-    client.start_tpcc_load().context("failed to start tpcc load")?;
+        time!("wait_for_load_completion" {
+            wait_for_load_completion(&client, database_identity)?
+        });
 
-    wait_for_load_completion(&client, database_identity)?;
-    client.shutdown();
+        time!("shutdown" {
+            client.shutdown()
+        });
 
-    log::info!("tpcc load for database {database_identity} finished");
-    Ok(())
+        log::info!("tpcc load for database {database_identity} finished");
+        Ok(())
+    })
 }
 
 fn build_load_request(