Move to local runtimes per connection

This avoids a bunch of context switches and cross-thread
synchronization, which ends up improving the performance of a simple
query by ~20%, from 252us to 216us.

Author: sfackler

postgres/Cargo.toml

@@ -10,16 +10,17 @@ readme = "../README.md"
 keywords = ["database", "postgres", "postgresql", "sql"]
 categories = ["database"]
 
+[[bench]]
+name = "bench"
+harness = false
+
 [package.metadata.docs.rs]
 all-features = true
 
 [badges]
 circle-ci = { repository = "sfackler/rust-postgres" }
 
 [features]
-default = ["runtime"]
-runtime = ["tokio-postgres/runtime", "tokio", "lazy_static", "log"]
-
 with-bit-vec-0_6 = ["tokio-postgres/with-bit-vec-0_6"]
 with-chrono-0_4 = ["tokio-postgres/with-chrono-0_4"]
 with-eui48-0_4 = ["tokio-postgres/with-eui48-0_4"]
@@ -32,11 +33,11 @@ with-uuid-0_8 = ["tokio-postgres/with-uuid-0_8"]
 bytes = "0.5"
 fallible-iterator = "0.2"
 futures = "0.3"
-tokio-postgres = { version = "=0.5.0-alpha.2", path = "../tokio-postgres", default-features = false }
+tokio-postgres = { version = "=0.5.0-alpha.2", path = "../tokio-postgres" }
 
-tokio = { version = "0.2", optional = true, features = ["rt-threaded"] }
-lazy_static = { version = "1.0", optional = true }
-log = { version = "0.4", optional = true }
+tokio = { version = "0.2", features = ["rt-core"] }
+log = "0.4"
 
 [dev-dependencies]
+criterion = "0.3"
 tokio = "0.2"

postgres/benches/bench.rs

@@ -0,0 +1,17 @@
+use criterion::{criterion_group, criterion_main, Criterion};
+use postgres::{Client, NoTls};
+
+// spawned: 249us 252us 255us
+// local: 214us 216us 219us
+fn query_prepared(c: &mut Criterion) {
+    let mut client = Client::connect("host=localhost port=5433 user=postgres", NoTls).unwrap();
+
+    let stmt = client.prepare("SELECT $1::INT8").unwrap();
+
+    c.bench_function("query_prepared", move |b| {
+        b.iter(|| client.query(&stmt, &[&1i64]).unwrap())
+    });
+}
+
+criterion_group!(group, query_prepared);
+criterion_main!(group);

postgres/src/client.rs

@@ -1,27 +1,25 @@
-#[cfg(feature = "runtime")]
-use crate::Config;
-use crate::{CopyInWriter, CopyOutReader, RowIter, Statement, ToStatement, Transaction};
-use futures::executor;
+use crate::{Config, CopyInWriter, CopyOutReader, RowIter, Statement, ToStatement, Transaction};
+use tokio::runtime::Runtime;
 use tokio_postgres::tls::{MakeTlsConnect, TlsConnect};
 use tokio_postgres::types::{ToSql, Type};
-#[cfg(feature = "runtime")]
-use tokio_postgres::Socket;
-use tokio_postgres::{Error, Row, SimpleQueryMessage};
+use tokio_postgres::{Error, Row, SimpleQueryMessage, Socket};
 
 /// A synchronous PostgreSQL client.
-///
-/// This is a lightweight wrapper over the asynchronous tokio_postgres `Client`.
-pub struct Client(tokio_postgres::Client);
+pub struct Client {
+    runtime: Runtime,
+    client: tokio_postgres::Client,
+}
 
 impl Client {
+    pub(crate) fn new(runtime: Runtime, client: tokio_postgres::Client) -> Client {
+        Client { runtime, client }
+    }
+
     /// A convenience function which parses a configuration string into a `Config` and then connects to the database.
     ///
     /// See the documentation for [`Config`] for information about the connection syntax.
     ///
-    /// Requires the `runtime` Cargo feature (enabled by default).
-    ///
     /// [`Config`]: config/struct.Config.html
-    #[cfg(feature = "runtime")]
     pub fn connect<T>(params: &str, tls_mode: T) -> Result<Client, Error>
     where
         T: MakeTlsConnect<Socket> + 'static + Send,
@@ -78,7 +76,7 @@ impl Client {
     where
         T: ?Sized + ToStatement,
     {
-        executor::block_on(self.0.execute(query, params))
+        self.runtime.block_on(self.client.execute(query, params))
     }
 
     /// Executes a statement, returning the resulting rows.
@@ -114,7 +112,7 @@ impl Client {
     where
         T: ?Sized + ToStatement,
     {
-        executor::block_on(self.0.query(query, params))
+        self.runtime.block_on(self.client.query(query, params))
     }
 
     /// Executes a statement which returns a single row, returning it.
@@ -151,7 +149,7 @@ impl Client {
     where
         T: ?Sized + ToStatement,
     {
-        executor::block_on(self.0.query_one(query, params))
+        self.runtime.block_on(self.client.query_one(query, params))
     }
 
     /// Executes a statement which returns zero or one rows, returning it.
@@ -197,7 +195,7 @@ impl Client {
     where
         T: ?Sized + ToStatement,
     {
-        executor::block_on(self.0.query_opt(query, params))
+        self.runtime.block_on(self.client.query_opt(query, params))
     }
 
     /// A maximally-flexible version of `query`.
@@ -235,8 +233,10 @@ impl Client {
         I: IntoIterator<Item = &'a dyn ToSql>,
         I::IntoIter: ExactSizeIterator,
     {
-        let stream = executor::block_on(self.0.query_raw(query, params))?;
-        Ok(RowIter::new(stream))
+        let stream = self
+            .runtime
+            .block_on(self.client.query_raw(query, params))?;
+        Ok(RowIter::new(&mut self.runtime, stream))
     }
 
     /// Creates a new prepared statement.
@@ -263,7 +263,7 @@ impl Client {
     /// # }
     /// ```
     pub fn prepare(&mut self, query: &str) -> Result<Statement, Error> {
-        executor::block_on(self.0.prepare(query))
+        self.runtime.block_on(self.client.prepare(query))
     }
 
     /// Like `prepare`, but allows the types of query parameters to be explicitly specified.
@@ -294,7 +294,8 @@ impl Client {
     /// # }
     /// ```
     pub fn prepare_typed(&mut self, query: &str, types: &[Type]) -> Result<Statement, Error> {
-        executor::block_on(self.0.prepare_typed(query, types))
+        self.runtime
+            .block_on(self.client.prepare_typed(query, types))
     }
 
     /// Executes a `COPY FROM STDIN` statement, returning the number of rows created.
@@ -327,8 +328,8 @@ impl Client {
     where
         T: ?Sized + ToStatement,
     {
-        let sink = executor::block_on(self.0.copy_in(query, params))?;
-        Ok(CopyInWriter::new(sink))
+        let sink = self.runtime.block_on(self.client.copy_in(query, params))?;
+        Ok(CopyInWriter::new(&mut self.runtime, sink))
     }
 
     /// Executes a `COPY TO STDOUT` statement, returning a reader of the resulting data.
@@ -358,8 +359,8 @@ impl Client {
     where
         T: ?Sized + ToStatement,
     {
-        let stream = executor::block_on(self.0.copy_out(query, params))?;
-        CopyOutReader::new(stream)
+        let stream = self.runtime.block_on(self.client.copy_out(query, params))?;
+        CopyOutReader::new(&mut self.runtime, stream)
     }
 
     /// Executes a sequence of SQL statements using the simple query protocol.
@@ -378,7 +379,7 @@ impl Client {
     /// functionality to safely imbed that data in the request. Do not form statements via string concatenation and pass
     /// them to this method!
     pub fn simple_query(&mut self, query: &str) -> Result<Vec<SimpleQueryMessage>, Error> {
-        executor::block_on(self.0.simple_query(query))
+        self.runtime.block_on(self.client.simple_query(query))
     }
 
     /// Executes a sequence of SQL statements using the simple query protocol.
@@ -392,7 +393,7 @@ impl Client {
     /// functionality to safely embed that data in the request. Do not form statements via string concatenation and pass
     /// them to this method!
     pub fn batch_execute(&mut self, query: &str) -> Result<(), Error> {
-        executor::block_on(self.0.batch_execute(query))
+        self.runtime.block_on(self.client.batch_execute(query))
     }
 
     /// Begins a new database transaction.
@@ -416,35 +417,14 @@ impl Client {
     /// # }
     /// ```
     pub fn transaction(&mut self) -> Result<Transaction<'_>, Error> {
-        let transaction = executor::block_on(self.0.transaction())?;
-        Ok(Transaction::new(transaction))
+        let transaction = self.runtime.block_on(self.client.transaction())?;
+        Ok(Transaction::new(&mut self.runtime, transaction))
     }
 
     /// Determines if the client's connection has already closed.
     ///
     /// If this returns `true`, the client is no longer usable.
     pub fn is_closed(&self) -> bool {
-        self.0.is_closed()
-    }
-
-    /// Returns a shared reference to the inner nonblocking client.
-    pub fn get_ref(&self) -> &tokio_postgres::Client {
-        &self.0
-    }
-
-    /// Returns a mutable reference to the inner nonblocking client.
-    pub fn get_mut(&mut self) -> &mut tokio_postgres::Client {
-        &mut self.0
-    }
-
-    /// Consumes the client, returning the inner nonblocking client.
-    pub fn into_inner(self) -> tokio_postgres::Client {
-        self.0
-    }
-}
-
-impl From<tokio_postgres::Client> for Client {
-    fn from(c: tokio_postgres::Client) -> Client {
-        Client(c)
+        self.client.is_closed()
     }
 }

postgres/src/config.rs

@@ -2,23 +2,19 @@
 //!
 //! Requires the `runtime` Cargo feature (enabled by default).
 
-use crate::{Client, RUNTIME};
-use futures::{executor, FutureExt};
+use crate::Client;
+use futures::FutureExt;
 use log::error;
 use std::fmt;
-use std::future::Future;
 use std::path::Path;
-use std::pin::Pin;
 use std::str::FromStr;
-use std::sync::{mpsc, Arc};
 use std::time::Duration;
+use tokio::runtime;
 #[doc(inline)]
 pub use tokio_postgres::config::{ChannelBinding, SslMode, TargetSessionAttrs};
 use tokio_postgres::tls::{MakeTlsConnect, TlsConnect};
 use tokio_postgres::{Error, Socket};
 
-type Spawn = dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>) + Sync + Send;
-
 /// Connection configuration.
 ///
 /// Configuration can be parsed from libpq-style connection strings. These strings come in two formats:
@@ -95,7 +91,6 @@ type Spawn = dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>) + Sync + Send;
 #[derive(Clone)]
 pub struct Config {
     config: tokio_postgres::Config,
-    spawner: Option<Arc<Spawn>>,
 }
 
 impl fmt::Debug for Config {
@@ -117,7 +112,6 @@ impl Config {
     pub fn new() -> Config {
         Config {
             config: tokio_postgres::Config::new(),
-            spawner: None,
         }
     }
 
@@ -242,17 +236,6 @@ impl Config {
         self
     }
 
-    /// Sets the spawner used to run the connection futures.
-    ///
-    /// Defaults to a postgres-specific tokio `Runtime`.
-    pub fn spawner<F>(&mut self, spawn: F) -> &mut Config
-    where
-        F: Fn(Pin<Box<dyn Future<Output = ()> + Send>>) + 'static + Sync + Send,
-    {
-        self.spawner = Some(Arc::new(spawn));
-        self
-    }
-
     /// Opens a connection to a PostgreSQL database.
     pub fn connect<T>(&self, tls: T) -> Result<Client, Error>
     where
@@ -261,38 +244,23 @@ impl Config {
         T::Stream: Send,
         <T::TlsConnect as TlsConnect<Socket>>::Future: Send,
     {
-        let (client, connection) = match &self.spawner {
-            Some(spawn) => {
-                let (tx, rx) = mpsc::channel();
-                let config = self.config.clone();
-                let connect = async move {
-                    let r = config.connect(tls).await;
-                    let _ = tx.send(r);
-                };
-                spawn(Box::pin(connect));
-                rx.recv().unwrap()?
-            }
-            None => {
-                let connect = self.config.connect(tls);
-                RUNTIME.handle().enter(|| executor::block_on(connect))?
-            }
-        };
+        let mut runtime = runtime::Builder::new()
+            .enable_all()
+            .basic_scheduler()
+            .build()
+            .unwrap(); // FIXME don't unwrap
+
+        let (client, connection) = runtime.block_on(self.config.connect(tls))?;
 
+        // FIXME don't spawn this so error reporting is less weird.
         let connection = connection.map(|r| {
             if let Err(e) = r {
                 error!("postgres connection error: {}", e)
             }
         });
-        match &self.spawner {
-            Some(spawn) => {
-                spawn(Box::pin(connection));
-            }
-            None => {
-                RUNTIME.spawn(connection);
-            }
-        }
+        runtime.spawn(connection);
 
-        Ok(Client::from(client))
+        Ok(Client::new(runtime, client))
     }
 }
 
@@ -306,9 +274,6 @@ impl FromStr for Config {
 
 impl From<tokio_postgres::Config> for Config {
     fn from(config: tokio_postgres::Config) -> Config {
-        Config {
-            config,
-            spawner: None,
-        }
+        Config { config }
     }
 }