treewide: Favor native async traits

Native async traits were stabilized as of Rust 1.75, so we no longer need
the async_trait crate in many situations. This commit replaces the 3rd
party crate with the native version everywhere we can. The areas of the
code that still require the 3rd party crate include:

- Any trait that is used as a trait object (this is not supported
  natively by Rust yet)
- Certain traits that returned lifetime errors when attempting to remove
  the `#[async_trait]` macro (these errors included a message that said
  the error was a known limitation and would be removed in the future)
- The trait in proptest-stateful, whose interface I didn't want to change
  without further discussion, since it's a publicly-available trait

Change-Id: I5c761c075966e4fcebbb6d4955608107cf871b7c

Cargo.toml

@@ -113,7 +113,7 @@ derive_more = "0.99.17"
 diff = "0.1.10"
 enum-display-derive = "0.1.1"
 enum-kinds = "0.5.1"
-enum_dispatch = "0.3.7"
+enum_dispatch = "0.3.13"
 envy = "0.4"
 fail = "0.5.0"
 fallible-iterator = "0.2.0"

benchmarks/src/benchmark.rs

@@ -14,7 +14,6 @@ use std::collections::HashMap;
 use std::str::FromStr;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseConnection, DatabaseType, DatabaseURL};
 use enum_dispatch::enum_dispatch;
@@ -265,7 +264,8 @@ impl BenchmarkOutput {
 
 /// The set of control functions needed to execute the benchmark in
 /// the `BenchmarkRunner`.
-#[async_trait]
+// Only used internally
+#[allow(async_fn_in_trait)]
 #[enum_dispatch]
 pub trait BenchmarkControl {
     /// Any code required to perform setup of the benchmark goes here. This

benchmarks/src/cache_hit_benchmark.rs

@@ -4,7 +4,6 @@ use std::str::FromStr;
 use std::time::Instant;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseConnection, DatabaseURL, QueryableConnection};
 use metrics::Unit;
@@ -36,7 +35,6 @@ pub struct CacheHitBenchmark {
     num_cache_misses: u32,
 }
 
-#[async_trait]
 impl BenchmarkControl for CacheHitBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator

benchmarks/src/eviction_benchmark.rs

@@ -17,7 +17,6 @@ use std::sync::atomic::{AtomicBool, AtomicU64};
 use std::time::{Duration, Instant};
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseURL, QueryableConnection};
 use metrics::Unit;
@@ -83,7 +82,6 @@ pub struct EvictionBenchmarkParams {
     target_hit_rate: f64,
 }
 
-#[async_trait]
 impl BenchmarkControl for EvictionBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator
@@ -335,7 +333,6 @@ async fn metrics_task(
     }
 }
 
-#[async_trait]
 impl MultithreadBenchmark for EvictionBenchmark {
     type BenchmarkResult = EvictionBenchmarkResultBatch;
     type Parameters = EvictionBenchmarkParams;

benchmarks/src/lib.rs

@@ -23,7 +23,6 @@
 //!     row_count: u32,
 //! }
 //!
-//! #[async_trait::async_trait]
 //! impl BenchmarkControl for MyBenchmark {
 //!     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
 //!         let mut conn = deployment.connect_to_setup().await?;

benchmarks/src/migration_benchmark.rs

@@ -4,7 +4,6 @@ use std::str::FromStr;
 use std::time::Instant;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::DatabaseURL;
 use metrics::Unit;
@@ -33,7 +32,6 @@ pub struct MigrationBenchmark {
     num_migrations: u32,
 }
 
-#[async_trait]
 impl BenchmarkControl for MigrationBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator

benchmarks/src/query_benchmark.rs

@@ -8,7 +8,6 @@ use std::str::FromStr;
 use std::time::{Duration, Instant};
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseURL, QueryableConnection};
 use metrics::Unit;
@@ -60,7 +59,6 @@ pub struct QueryBenchmarkThreadParams {
     query: ArbitraryQueryParameters,
 }
 
-#[async_trait]
 impl BenchmarkControl for QueryBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator
@@ -155,7 +153,6 @@ impl QueryBenchmarkResultBatch {
     }
 }
 
-#[async_trait]
 impl MultithreadBenchmark for QueryBenchmark {
     type BenchmarkResult = QueryBenchmarkResultBatch;
     type Parameters = QueryBenchmarkThreadParams;

benchmarks/src/read_write_benchmark.rs

@@ -5,7 +5,6 @@ use std::str::FromStr;
 use std::time::{Duration, Instant};
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseURL, QueryableConnection};
 use metrics::Unit;
@@ -84,7 +83,6 @@ pub struct ReadWriteBenchmarkThreadParams {
     upstream_conn_str: String,
 }
 
-#[async_trait]
 impl BenchmarkControl for ReadWriteBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator
@@ -190,7 +188,6 @@ impl ReadWriteBenchmarkResultBatch {
     }
 }
 
-#[async_trait]
 impl MultithreadBenchmark for ReadWriteBenchmark {
     type BenchmarkResult = ReadWriteBenchmarkResultBatch;
     type Parameters = ReadWriteBenchmarkThreadParams;

benchmarks/src/scale_connections.rs

@@ -3,7 +3,6 @@ use std::str::FromStr;
 use std::time::Instant;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::DatabaseURL;
 use metrics::Unit;
@@ -27,7 +26,6 @@ pub struct ScaleConnections {
     parallel: bool,
 }
 
-#[async_trait]
 impl BenchmarkControl for ScaleConnections {
     async fn setup(&self, _: &DeploymentParameters) -> Result<()> {
         Ok(())

benchmarks/src/scale_views.rs

@@ -11,7 +11,6 @@ use std::str::FromStr;
 use std::time::Instant;
 
 use anyhow::{bail, Result};
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseURL, QueryableConnection};
 use itertools::Itertools;
@@ -53,7 +52,6 @@ fn get_columns(num_views: usize, param_count: usize) -> Vec<String> {
     (0..num_columns).map(|i| format!("c{}", i)).collect()
 }
 
-#[async_trait]
 impl BenchmarkControl for ScaleViews {
     /// Creates a table with enough columns that we can create `num_views` off a
     /// combination of the columns.

benchmarks/src/single_query_benchmark.rs

@@ -6,7 +6,6 @@ use std::str::FromStr;
 use std::time::Instant;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseConnection, DatabaseURL, QueryableConnection};
 use metrics::Unit;
@@ -37,7 +36,6 @@ pub struct SingleQueryBenchmark {
     ad_hoc: bool,
 }
 
-#[async_trait]
 impl BenchmarkControl for SingleQueryBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator

benchmarks/src/template.rs

@@ -1,7 +1,6 @@
 use std::collections::HashMap;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use metrics::Unit;
 use serde::{Deserialize, Serialize};
@@ -16,7 +15,6 @@ pub struct Template {
     // Benchmark specific parameters go here.
 }
 
-#[async_trait]
 impl BenchmarkControl for Template {
     async fn setup(&self, _: &DeploymentParameters) -> Result<()> {
         // Any code required to setup the benchmark goes. This may include

benchmarks/src/utils/multi_thread.rs

@@ -1,7 +1,7 @@
+use std::future::Future;
 use std::time::Duration;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use futures::stream::futures_unordered::FuturesUnordered;
 use futures::StreamExt;
 use tokio::select;
@@ -14,26 +14,25 @@ use crate::benchmark::BenchmarkResults;
 /// A group of methods that facilitate executing a single benchmark from multiple
 /// threads. This should be used in conjunction with `run_multithread_benchmark`
 /// to spawn the threads to run the benchmark.
-#[async_trait]
 pub(crate) trait MultithreadBenchmark {
     /// The result messages passed to the result's thread via an UnboundedSender.
     type BenchmarkResult: Send;
     /// Thset of parameters used to initialize the benchmark threads.
     type Parameters: Clone;
     /// Process a batch of benchmark results collected over `interval`. This aggregates
     /// all updates send on the `sender` parameter fo `benchmark_thread`.
-    async fn handle_benchmark_results(
+    fn handle_benchmark_results(
         results: Vec<Self::BenchmarkResult>,
         interval: Duration,
         results: &mut BenchmarkResults,
-    ) -> Result<()>;
+    ) -> impl Future<Output = Result<()>> + Send;
 
     /// Benchmarking code that is initialized using `params` that sends `BenchmarkResult`
     /// to be batched along `sender`.
-    async fn benchmark_thread(
+    fn benchmark_thread(
         params: Self::Parameters,
         sender: UnboundedSender<Self::BenchmarkResult>,
-    ) -> Result<()>;
+    ) -> impl Future<Output = Result<()>> + Send;
 }
 
 /// Returns after `duration` if it is Some, otherwise, never returns. Useful

benchmarks/src/workload_emulator.rs

@@ -15,7 +15,6 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::{Parser, ValueEnum};
 use database_utils::{DatabaseConnection, DatabaseStatement, QueryableConnection};
 use metrics::Unit;
@@ -178,7 +177,6 @@ impl WorkloadResultBatch {
     }
 }
 
-#[async_trait]
 impl BenchmarkControl for WorkloadEmulator {
     async fn setup(&self, deployment: &DeploymentParameters) -> anyhow::Result<()> {
         if let Some(ref data_generator) = self.data_generator {
@@ -472,7 +470,6 @@ fn ttl_jitter(ttl_secs: u32) -> u32 {
     ttl_secs + (ttl_secs as f32 * variance) as u32
 }
 
-#[async_trait]
 impl MultithreadBenchmark for WorkloadEmulator {
     type BenchmarkResult = WorkloadResultBatch;
     type Parameters = WorkloadThreadParams;

benchmarks/src/write_benchmark.rs

@@ -6,7 +6,6 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use anyhow::{bail, Result};
-use async_trait::async_trait;
 use clap::{Parser, ValueHint};
 use database_utils::{DatabaseConnection, DatabaseError, DatabaseURL, QueryableConnection};
 use itertools::Itertools;
@@ -167,7 +166,6 @@ async fn create_indices(
     Ok(())
 }
 
-#[async_trait]
 impl BenchmarkControl for WriteBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         let mut conn = DatabaseURL::from_str(&deployment.target_conn_str)?
@@ -224,7 +222,6 @@ impl BenchmarkControl for WriteBenchmark {
     }
 }
 
-#[async_trait]
 impl MultithreadBenchmark for WriteBenchmark {
     type BenchmarkResult = u128;
     type Parameters = WriteBenchmarkThreadData;

benchmarks/src/write_latency_benchmark.rs

@@ -3,7 +3,6 @@ use std::convert::TryFrom;
 use std::time::Instant;
 
 use anyhow::{anyhow, bail, Result};
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::QueryableConnection;
 use metrics::Unit;
@@ -34,7 +33,6 @@ pub struct WriteLatencyBenchmark {
     updates: u32,
 }
 
-#[async_trait]
 impl BenchmarkControl for WriteLatencyBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator

database-utils/Cargo.toml

@@ -18,7 +18,6 @@ clap = { workspace = true, features = ["derive","env"] }
 serde = { workspace = true, features = ["derive"] }
 readyset-util = { path = "../readyset-util" }
 readyset-errors = { path = "../readyset-errors" }
-async-trait = { workspace = true }
 nom-sql = { path = "../nom-sql" }
 deadpool-postgres = { workspace = true }
 num_cpus = { workspace = true }