sqlx_build_trust_core/pool/
mod.rs

1//! Provides the connection pool for asynchronous SQLx connections.
2//!
3//! Opening a database connection for each and every operation to the database can quickly
4//! become expensive. Furthermore, sharing a database connection between threads and functions
5//! can be difficult to express in Rust.
6//!
7//! A connection pool is a standard technique that can manage opening and re-using connections.
8//! Normally it also enforces a maximum number of connections as these are an expensive resource
9//! on the database server.
10//!
11//! SQLx provides a canonical connection pool implementation intended to satisfy the majority
12//! of use cases.
13//!
14//! See [Pool][crate::pool::Pool] for details.
15//!
16//! Type aliases are provided for each database to make it easier to sprinkle `Pool` through
17//! your codebase:
18//!
19//! * [MssqlPool][crate::mssql::MssqlPool] (MSSQL)
20//! * [MySqlPool][crate::mysql::MySqlPool] (MySQL)
21//! * [PgPool][crate::postgres::PgPool] (PostgreSQL)
22//! * [SqlitePool][crate::sqlite::SqlitePool] (SQLite)
23//!
24//! # Opening a connection pool
25//!
26//! A new connection pool with a default configuration can be created by supplying `Pool`
27//! with the database driver and a connection string.
28//!
29//! ```rust,ignore
30//! use sqlx::Pool;
31//! use sqlx::postgres::Postgres;
32//!
33//! let pool = Pool::<Postgres>::connect("postgres://").await?;
34//! ```
35//!
36//! For convenience, database-specific type aliases are provided:
37//!
38//! ```rust,ignore
39//! use sqlx::mssql::MssqlPool;
40//!
41//! let pool = MssqlPool::connect("mssql://").await?;
42//! ```
43//!
44//! # Using a connection pool
45//!
46//! A connection pool implements [`Executor`][crate::executor::Executor] and can be used directly
47//! when executing a query. Notice that only an immutable reference (`&Pool`) is needed.
48//!
49//! ```rust,ignore
50//! sqlx::query("DELETE FROM articles").execute(&pool).await?;
51//! ```
52//!
53//! A connection or transaction may also be manually acquired with
54//! [`Pool::acquire`] or
55//! [`Pool::begin`].
56
57use self::inner::PoolInner;
58#[cfg(all(
59    any(
60        feature = "postgres",
61        feature = "mysql",
62        feature = "mssql",
63        feature = "sqlite"
64    ),
65    feature = "any"
66))]
67use crate::any::{Any, AnyKind};
68use crate::connection::Connection;
69use crate::database::Database;
70use crate::error::Error;
71use crate::transaction::Transaction;
72use event_listener::EventListener;
73use futures_core::FusedFuture;
74use futures_util::FutureExt;
75use std::fmt;
76use std::future::Future;
77use std::pin::Pin;
78use std::sync::Arc;
79use std::task::{Context, Poll};
80use std::time::{Duration, Instant};
81
82#[macro_use]
83mod executor;
84
85#[macro_use]
86pub mod maybe;
87
88mod connection;
89mod inner;
90mod options;
91
92pub use self::connection::PoolConnection;
93pub use self::options::{PoolConnectionMetadata, PoolOptions};
94
95#[doc(hidden)]
96pub use self::maybe::MaybePoolConnection;
97
98/// An asynchronous pool of SQLx database connections.
99///
100/// Create a pool with [Pool::connect] or [Pool::connect_with] and then call [Pool::acquire]
101/// to get a connection from the pool; when the connection is dropped it will return to the pool
102/// so it can be reused.
103///
104/// You can also pass `&Pool` directly anywhere an `Executor` is required; this will automatically
105/// checkout a connection for you.
106///
107/// See [the module documentation](crate::pool) for examples.
108///
109/// The pool has a maximum connection limit that it will not exceed; if `acquire()` is called
110/// when at this limit and all connections are checked out, the task will be made to wait until
111/// a connection becomes available.
112///
113/// You can configure the connection limit, and other parameters, using [PoolOptions][crate::pool::PoolOptions].
114///
115/// Calls to `acquire()` are fair, i.e. fulfilled on a first-come, first-serve basis.
116///
117/// `Pool` is `Send`, `Sync` and `Clone`. It is intended to be created once at the start of your
118/// application/daemon/web server/etc. and then shared with all tasks throughout the process'
119/// lifetime. How best to accomplish this depends on your program architecture.
120///
121/// In Actix-Web, for example, you can share a single pool with all request handlers using [web::Data].
122///
123/// Cloning `Pool` is cheap as it is simply a reference-counted handle to the inner pool state.
124/// When the last remaining handle to the pool is dropped, the connections owned by the pool are
125/// immediately closed (also by dropping). `PoolConnection` returned by [Pool::acquire] and
126/// `Transaction` returned by [Pool::begin] both implicitly hold a reference to the pool for
127/// their lifetimes.
128///
129/// If you prefer to explicitly shutdown the pool and gracefully close its connections (which
130/// depending on the database type, may include sending a message to the database server that the
131/// connection is being closed), you can call [Pool::close] which causes all waiting and subsequent
132/// calls to [Pool::acquire] to return [Error::PoolClosed], and waits until all connections have
133/// been returned to the pool and gracefully closed.
134///
135/// Type aliases are provided for each database to make it easier to sprinkle `Pool` through
136/// your codebase:
137///
138/// * [MssqlPool][crate::mssql::MssqlPool] (MSSQL)
139/// * [MySqlPool][crate::mysql::MySqlPool] (MySQL)
140/// * [PgPool][crate::postgres::PgPool] (PostgreSQL)
141/// * [SqlitePool][crate::sqlite::SqlitePool] (SQLite)
142///
143/// [web::Data]: https://docs.rs/actix-web/3/actix_web/web/struct.Data.html
144///
145/// ### Note: Drop Behavior
146/// Due to a lack of async `Drop`, dropping the last `Pool` handle may not immediately clean
147/// up connections by itself. The connections will be dropped locally, which is sufficient for
148/// SQLite, but for client/server databases like MySQL and Postgres, that only closes the
149/// client side of the connection. The server will not know the connection is closed until
150/// potentially much later: this is usually dictated by the TCP keepalive timeout in the server
151/// settings.
152///
153/// Because the connection may not be cleaned up immediately on the server side, you may run
154/// into errors regarding connection limits if you are creating and dropping many pools in short
155/// order.
156///
157/// We recommend calling [`.close().await`] to gracefully close the pool and its connections
158/// when you are done using it. This will also wake any tasks that are waiting on an `.acquire()`
159/// call, so for long-lived applications it's a good idea to call `.close()` during shutdown.
160///
161/// If you're writing tests, consider using `#[sqlx::test]` which handles the lifetime of
162/// the pool for you.
163///
164/// [`.close().await`]: Pool::close
165///
166/// ### Why Use a Pool?
167///
168/// A single database connection (in general) cannot be used by multiple threads simultaneously
169/// for various reasons, but an application or web server will typically need to execute numerous
170/// queries or commands concurrently (think of concurrent requests against a web server; many or all
171/// of them will probably need to hit the database).
172///
173/// You could place the connection in a `Mutex` but this will make it a huge bottleneck.
174///
175/// Naively, you might also think to just open a new connection per request, but this
176/// has a number of other caveats, generally due to the high overhead involved in working with
177/// a fresh connection. Examples to follow.
178///
179/// Connection pools facilitate reuse of connections to _amortize_ these costs, helping to ensure
180/// that you're not paying for them each time you need a connection.
181///
182/// ##### 1. Overhead of Opening a Connection
183/// Opening a database connection is not exactly a cheap operation.
184///
185/// For SQLite, it means numerous requests to the filesystem and memory allocations, while for
186/// server-based databases it involves performing DNS resolution, opening a new TCP connection and
187/// allocating buffers.
188///
189/// Each connection involves a nontrivial allocation of resources for the database server, usually
190/// including spawning a new thread or process specifically to handle the connection, both for
191/// concurrency and isolation of faults.
192///
193/// Additionally, database connections typically involve a complex handshake including
194/// authentication, negotiation regarding connection parameters (default character sets, timezones,
195/// locales, supported features) and upgrades to encrypted tunnels.
196///
197/// If `acquire()` is called on a pool with all connections checked out but it is not yet at its
198/// connection limit (see next section), then a new connection is immediately opened, so this pool
199/// does not _automatically_ save you from the overhead of creating a new connection.
200///
201/// However, because this pool by design enforces _reuse_ of connections, this overhead cost
202/// is not paid each and every time you need a connection. In fact, if you set
203/// [the `min_connections` option in PoolOptions][PoolOptions::min_connections], the pool will
204/// create that many connections up-front so that they are ready to go when a request comes in,
205/// and maintain that number on a best-effort basis for consistent performance.
206///
207/// ##### 2. Connection Limits (MySQL, MSSQL, Postgres)
208/// Database servers usually place hard limits on the number of connections that are allowed open at
209/// any given time, to maintain performance targets and prevent excessive allocation of resources,
210/// such as RAM, journal files, disk caches, etc.
211///
212/// These limits have different defaults per database flavor, and may vary between different
213/// distributions of the same database, but are typically configurable on server start;
214/// if you're paying for managed database hosting then the connection limit will typically vary with
215/// your pricing tier.
216///
217/// In MySQL, the default limit is typically 150, plus 1 which is reserved for a user with the
218/// `CONNECTION_ADMIN` privilege so you can still access the server to diagnose problems even
219/// with all connections being used.
220///
221/// In MSSQL the only documentation for the default maximum limit is that it depends on the version
222/// and server configuration.
223///
224/// In Postgres, the default limit is typically 100, minus 3 which are reserved for superusers
225/// (putting the default limit for unprivileged users at 97 connections).
226///
227/// In any case, exceeding these limits results in an error when opening a new connection, which
228/// in a web server context will turn into a `500 Internal Server Error` if not handled, but should
229/// be turned into either `403 Forbidden` or `429 Too Many Requests` depending on your rate-limiting
230/// scheme. However, in a web context, telling a client "go away, maybe try again later" results in
231/// a sub-optimal user experience.
232///
233/// Instead with a connection pool, clients are made to wait in a fair queue for a connection to
234/// become available; by using a single connection pool for your whole application, you can ensure
235/// that you don't exceed the connection limit of your database server while allowing response
236/// time to degrade gracefully at high load.
237///
238/// Of course, if multiple applications are connecting to the same database server, then you
239/// should ensure that the connection limits for all applications add up to your server's maximum
240/// connections or less.
241///
242/// ##### 3. Resource Reuse
243/// The first time you execute a query against your database, the database engine must first turn
244/// the SQL into an actionable _query plan_ which it may then execute against the database. This
245/// involves parsing the SQL query, validating and analyzing it, and in the case of Postgres 12+ and
246/// SQLite, generating code to execute the query plan (native or bytecode, respectively).
247///
248/// These database servers provide a way to amortize this overhead by _preparing_ the query,
249/// associating it with an object ID and placing its query plan in a cache to be referenced when
250/// it is later executed.
251///
252/// Prepared statements have other features, like bind parameters, which make them safer and more
253/// ergonomic to use as well. By design, SQLx pushes you towards using prepared queries/statements
254/// via the [Query][crate::query::Query] API _et al._ and the `query!()` macro _et al._, for
255/// reasons of safety, ergonomics, and efficiency.
256///
257/// However, because database connections are typically isolated from each other in the database
258/// server (either by threads or separate processes entirely), they don't typically share prepared
259/// statements between connections so this work must be redone _for each connection_.
260///
261/// As with section 1, by facilitating reuse of connections, `Pool` helps to ensure their prepared
262/// statements (and thus cached query plans) can be reused as much as possible, thus amortizing
263/// the overhead involved.
264///
265/// Depending on the database server, a connection will have caches for all kinds of other data as
266/// well and queries will generally benefit from these caches being "warm" (populated with data).
267pub struct Pool<DB: Database>(pub(crate) Arc<PoolInner<DB>>);
268
269/// A future that resolves when the pool is closed.
270///
271/// See [`Pool::close_event()`] for details.
272pub struct CloseEvent {
273    listener: Option<EventListener>,
274}
275
276impl<DB: Database> Pool<DB> {
277    /// Create a new connection pool with a default pool configuration and
278    /// the given connection URL, and immediately establish one connection.
279    ///
280    /// Refer to the relevant `ConnectOptions` impl for your database for the expected URL format:
281    ///
282    /// * Postgres: [`PgConnectOptions`][crate::postgres::PgConnectOptions]
283    /// * MySQL: [`MySqlConnectOptions`][crate::mysql::MySqlConnectOptions]
284    /// * SQLite: [`SqliteConnectOptions`][crate::sqlite::SqliteConnectOptions]
285    /// * MSSQL: [`MssqlConnectOptions`][crate::mssql::MssqlConnectOptions]
286    ///
287    /// The default configuration is mainly suited for testing and light-duty applications.
288    /// For production applications, you'll likely want to make at least few tweaks.
289    ///
290    /// See [`PoolOptions::new()`] for details.
291    pub async fn connect(url: &str) -> Result<Self, Error> {
292        PoolOptions::<DB>::new().connect(url).await
293    }
294
295    /// Create a new connection pool with a default pool configuration and
296    /// the given `ConnectOptions`, and immediately establish one connection.
297    ///
298    /// The default configuration is mainly suited for testing and light-duty applications.
299    /// For production applications, you'll likely want to make at least few tweaks.
300    ///
301    /// See [`PoolOptions::new()`] for details.
302    pub async fn connect_with(
303        options: <DB::Connection as Connection>::Options,
304    ) -> Result<Self, Error> {
305        PoolOptions::<DB>::new().connect_with(options).await
306    }
307
308    /// Create a new connection pool with a default pool configuration and
309    /// the given connection URL.
310    ///
311    /// The pool will establish connections only as needed.
312    ///
313    /// Refer to the relevant [`ConnectOptions`] impl for your database for the expected URL format:
314    ///
315    /// * Postgres: [`PgConnectOptions`][crate::postgres::PgConnectOptions]
316    /// * MySQL: [`MySqlConnectOptions`][crate::mysql::MySqlConnectOptions]
317    /// * SQLite: [`SqliteConnectOptions`][crate::sqlite::SqliteConnectOptions]
318    /// * MSSQL: [`MssqlConnectOptions`][crate::mssql::MssqlConnectOptions]
319    ///
320    /// The default configuration is mainly suited for testing and light-duty applications.
321    /// For production applications, you'll likely want to make at least few tweaks.
322    ///
323    /// See [`PoolOptions::new()`] for details.
324    pub fn connect_lazy(url: &str) -> Result<Self, Error> {
325        PoolOptions::<DB>::new().connect_lazy(url)
326    }
327
328    /// Create a new connection pool with a default pool configuration and
329    /// the given `ConnectOptions`.
330    ///
331    /// The pool will establish connections only as needed.
332    ///
333    /// The default configuration is mainly suited for testing and light-duty applications.
334    /// For production applications, you'll likely want to make at least few tweaks.
335    ///
336    /// See [`PoolOptions::new()`] for details.
337    pub fn connect_lazy_with(options: <DB::Connection as Connection>::Options) -> Self {
338        PoolOptions::<DB>::new().connect_lazy_with(options)
339    }
340
341    /// Retrieves a connection from the pool.
342    ///
343    /// The total time this method is allowed to execute is capped by
344    /// [`PoolOptions::acquire_timeout`].
345    /// If that timeout elapses, this will return [`Error::PoolClosed`].
346    ///
347    /// ### Note: Cancellation/Timeout May Drop Connections
348    /// If `acquire` is cancelled or times out after it acquires a connection from the idle queue or
349    /// opens a new one, it will drop that connection because we don't want to assume it
350    /// is safe to return to the pool, and testing it to see if it's safe to release could introduce
351    /// subtle bugs if not implemented correctly. To avoid that entirely, we've decided to not
352    /// gracefully handle cancellation here.
353    ///
354    /// However, if your workload is sensitive to dropped connections such as using an in-memory
355    /// SQLite database with a pool size of 1, you can pretty easily ensure that a cancelled
356    /// `acquire()` call will never drop connections by tweaking your [`PoolOptions`]:
357    ///
358    /// * Set [`test_before_acquire(false)`][PoolOptions::test_before_acquire]
359    /// * Never set [`before_acquire`][PoolOptions::before_acquire] or
360    ///   [`after_connect`][PoolOptions::after_connect].
361    ///
362    /// This should eliminate any potential `.await` points between acquiring a connection and
363    /// returning it.
364    pub fn acquire(&self) -> impl Future<Output = Result<PoolConnection<DB>, Error>> + 'static {
365        let shared = self.0.clone();
366        async move { shared.acquire().await.map(|conn| conn.reattach()) }
367    }
368
369    /// Attempts to retrieve a connection from the pool if there is one available.
370    ///
371    /// Returns `None` immediately if there are no idle connections available in the pool
372    /// or there are tasks waiting for a connection which have yet to wake.
373    pub fn try_acquire(&self) -> Option<PoolConnection<DB>> {
374        self.0.try_acquire().map(|conn| conn.into_live().reattach())
375    }
376
377    /// Retrieves a connection and immediately begins a new transaction.
378    pub async fn begin(&self) -> Result<Transaction<'static, DB>, Error> {
379        Ok(Transaction::begin(MaybePoolConnection::PoolConnection(self.acquire().await?)).await?)
380    }
381
382    /// Attempts to retrieve a connection and immediately begins a new transaction if successful.
383    pub async fn try_begin(&self) -> Result<Option<Transaction<'static, DB>>, Error> {
384        match self.try_acquire() {
385            Some(conn) => Transaction::begin(MaybePoolConnection::PoolConnection(conn))
386                .await
387                .map(Some),
388
389            None => Ok(None),
390        }
391    }
392
393    /// Shut down the connection pool, immediately waking all tasks waiting for a connection.
394    ///
395    /// Upon calling this method, any currently waiting or subsequent calls to [`Pool::acquire`] and
396    /// the like will immediately return [`Error::PoolClosed`] and no new connections will be opened.
397    /// Checked-out connections are unaffected, but will be gracefully closed on-drop
398    /// rather than being returned to the pool.
399    ///
400    /// Returns a `Future` which can be `.await`ed to ensure all connections are
401    /// gracefully closed. It will first close any idle connections currently waiting in the pool,
402    /// then wait for all checked-out connections to be returned or closed.
403    ///
404    /// Waiting for connections to be gracefully closed is optional, but will allow the database
405    /// server to clean up the resources sooner rather than later. This is especially important
406    /// for tests that create a new pool every time, otherwise you may see errors about connection
407    /// limits being exhausted even when running tests in a single thread.
408    ///
409    /// If the returned `Future` is not run to completion, any remaining connections will be dropped
410    /// when the last handle for the given pool instance is dropped, which could happen in a task
411    /// spawned by `Pool` internally and so may be unpredictable otherwise.
412    ///
413    /// `.close()` may be safely called and `.await`ed on multiple handles concurrently.
414    pub fn close(&self) -> impl Future<Output = ()> + '_ {
415        self.0.close()
416    }
417
418    /// Returns `true` if [`.close()`][Pool::close] has been called on the pool, `false` otherwise.
419    pub fn is_closed(&self) -> bool {
420        self.0.is_closed()
421    }
422
423    /// Get a future that resolves when [`Pool::close()`] is called.
424    ///
425    /// If the pool is already closed, the future resolves immediately.
426    ///
427    /// This can be used to cancel long-running operations that hold onto a [`PoolConnection`]
428    /// so they don't prevent the pool from closing (which would otherwise wait until all
429    /// connections are returned).
430    ///
431    /// Examples
432    /// ========
433    /// These examples use Postgres and Tokio, but should suffice to demonstrate the concept.
434    ///
435    /// Do something when the pool is closed:
436    /// ```rust,no_run
437    /// # #[cfg(feature = "postgres")]
438    /// # async fn bleh() -> sqlx_core::error::Result<()> {
439    /// use sqlx::PgPool;
440    ///
441    /// let pool = PgPool::connect("postgresql://...").await?;
442    ///
443    /// let pool2 = pool.clone();
444    ///
445    /// tokio::spawn(async move {
446    ///     // Demonstrates that `CloseEvent` is itself a `Future` you can wait on.
447    ///     // This lets you implement any kind of on-close event that you like.
448    ///     pool2.close_event().await;
449    ///
450    ///     println!("Pool is closing!");
451    ///
452    ///     // Imagine maybe recording application statistics or logging a report, etc.
453    /// });
454    ///
455    /// // The rest of the application executes normally...
456    ///
457    /// // Close the pool before the application exits...
458    /// pool.close().await;
459    ///
460    /// # Ok(())
461    /// # }
462    /// ```
463    ///
464    /// Cancel a long-running operation:
465    /// ```rust,no_run
466    /// # #[cfg(feature = "postgres")]
467    /// # async fn bleh() -> sqlx_core::error::Result<()> {
468    /// use sqlx::{Executor, PgPool};
469    ///
470    /// let pool = PgPool::connect("postgresql://...").await?;
471    ///
472    /// let pool2 = pool.clone();
473    ///
474    /// tokio::spawn(async move {
475    ///     pool2.close_event().do_until(async {
476    ///         // This statement normally won't return for 30 days!
477    ///         // (Assuming the connection doesn't time out first, of course.)
478    ///         pool2.execute("SELECT pg_sleep('30 days')").await;
479    ///
480    ///         // If the pool is closed before the statement completes, this won't be printed.
481    ///         // This is because `.do_until()` cancels the future it's given if the
482    ///         // pool is closed first.
483    ///         println!("Waited!");
484    ///     }).await;
485    /// });
486    ///
487    /// // This normally wouldn't return until the above statement completed and the connection
488    /// // was returned to the pool. However, thanks to `.do_until()`, the operation was
489    /// // cancelled as soon as we called `.close().await`.
490    /// pool.close().await;
491    ///
492    /// # Ok(())
493    /// # }
494    /// ```
495    pub fn close_event(&self) -> CloseEvent {
496        self.0.close_event()
497    }
498
499    /// Returns the number of connections currently active. This includes idle connections.
500    pub fn size(&self) -> u32 {
501        self.0.size()
502    }
503
504    /// Returns the number of connections active and idle (not in use).
505    ///
506    /// As of 0.6.0, this has been fixed to use a separate atomic counter and so should be fine to
507    /// call even at high load.
508    ///
509    /// This previously called [`crossbeam::queue::ArrayQueue::len()`] which waits for the head and
510    /// tail pointers to be in a consistent state, which may never happen at high levels of churn.
511    pub fn num_idle(&self) -> usize {
512        self.0.num_idle()
513    }
514
515    /// Gets a clone of the connection options for this pool
516    pub fn connect_options(&self) -> Arc<<DB::Connection as Connection>::Options> {
517        self.0
518            .connect_options
519            .read()
520            .expect("write-lock holder panicked")
521            .clone()
522    }
523
524    /// Updates the connection options this pool will use when opening any future connections.  Any
525    /// existing open connection in the pool will be left as-is.
526    pub fn set_connect_options(&self, connect_options: <DB::Connection as Connection>::Options) {
527        // technically write() could also panic if the current thread already holds the lock,
528        // but because this method can't be re-entered by the same thread that shouldn't be a problem
529        let mut guard = self
530            .0
531            .connect_options
532            .write()
533            .expect("write-lock holder panicked");
534        *guard = Arc::new(connect_options);
535    }
536
537    /// Get the options for this pool
538    pub fn options(&self) -> &PoolOptions<DB> {
539        &self.0.options
540    }
541}
542
543#[cfg(all(
544    any(
545        feature = "postgres",
546        feature = "mysql",
547        feature = "mssql",
548        feature = "sqlite"
549    ),
550    feature = "any"
551))]
552impl Pool<Any> {
553    /// Returns the database driver currently in-use by this `Pool`.
554    ///
555    /// Determined by the connection URL.
556    pub fn any_kind(&self) -> AnyKind {
557        self.0
558            .connect_options
559            .read()
560            .expect("write-lock holder panicked")
561            .kind()
562    }
563}
564
565/// Returns a new [Pool] tied to the same shared connection pool.
566impl<DB: Database> Clone for Pool<DB> {
567    fn clone(&self) -> Self {
568        Self(Arc::clone(&self.0))
569    }
570}
571
572impl<DB: Database> fmt::Debug for Pool<DB> {
573    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
574        fmt.debug_struct("Pool")
575            .field("size", &self.0.size())
576            .field("num_idle", &self.0.num_idle())
577            .field("is_closed", &self.0.is_closed())
578            .field("options", &self.0.options)
579            .finish()
580    }
581}
582
583impl CloseEvent {
584    /// Execute the given future until it returns or the pool is closed.
585    ///
586    /// Cancels the future and returns `Err(PoolClosed)` if/when the pool is closed.
587    /// If the pool was already closed, the future is never run.
588    pub async fn do_until<Fut: Future>(&mut self, fut: Fut) -> Result<Fut::Output, Error> {
589        // Check that the pool wasn't closed already.
590        //
591        // We use `poll_immediate()` as it will use the correct waker instead of
592        // a no-op one like `.now_or_never()`, but it won't actually suspend execution here.
593        futures_util::future::poll_immediate(&mut *self)
594            .await
595            .map_or(Ok(()), |_| Err(Error::PoolClosed))?;
596
597        futures_util::pin_mut!(fut);
598
599        // I find that this is clearer in intent than `futures_util::future::select()`
600        // or `futures_util::select_biased!{}` (which isn't enabled anyway).
601        futures_util::future::poll_fn(|cx| {
602            // Poll `fut` first as the wakeup event is more likely for it than `self`.
603            if let Poll::Ready(ret) = fut.as_mut().poll(cx) {
604                return Poll::Ready(Ok(ret));
605            }
606
607            // Can't really factor out mapping to `Err(Error::PoolClosed)` though it seems like
608            // we should because that results in a different `Ok` type each time.
609            //
610            // Ideally we'd map to something like `Result<!, Error>` but using `!` as a type
611            // is not allowed on stable Rust yet.
612            self.poll_unpin(cx).map(|_| Err(Error::PoolClosed))
613        })
614        .await
615    }
616}
617
618impl Future for CloseEvent {
619    type Output = ();
620
621    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
622        if let Some(listener) = &mut self.listener {
623            futures_core::ready!(listener.poll_unpin(cx));
624        }
625
626        // `EventListener` doesn't like being polled after it yields, and even if it did it
627        // would probably just wait for the next event, neither of which we want.
628        //
629        // So this way, once we get our close event, we fuse this future to immediately return.
630        self.listener = None;
631
632        Poll::Ready(())
633    }
634}
635
636impl FusedFuture for CloseEvent {
637    fn is_terminated(&self) -> bool {
638        self.listener.is_none()
639    }
640}
641
642/// get the time between the deadline and now and use that as our timeout
643///
644/// returns `Error::PoolTimedOut` if the deadline is in the past
645fn deadline_as_timeout<DB: Database>(deadline: Instant) -> Result<Duration, Error> {
646    deadline
647        .checked_duration_since(Instant::now())
648        .ok_or(Error::PoolTimedOut)
649}
650
651#[test]
652#[allow(dead_code)]
653fn assert_pool_traits() {
654    fn assert_send_sync<T: Send + Sync>() {}
655    fn assert_clone<T: Clone>() {}
656
657    fn assert_pool<DB: Database>() {
658        assert_send_sync::<Pool<DB>>();
659        assert_clone::<Pool<DB>>();
660    }
661}