From cfb8c8008469b217450722b264cbb42a94823e02 Mon Sep 17 00:00:00 2001 From: Ethan Donowitz Date: Tue, 23 Apr 2024 17:19:02 -0400 Subject: [PATCH] treewide: Favor native async traits i am a commit message Change-Id: I5c761c075966e4fcebbb6d4955608107cf871b7c --- Cargo.lock | 10 +++------- Cargo.toml | 2 +- benchmarks/src/benchmark.rs | 4 ++-- benchmarks/src/cache_hit_benchmark.rs | 2 -- benchmarks/src/eviction_benchmark.rs | 3 --- benchmarks/src/lib.rs | 1 - benchmarks/src/migration_benchmark.rs | 2 -- benchmarks/src/query_benchmark.rs | 3 --- benchmarks/src/read_write_benchmark.rs | 3 --- benchmarks/src/scale_connections.rs | 2 -- benchmarks/src/scale_views.rs | 2 -- benchmarks/src/single_query_benchmark.rs | 2 -- benchmarks/src/template.rs | 2 -- benchmarks/src/utils/multi_thread.rs | 11 +++++------ benchmarks/src/workload_emulator.rs | 3 --- benchmarks/src/write_benchmark.rs | 3 --- benchmarks/src/write_latency_benchmark.rs | 2 -- database-utils/Cargo.toml | 1 - database-utils/src/connection.rs | 7 ++----- mysql-srv/Cargo.toml | 1 - mysql-srv/src/lib.rs | 5 ++--- mysql-srv/tests/main.rs | 6 ++---- psql-srv/Cargo.toml | 1 - psql-srv/src/lib.rs | 4 ++-- psql-srv/src/protocol.rs | 2 -- psql-srv/tests/authentication.rs | 2 -- psql-srv/tests/errors.rs | 2 -- psql-srv/tests/tls.rs | 2 -- readyset-adapter/src/upstream_database.rs | 4 ++-- readyset-mysql/src/backend.rs | 6 ++---- readyset-psql/benches/proxy.rs | 2 -- readyset-psql/src/backend.rs | 2 -- readyset/Cargo.toml | 1 - readyset/src/lib.rs | 13 ++++++++----- readyset/src/mysql.rs | 2 -- readyset/src/psql.rs | 2 -- 36 files changed, 31 insertions(+), 91 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1e9d5928c5..f843c22cd9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1252,7 +1252,6 @@ dependencies = [ name = "database-utils" version = "0.1.0" dependencies = [ - "async-trait", "clap 4.3.2", "deadpool-postgres", "derive_more", @@ -1585,14 +1584,14 @@ dependencies = [ [[package]] name = "enum_dispatch" -version = "0.3.8" +version = "0.3.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0eb359f1476bf611266ac1f5355bc14aeca37b299d0ebccc038ee7058891c9cb" +checksum = "aa18ce2bc66555b3218614519ac839ddb759a7d6720732f979ef8d13be147ecd" dependencies = [ "once_cell", "proc-macro2 1.0.69", "quote 1.0.33", - "syn 1.0.109", + "syn 2.0.39", ] [[package]] @@ -2923,7 +2922,6 @@ dependencies = [ name = "mysql-srv" version = "0.8.8" dependencies = [ - "async-trait", "byteorder", "chrono", "futures", @@ -4005,7 +4003,6 @@ dependencies = [ name = "psql-srv" version = "0.1.0" dependencies = [ - "async-trait", "base64 0.21.0", "bit-vec", "bytes", @@ -4276,7 +4273,6 @@ name = "readyset" version = "1.2.0" dependencies = [ "anyhow", - "async-trait", "chrono", "clap 4.3.2", "crossbeam-skiplist", diff --git a/Cargo.toml b/Cargo.toml index 4be2086470..4e98e58331 100644 --- a/Cargo.toml +++ b/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" diff --git a/benchmarks/src/benchmark.rs b/benchmarks/src/benchmark.rs index 10d5d155bf..7d04266aa6 100644 --- a/benchmarks/src/benchmark.rs +++ b/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 diff --git a/benchmarks/src/cache_hit_benchmark.rs b/benchmarks/src/cache_hit_benchmark.rs index a12b81f3ae..0400c4e066 100644 --- a/benchmarks/src/cache_hit_benchmark.rs +++ b/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 diff --git a/benchmarks/src/eviction_benchmark.rs b/benchmarks/src/eviction_benchmark.rs index 4a69fdbe43..a69d72d6d5 100644 --- a/benchmarks/src/eviction_benchmark.rs +++ b/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; diff --git a/benchmarks/src/lib.rs b/benchmarks/src/lib.rs index ebc5a8d419..be956c5e53 100644 --- a/benchmarks/src/lib.rs +++ b/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?; diff --git a/benchmarks/src/migration_benchmark.rs b/benchmarks/src/migration_benchmark.rs index c2cdd6c13d..cb94bfcb2f 100644 --- a/benchmarks/src/migration_benchmark.rs +++ b/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 diff --git a/benchmarks/src/query_benchmark.rs b/benchmarks/src/query_benchmark.rs index 8c9adfc441..68e5603895 100644 --- a/benchmarks/src/query_benchmark.rs +++ b/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; diff --git a/benchmarks/src/read_write_benchmark.rs b/benchmarks/src/read_write_benchmark.rs index a1e10ae3e4..2fa846962d 100644 --- a/benchmarks/src/read_write_benchmark.rs +++ b/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; diff --git a/benchmarks/src/scale_connections.rs b/benchmarks/src/scale_connections.rs index bc8e256996..0e567cf0b7 100644 --- a/benchmarks/src/scale_connections.rs +++ b/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(()) diff --git a/benchmarks/src/scale_views.rs b/benchmarks/src/scale_views.rs index 0a86c432e4..5e6a49953b 100644 --- a/benchmarks/src/scale_views.rs +++ b/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 { (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. diff --git a/benchmarks/src/single_query_benchmark.rs b/benchmarks/src/single_query_benchmark.rs index 9acfa1ce24..e5405fb315 100644 --- a/benchmarks/src/single_query_benchmark.rs +++ b/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 diff --git a/benchmarks/src/template.rs b/benchmarks/src/template.rs index d6695213ec..e2edce8faf 100644 --- a/benchmarks/src/template.rs +++ b/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 diff --git a/benchmarks/src/utils/multi_thread.rs b/benchmarks/src/utils/multi_thread.rs index 78a790f49e..7c35aa2d38 100644 --- a/benchmarks/src/utils/multi_thread.rs +++ b/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,7 +14,6 @@ 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; @@ -22,18 +21,18 @@ pub(crate) trait MultithreadBenchmark { 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, interval: Duration, results: &mut BenchmarkResults, - ) -> Result<()>; + ) -> impl Future> + 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, - ) -> Result<()>; + ) -> impl Future> + Send; } /// Returns after `duration` if it is Some, otherwise, never returns. Useful diff --git a/benchmarks/src/workload_emulator.rs b/benchmarks/src/workload_emulator.rs index 38958ac114..1f8e9836ac 100644 --- a/benchmarks/src/workload_emulator.rs +++ b/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; diff --git a/benchmarks/src/write_benchmark.rs b/benchmarks/src/write_benchmark.rs index 4454a5b33f..dedfdfa3b8 100644 --- a/benchmarks/src/write_benchmark.rs +++ b/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; diff --git a/benchmarks/src/write_latency_benchmark.rs b/benchmarks/src/write_latency_benchmark.rs index 5aec2b1157..b789a9bc5b 100644 --- a/benchmarks/src/write_latency_benchmark.rs +++ b/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 diff --git a/database-utils/Cargo.toml b/database-utils/Cargo.toml index faad81a580..2500fbe627 100644 --- a/database-utils/Cargo.toml +++ b/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 } diff --git a/database-utils/src/connection.rs b/database-utils/src/connection.rs index 160c908367..9bc3f89699 100644 --- a/database-utils/src/connection.rs +++ b/database-utils/src/connection.rs @@ -3,7 +3,6 @@ use std::hash::{Hash, Hasher}; use std::marker::{Send, Sync}; use std::str; -use async_trait::async_trait; use derive_more::From; use futures::TryStreamExt; use mysql::prelude::AsQuery; @@ -15,7 +14,8 @@ use {mysql_async as mysql, tokio_postgres as pgsql}; use crate::error::{ConnectionType, DatabaseError}; -#[async_trait] +// Only used internally +#[allow(async_fn_in_trait)] pub trait QueryableConnection: Send { /// Executes query_drop for either mysql or postgres, whichever is the underlying /// connection variant. @@ -121,7 +121,6 @@ pub enum DatabaseConnection { PostgreSQLPool(deadpool_postgres::Client), } -#[async_trait] impl QueryableConnection for DatabaseConnection { async fn query_drop(&mut self, stmt: Q) -> Result<(), DatabaseError> where @@ -316,7 +315,6 @@ impl DatabaseConnectionPool { } } -#[async_trait] impl QueryableConnection for DatabaseConnectionPool { async fn query_drop(&mut self, stmt: Q) -> Result<(), DatabaseError> where @@ -670,7 +668,6 @@ pub enum Transaction<'a> { PostgresPool(deadpool_postgres::Transaction<'a>), } -#[async_trait] impl<'a> QueryableConnection for Transaction<'a> { async fn query_drop(&mut self, stmt: Q) -> Result<(), DatabaseError> where diff --git a/mysql-srv/Cargo.toml b/mysql-srv/Cargo.toml index 41cba83cc1..340e1f0bc2 100644 --- a/mysql-srv/Cargo.toml +++ b/mysql-srv/Cargo.toml @@ -19,7 +19,6 @@ byteorder = { workspace = true } chrono = { workspace = true } time = { workspace = true } getrandom = { workspace = true } -async-trait = { workspace = true } tokio = { workspace = true, features = ["full"] } thiserror = { workspace = true } sha-1 = { workspace = true } diff --git a/mysql-srv/src/lib.rs b/mysql-srv/src/lib.rs index e449c6cafa..e6a8ec05e6 100644 --- a/mysql-srv/src/lib.rs +++ b/mysql-srv/src/lib.rs @@ -29,7 +29,6 @@ //! use tokio::io::AsyncWrite; //! //! struct Backend; -//! #[async_trait] //! impl MySqlShim for Backend { //! async fn on_prepare( //! &mut self, @@ -171,7 +170,6 @@ use std::collections::HashMap; use std::io; use std::sync::Arc; -use async_trait::async_trait; use constants::{CLIENT_PLUGIN_AUTH, CONNECT_WITH_DB, PROTOCOL_41, RESERVED, SECURE_CONNECTION}; use error::{other_error, OtherErrorKind}; use mysql_common::constants::CapabilityFlags; @@ -250,7 +248,8 @@ pub enum QueryResultsResponse { } /// Implementors of this trait can be used to drive a MySQL-compatible database backend. -#[async_trait] +// Only used internally +#[allow(async_fn_in_trait)] pub trait MySqlShim { /// Called when the client issues a request to prepare `query` for later execution. /// diff --git a/mysql-srv/tests/main.rs b/mysql-srv/tests/main.rs index a77ebb6862..b873b98bb9 100644 --- a/mysql-srv/tests/main.rs +++ b/mysql-srv/tests/main.rs @@ -14,7 +14,6 @@ use std::marker::PhantomData; use std::pin::Pin; use std::{io, net, thread}; -use async_trait::async_trait; use mysql::prelude::Queryable; use mysql::Row; use mysql_srv::{ @@ -38,7 +37,6 @@ struct TestingShim { _phantom: PhantomData, } -#[async_trait] impl MySqlShim for TestingShim where Q: for<'a> FnMut( @@ -118,7 +116,7 @@ where ) -> QueryResultsResponse { if query.starts_with("SELECT @@") || query.starts_with("select @@") { let var = &query.get(b"SELECT @@".len()..); - return match var { + match var { Some("max_allowed_packet") => { let cols = &[Column { table: String::new(), @@ -133,7 +131,7 @@ where QueryResultsResponse::IoResult(w.finish().await) } _ => QueryResultsResponse::IoResult(results.completed(0, 0, None).await), - }; + } } else { QueryResultsResponse::IoResult((self.on_q)(query, results).await) } diff --git a/psql-srv/Cargo.toml b/psql-srv/Cargo.toml index 93fee6ffbf..ca46a9e213 100644 --- a/psql-srv/Cargo.toml +++ b/psql-srv/Cargo.toml @@ -9,7 +9,6 @@ description = "Bindings for emulating a PostgreSQL server" readme = "README.md" [dependencies] -async-trait = { workspace = true } base64 = "0.21" bit-vec = { workspace = true, features = ["serde"] } bytes = { workspace = true } diff --git a/psql-srv/src/lib.rs b/psql-srv/src/lib.rs index f982b190ac..325435e47e 100644 --- a/psql-srv/src/lib.rs +++ b/psql-srv/src/lib.rs @@ -29,7 +29,6 @@ mod value; use std::collections::HashMap; use std::sync::Arc; -use async_trait::async_trait; use futures::Stream; use nom_sql::SqlIdentifier; use postgres::SimpleQueryMessage; @@ -62,7 +61,8 @@ pub enum Credentials<'a> { /// A trait for implementing a SQL backend that produces responses to SQL query statements. This /// trait is the primary interface for the `psql-srv` crate. -#[async_trait] +// Only used internally +#[allow(async_fn_in_trait)] pub trait PsqlBackend { /// An associated type representing a resultset returned by a SQL query, which can be iterated /// to produce `Self::Row`s. diff --git a/psql-srv/src/protocol.rs b/psql-srv/src/protocol.rs index 020589cd22..900988ef33 100644 --- a/psql-srv/src/protocol.rs +++ b/psql-srv/src/protocol.rs @@ -1076,7 +1076,6 @@ mod tests { use std::task::Poll; use std::{io, vec}; - use async_trait::async_trait; use bytes::BytesMut; use futures::task::Context; use futures::{stream, TryStreamExt}; @@ -1134,7 +1133,6 @@ mod tests { } } - #[async_trait] impl PsqlBackend for Backend { type Resultset = stream::Iter>>; diff --git a/psql-srv/tests/authentication.rs b/psql-srv/tests/authentication.rs index b74afcd8f2..4652a6ba8b 100644 --- a/psql-srv/tests/authentication.rs +++ b/psql-srv/tests/authentication.rs @@ -2,7 +2,6 @@ use std::collections::HashMap; use std::sync::Arc; use std::vec; -use async_trait::async_trait; use futures::stream; use postgres::config::{ChannelBinding, SslMode}; use postgres::error::SqlState; @@ -34,7 +33,6 @@ struct ScramSha256Backend { password: &'static str, } -#[async_trait] impl PsqlBackend for ScramSha256Backend { type Resultset = stream::Iter>>; diff --git a/psql-srv/tests/errors.rs b/psql-srv/tests/errors.rs index e028900eed..b1e983971f 100644 --- a/psql-srv/tests/errors.rs +++ b/psql-srv/tests/errors.rs @@ -1,7 +1,6 @@ use std::collections::HashMap; use std::vec; -use async_trait::async_trait; use futures::{stream, Future}; use postgres::NoTls; use postgres_protocol::Oid; @@ -27,7 +26,6 @@ enum ErrorPosition { #[derive(Clone, Copy)] struct ErrorBackend(ErrorPosition); -#[async_trait] impl PsqlBackend for ErrorBackend { type Resultset = stream::Iter>>; diff --git a/psql-srv/tests/tls.rs b/psql-srv/tests/tls.rs index 9e332f489e..ea320a0525 100644 --- a/psql-srv/tests/tls.rs +++ b/psql-srv/tests/tls.rs @@ -2,7 +2,6 @@ use std::collections::HashMap; use std::sync::Arc; use std::vec; -use async_trait::async_trait; use database_utils::{DatabaseURL, QueryableConnection}; use futures::stream; use postgres_protocol::Oid; @@ -28,7 +27,6 @@ impl TryFrom for psql_srv::PsqlValue { } } -#[async_trait] impl PsqlBackend for TestBackend { type Resultset = stream::Iter>>; diff --git a/readyset-adapter/src/upstream_database.rs b/readyset-adapter/src/upstream_database.rs index 9fe6e57eaf..d93d50a0bd 100644 --- a/readyset-adapter/src/upstream_database.rs +++ b/readyset-adapter/src/upstream_database.rs @@ -357,11 +357,11 @@ where self.upstream().await?.start_tx(stmt).await } - async fn commit<'a>(&'a mut self) -> Result, Self::Error> { + async fn commit(&mut self) -> Result, Self::Error> { self.upstream().await?.commit().await } - async fn rollback<'a>(&'a mut self) -> Result, Self::Error> { + async fn rollback(&mut self) -> Result, Self::Error> { self.upstream().await?.rollback().await } diff --git a/readyset-mysql/src/backend.rs b/readyset-mysql/src/backend.rs index 9660bcbc8c..1fce235982 100644 --- a/readyset-mysql/src/backend.rs +++ b/readyset-mysql/src/backend.rs @@ -5,7 +5,6 @@ use std::convert::TryFrom; use std::fmt::Formatter; use std::ops::{Deref, DerefMut}; -use async_trait::async_trait; use futures_util::StreamExt; use itertools::{izip, Itertools}; use mysql_async::consts::StatusFlags; @@ -498,7 +497,6 @@ where } } -#[async_trait] impl MySqlShim for Backend where W: AsyncWrite + Unpin + Send + 'static, @@ -600,7 +598,7 @@ where Err(e) => info.error(e.error_kind(), e.to_string().as_bytes()).await, }; - Ok(res?) + res } async fn on_execute( @@ -741,7 +739,7 @@ where } let query_result = self.query(query).await; - return handle_query_result(query_result, results).await; + handle_query_result(query_result, results).await } fn password_for_username(&self, username: &str) -> Option> { diff --git a/readyset-psql/benches/proxy.rs b/readyset-psql/benches/proxy.rs index 4afd5012e5..6f163b9b15 100644 --- a/readyset-psql/benches/proxy.rs +++ b/readyset-psql/benches/proxy.rs @@ -18,7 +18,6 @@ use std::pin::Pin; use std::task::{Context, Poll}; use std::{io, vec}; -use async_trait::async_trait; use criterion::{criterion_group, criterion_main, Criterion}; use futures::future::{try_select, Either}; use futures::stream::Peekable; @@ -128,7 +127,6 @@ impl Backend { } } -#[async_trait] impl PsqlBackend for Backend { type Resultset = ResultStream; diff --git a/readyset-psql/src/backend.rs b/readyset-psql/src/backend.rs index 35861f8b26..b81be12562 100644 --- a/readyset-psql/src/backend.rs +++ b/readyset-psql/src/backend.rs @@ -4,7 +4,6 @@ use std::ops::Deref; use std::str::FromStr; use std::sync::Arc; -use async_trait::async_trait; use clap::ValueEnum; use eui48::MacAddressFormat; use postgres_types::{Oid, Type}; @@ -121,7 +120,6 @@ impl Backend { } } -#[async_trait] impl ps::PsqlBackend for Backend { type Resultset = Resultset; diff --git a/readyset/Cargo.toml b/readyset/Cargo.toml index bee6d98a92..565640c46b 100644 --- a/readyset/Cargo.toml +++ b/readyset/Cargo.toml @@ -8,7 +8,6 @@ edition = "2021" [dependencies] anyhow = { workspace = true } -async-trait = { workspace = true } clap = { workspace = true, features = ["derive","env"] } futures-util = { workspace = true } fail = { workspace = true } diff --git a/readyset/src/lib.rs b/readyset/src/lib.rs index 8265092b8c..876b24808d 100644 --- a/readyset/src/lib.rs +++ b/readyset/src/lib.rs @@ -6,6 +6,7 @@ pub mod psql; mod query_logger; use std::collections::HashMap; use std::fs::remove_dir_all; +use std::future::Future; use std::io; use std::marker::Send; use std::net::{IpAddr, Ipv4Addr, SocketAddr, ToSocketAddrs}; @@ -16,7 +17,6 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime}; use anyhow::{anyhow, bail}; -use async_trait::async_trait; use clap::builder::NonEmptyStringValueParser; use clap::{ArgGroup, Parser, ValueEnum}; use crossbeam_skiplist::SkipSet; @@ -74,19 +74,22 @@ const UPSTREAM_CONNECTION_TIMEOUT: Duration = Duration::from_secs(5); /// Retry interval to use when attempting to connect to the upstream database const UPSTREAM_CONNECTION_RETRY_INTERVAL: Duration = Duration::from_secs(1); -#[async_trait] pub trait ConnectionHandler { type UpstreamDatabase: UpstreamDatabase; type Handler: QueryHandler; - async fn process_connection( + fn process_connection( &mut self, stream: net::TcpStream, backend: Backend, - ); + ) -> impl Future + Send; /// Return an immediate error to a newly-established connection, then immediately disconnect - async fn immediate_error(self, stream: net::TcpStream, error_message: String); + fn immediate_error( + self, + stream: net::TcpStream, + error_message: String, + ) -> impl Future + Send; } /// How to behave when receiving unsupported `SET` statements. diff --git a/readyset/src/mysql.rs b/readyset/src/mysql.rs index 60017f1af3..6438ecfbd6 100644 --- a/readyset/src/mysql.rs +++ b/readyset/src/mysql.rs @@ -1,4 +1,3 @@ -use async_trait::async_trait; use mysql_srv::MySqlIntermediary; use readyset_adapter::upstream_database::LazyUpstream; use readyset_mysql::{MySqlQueryHandler, MySqlUpstream}; @@ -13,7 +12,6 @@ pub struct MySqlHandler { pub enable_statement_logging: bool, } -#[async_trait] impl ConnectionHandler for MySqlHandler { type UpstreamDatabase = LazyUpstream; type Handler = MySqlQueryHandler; diff --git a/readyset/src/psql.rs b/readyset/src/psql.rs index fa4aff04d3..986a265052 100644 --- a/readyset/src/psql.rs +++ b/readyset/src/psql.rs @@ -1,7 +1,6 @@ use std::io::Read; use std::sync::Arc; -use async_trait::async_trait; use clap::Parser; use readyset_adapter::upstream_database::LazyUpstream; use readyset_errors::ReadySetResult; @@ -92,7 +91,6 @@ impl PsqlHandler { } } -#[async_trait] impl ConnectionHandler for PsqlHandler { type UpstreamDatabase = LazyUpstream; type Handler = PostgreSqlQueryHandler;