flaque 0.1.0

// Copyright 2026 Ivan Enderlin <ivan@mnt.io>. Licensed under the EUPL, version
// 1.2. See the `LICENSE` file.

//! The almighty pool!
//!
//! # Quick Overview
//!
//! A _pool_ is a collection of _objects_ of limited size. When an object is
//! fetched from the pool, the number of available objects decreases by one.
//! Once the fetched object is dropped, it goes back into the _pool_, and the
//! number of available objects is increased by one.
//!
//! All objects are lazily created with a _builder_. Objects are _recycled_,
//! i.e. when an object has been built and fetched once, it can be mutated, and
//! when it goes back in the pool, it stays as is, untouched.
//!
//! ```rust
//! use flaque::Pool;
//!
//! // Let's define what the pool will hold.
//! struct T {
//!     x: u8,
//! }
//!
//! // Let's create the pool: capacity of 8, and the builder for the `T`s.
//! let pool = Pool::new(8, || T { x: 0 });
//!
//! // Let's get an object from the pool.
//! let object_0 = pool.try_get().unwrap();
//! ```
//!
//! The main methods are:
//!
//! - [`Pool::try_get`] if one wants to synchronously get an object from the
//!   pool,
//! - [`Pool::get`] if one wants to asynchronously get an object from the pool.
//!
//! An object is represented by the [`Object`] type. When it drops, the value
//! returns to the pool.
//!
//! # Features
//!
//! ## Lock- and wait-free
//!
//! This pool implementation is lock-free and sometimes wait-free. It is
//! designed to be _fast_ and _consistent_.
//!
//! | Operation | Lock-free | Wait-free |
//! |-|-|-|
//! | `Pool::try_get` | yes | yes |
//! | `Pool::get` | yes | no |
//! | `Object::drop` | yes | yes |
//! | `Pool::close` | yes | no |
//! | `Pool::reopen` | yes | yes |
//!
//! If you're not familiar with these definitions, please
//! [read _Non-blocking algorithm_][Non-blocking algorithm].
//!
//! ## `Get` and `Close` futures
//!
//! `Pool::get` returns a `Get` future. It resolves to `Some(object)` once one
//! is available in the pool, otherwise it returns to `None` if the pool is
//! closed.
//!
//! `Pool::close` returns a `Close` future. It resolves once all `Object`s are
//! back into the pool, i.e. when they are all dropped. `Pool::close` takes
//! `self`, i.e. it takes ownership of the pool. It will be returned back once
//! the future resolves. This guarantees that it is impossible to reopen the
//! pool, and to fetch new `Object`s, while waiting for all `Object`s to return
//! back into the pool. Thus, this is not unusual to see:
//!
//! ```rust
//! let pool = pool.close().await;
//! ```
//!
//! [Non-blocking algorithm]: https://en.wikipedia.org/wiki/Non-blocking_algorithm

#[cfg(not(loom))]
mod sys {
    pub use std::{hint, sync};

    pub mod cell {
        /// Compatibility layer between `std` and `loom`'s `UnsafeCell`.
        #[derive(Debug)]
        pub struct UnsafeCell<T>(std::cell::UnsafeCell<T>);

        impl<T> UnsafeCell<T> {
            pub fn new(data: T) -> UnsafeCell<T> {
                UnsafeCell(std::cell::UnsafeCell::new(data))
            }

            pub fn with_mut<F, U>(&self, f: F) -> U
            where
                F: FnOnce(*mut T) -> U,
            {
                f(self.0.get())
            }
        }
    }
}

#[cfg(loom)]
mod sys {
    pub use loom::{cell, sync};

    pub mod hint {
        pub use loom::hint::*;

        pub fn cold_path() {}
    }
}

mod cache_padded;
mod slots;
mod try_lock;

use std::{
    cmp,
    mem::MaybeUninit,
    ops::{Deref, DerefMut, Not},
    pin::Pin,
    ptr,
    task::{Context, Poll, Waker},
};

use self::{
    slots::{SlotIndex, Slots},
    sys::{
        cell::UnsafeCell,
        hint,
        sync::{
            Arc,
            mpsc::{Receiver, Sender, channel},
        },
    },
    try_lock::TryLock,
};

/// Error returned by [`Pool::try_get`].
#[derive(Debug, PartialEq)]
pub enum AcquireError {
    /// The pool has been closed: it's impossible to acquire a new object.
    Closed,

    /// All objects are outside the pool, i.e. the pool has no vacant slot. Try
    /// later!
    AllSlotsOccupied,
}

/// Error returned by [`PoolInner::try_release`].
///
/// This is an internal error.
#[derive(Debug, PartialEq)]
enum ReleaseError {
    /// The given position is invalid.
    InvalidPosition,
}

/// The almighty pool!
///
/// See [the module's documentation to learn more][self].
pub struct Pool<T> {
    /// The inner thingy.
    ///
    /// It is behind an `Arc` to be shared with [`Object`] and [`Get`].
    inner: Arc<PoolInner<T>>,

    /// The size of the pool.
    size: usize,
}

/// The inner type of the [`Pool`].
struct PoolInner<T> {
    /// All the slots, stored in memory contiguously.
    ///
    /// Object values are not stored here, see [`Self::objects`]. A [`Slots`]
    /// represents whether multiple slots are occupied or vacant.
    blocks: Box<[Slots]>,

    /// All the object values, stored in memory contiguously.
    ///
    /// [`Self::blocks`] is used to get a position into this piece of memory.
    ///
    /// Objects are initially set to `None`. It becomes `Some` when acquired
    /// for the first time. The first value is built with [`Self::builder`].
    objects: Box<[UnsafeCell<Option<T>>]>,

    /// Objects are lazily built in [`Self::objects`]. When an object is
    /// acquired for the first time, it must be built… with this builder.
    builder: Box<dyn Fn() -> T>,

    /// The sender part of the channel used by the [`Get`] futures to send
    /// their wakers.
    ///
    /// See also [`Self::get_wakers_receiver`].
    get_wakers_sender: Sender<Waker>,

    /// The receiver part of the channel used to wake up [`Get`] futures.
    ///
    /// See also [`Self::get_wakers_sender`].
    get_wakers_receiver: Receiver<Waker>,

    /// The closing state, for the [`Close`] future.
    closing_state: TryLock<ClosingState>,
}

/// An [`Object`] acquired from a [`Pool`] with [`Pool::try_get`] or
/// [`Pool::get`].
///
/// It can either be a fresh new object (built by the builder given to
/// [`Pool::new`]), or a “recycled” object, i.e. an object that has previously
/// been acquired. An object conserves its state.
///
/// When an [`Object`] drops, it synchronously and atomically returns into the
/// pool.
///
/// This type wraps the real value `T`, which is the value one wants from the
/// pool. [`Object`] `deref`s and `deref_mut`s to `T`.
#[must_use = "`Object` will immediately return in the pool if unused"]
pub struct Object<T> {
    /// The position of [`Self`] in the [`Pool`].
    position: Position,

    /// The real value from the [`Pool`].
    value: MaybeUninit<T>,

    /// A pointer to [`Pool::inner`].
    pool_inner: Arc<PoolInner<T>>,
}

impl<T> Pool<T> {
    /// Create a new [`Pool`].
    ///
    /// The size is expressed as a _desired size_. Due to how this pool is
    /// implemented, the size is rounded to the next power of 2. One can use
    /// [`Self::size`] to get the actual size of the pool. The minimum size is
    /// also 8.
    ///
    /// ```rust
    /// # use flaque::Pool;
    /// #
    /// /// The type representing the objects inside the pool.
    /// struct T {
    ///     x: u8,
    /// }
    ///
    /// impl Default for T {
    ///     fn default() -> Self {
    ///         Self { x: 0 }
    ///     }
    /// }
    ///
    /// // Create a pool, containing up to 16 objects, despite 10 is given.
    /// let pool = Pool::new(10, T::default);
    /// assert_eq!(pool.size(), 16);
    ///
    /// // The smallest size is 8.
    /// let smallest_pool = Pool::new(1, T::default);
    /// assert_eq!(smallest_pool.size(), 8);
    /// ```
    pub fn new<B>(desired_size: usize, builder: B) -> Self
    where
        B: Fn() -> T + 'static,
    {
        let size = compute_size(desired_size);
        let (waker_sender, waker_receiver) = channel();

        Self {
            size,
            inner: Arc::new(PoolInner {
                blocks: {
                    let number_of_slots = size >> Slots::SLOTS_SIZE.ilog2();

                    let mut keepers = Vec::with_capacity(number_of_slots);

                    for _ in 0..number_of_slots {
                        keepers.push(Slots::new());
                    }

                    keepers.into_boxed_slice()
                },
                objects: {
                    let mut objects = Vec::with_capacity(size);

                    for _ in 0..size {
                        objects.push(UnsafeCell::new(None));
                    }

                    objects.into_boxed_slice()
                },
                builder: Box::new(builder),
                get_wakers_sender: waker_sender,
                get_wakers_receiver: waker_receiver,
                closing_state: TryLock::new(ClosingState::Opened),
            }),
        }
    }

    /// Returns the size of this [`Pool`].
    ///
    /// ```rust
    /// # use flaque::Pool;
    /// #
    /// # type T = bool;
    /// #
    /// // Create a pool, containing up to 16 objects, despite 10 is given.
    /// let pool = Pool::new(10, T::default);
    /// assert_eq!(pool.size(), 16);
    /// ```
    pub fn size(&self) -> usize {
        self.size
    }

    /// Returns whether this [`Pool`] is closed.
    ///
    /// It returns `false` before [`Pool::close`] is called, or after
    /// [`Pool::reopen`] are called, `true` otherwise.
    ///
    /// ```rust
    /// use std::ops::Not;
    /// # use flaque::Pool;
    /// #
    /// # type T = bool;
    /// #
    /// # futures_executor::block_on(async {
    /// let pool = Pool::new(8, T::default);
    ///
    /// // The pool is not closed.
    /// assert!(pool.is_closed().not());
    ///
    /// // Now let's call `Pool::close`. It returns a `Close` future that
    /// // returns the `Pool` back once resolved.
    /// let pool = pool.close().await;
    ///
    /// // At this step, the pool is entirely closed.
    /// assert!(pool.is_closed());
    ///
    /// // We can re-open it.
    /// pool.reopen();
    ///
    /// // Not closed anymore!
    /// assert!(pool.is_closed().not());
    /// # })
    /// ```
    pub fn is_closed(&self) -> bool {
        self.inner.is_closed()
    }

    /// Close this [`Pool`].
    ///
    /// The returned [`Close`] value is a future. It will resolve once all
    /// [`Object`]s have returned to this [`Pool`].
    ///
    /// Note that this method takes ownership of `self`. The pool will be
    /// returned back once the future resolves. This guarantees that it is
    /// impossible to reopen the pool, and to fetch new [`Object`]s, while
    /// waiting for all [`Object`]s to return back into the pool.
    ///
    /// ```rust
    /// use std::thread::spawn;
    /// # use flaque::Pool;
    /// #
    /// # type T = bool;
    /// #
    /// # futures_executor::block_on(async {
    /// // Create a pool.
    /// let pool = Pool::new(8, T::default);
    ///
    /// // Fetch 2 objects, and send them in threads for example.
    /// let object_0 = pool.try_get().unwrap();
    /// let object_1 = pool.try_get().unwrap();
    ///
    /// spawn(move || {
    ///     // Do something with `object_0`.
    ///     let object_0 = object_0;
    /// });
    ///
    /// spawn(move || {
    ///     // Do something with `object_1`.
    ///     let object_1 = object_1;
    /// });
    ///
    /// // Close the pool and wait for all objects to return.
    /// // Once done, we get `pool` back!
    /// let pool = pool.close().await;
    /// assert!(pool.is_closed());
    /// # })
    /// ```
    ///
    /// See [`Pool::reopen`] to reopen this pool.
    pub fn close(self) -> Close<T> {
        Close::new(self)
    }

    /// Reopen this [`Pool`] if closed by [`Pool::close`].
    ///
    /// Calling this method on a non-closed pool is just a waste of time but it
    /// won't panic.
    pub fn reopen(&self) {
        self.inner.reopen();
    }

    /// Try to synchronously get an [`Object`] from this pool.
    ///
    /// It errors with [`AcquireError`] if:
    ///
    /// - all objects are outside the pool,
    /// - the pool is closed.
    ///
    /// ```rust
    /// # use flaque::Pool;
    /// #
    /// # type T = bool;
    /// #
    /// // Create a pool.
    /// let pool = Pool::new(8, T::default);
    ///
    /// // Try to get an object.
    /// if let Ok(object) = pool.try_get() {
    ///     // Yepee!
    /// }
    /// ```
    ///
    /// See [`Pool::get`] for the asynchronous flavour.
    pub fn try_get(&self) -> Result<Object<T>, AcquireError> {
        let (position, object) = self.inner.try_get()?;

        Ok(Object::new(position, object, self.inner.clone()))
    }

    /// Like [`Pool::try_get`] but will wait on an object to be available, i.e.
    /// to be in the pool.
    ///
    /// [`Get`] resolves to `None` if this [`Pool`] has been closed while this
    /// future was waiting —see [`Pool::close`]—, otherwise it resolves to
    /// `Some(`[`Object`]`)` once an object is available in the pool.
    ///
    /// ```rust
    /// # use flaque::Pool;
    /// #
    /// # type T = bool;
    /// # futures_executor::block_on(async {
    /// // Create a pool.
    /// let pool = Pool::new(8, T::default);
    ///
    /// // Wait to get an object.
    /// if let Some(object) = pool.get().await {
    ///     // Yepee!
    /// } else {
    ///     // Oh no, the pool has been closed while awaiting…
    /// }
    /// # })
    /// ```
    pub fn get(&self) -> Get<T> {
        Get::new(&self.inner)
    }
}

#[inline]
fn compute_size(desired_size: usize) -> usize {
    cmp::max(Slots::SLOTS_SIZE as _, desired_size).next_power_of_two()
}

impl<T> PoolInner<T> {
    fn is_closed(&self) -> bool {
        loop {
            if let Some(closing_state) = self.closing_state.try_read_lock() {
                return matches!(*closing_state, ClosingState::Opened).not();
            }

            hint::cold_path();
            hint::spin_loop();
        }
    }

    fn close(&self) {
        loop {
            if let Some(mut closing_state) = self.closing_state.try_write_lock()
            {
                *closing_state = ClosingState::Closing(None);
                break;
            }

            hint::cold_path();
            hint::spin_loop();
        }

        // Wake all pending `Get` futures if any so that they can resolve
        // immediately to their “closed” state.
        self.wake_pending_get_futures();
    }

    fn reopen(&self) {
        loop {
            if let Some(mut closing_state) = self.closing_state.try_write_lock()
            {
                *closing_state = ClosingState::Opened;
                break;
            }

            hint::cold_path();
            hint::spin_loop();
        }
    }

    fn try_get(&self) -> Result<(Position, T), AcquireError> {
        if self.is_closed() {
            return Err(AcquireError::Closed);
        }

        for (block_index, slots) in self.blocks.iter().enumerate() {
            let Ok(slot_index) = slots.try_acquire() else {
                continue;
            };

            let object_position = Position { block_index, slot_index };
            let object_index = object_position.as_object_index();

            let object_value = {
                assert!(object_index < self.objects.len());

                let cell = &self.objects[object_index];

                cell.with_mut(|cell_ptr| {
                    let object_place = unsafe {
                        // SAFETY:
                        // - There is no other access to the wrapped value.
                        // - There is no other reference(s) to the wrapped
                        //   value.
                        &mut *cell_ptr
                    };

                    object_place.take().unwrap_or_else(|| (self.builder)())
                })
            };

            return Ok((object_position, object_value));
        }

        Err(AcquireError::AllSlotsOccupied)
    }

    fn try_release(
        &self,
        object_position: &Position,
        object_value: T,
    ) -> Result<(), ReleaseError> {
        let Some(block) = self.blocks.get(object_position.block_index) else {
            return Err(ReleaseError::InvalidPosition);
        };

        // Put `object` in `objects` before releasing the slot!
        // No one else has access to this position.
        let object_index = object_position.as_object_index();

        {
            assert!(object_index < self.objects.len());

            let cell = &self.objects[object_index];

            cell.with_mut(|cell_ptr| {
                // SAFETY:
                // - There is no other access to the wrapped value.
                // - There is no other reference(s) to the wrapped value.
                unsafe {
                    ptr::replace(cell_ptr, Some(object_value));
                }
            });
        };

        // Now we can release the slot.
        block.release(object_position.slot_index);

        // Wake all pending `Get` futures if any.
        self.wake_pending_get_futures();

        // Wake the `Close` future if any.
        self.wake_pending_close_future();

        Ok(())
    }

    fn wake_pending_get_futures(&self) {
        // Collect all wakers now, in case one waker awakes a future that will
        // add its own waker to the queue again while iterating. We don't want
        // that as it can create deadlock!
        let wakers =
            TryIter::new(&self.get_wakers_receiver).collect::<Vec<_>>();

        for waker in wakers {
            waker.wake()
        }
    }

    fn wake_pending_close_future(&self) {
        loop {
            if let Some(closing_state) = self.closing_state.try_read_lock() {
                // If there is a waker for the `Close` future, let's wake it by
                // reference.
                //
                // Why not calling `maybe_waker::take()` to own the waker and
                // then `waker.wake()`? Because it could create a race
                // condition:
                //
                // - thread 0 executes `<Close as Future>::poll`, enters the
                //   pending branch, and saves the waker.
                // - thread 1 releases 2 objects:
                //   - first object calls this function, takes the waker, calls
                //     it (it's not executed yet),
                //   - second object calls this function, no waker to take, it
                //     does nothing.
                // - thread 0 finally runs, it calls `<Close as Future>::poll`
                //   but what it _sees_ is the first object being released, but
                //   not the second one! It enters the pending branch again, and
                //   saves the waker.
                // - no thread can wake up the `Close` future again.
                //
                // We have a deadlock.
                //
                // To fix that, we take the waker by reference. That way, the
                // second object will _again_ wake up the `Close` future.
                if let ClosingState::Closing(maybe_waker) = &*closing_state
                    && let Some(waker) = maybe_waker.as_ref()
                {
                    waker.wake_by_ref();
                }

                break;
            }

            hint::cold_path();
            hint::spin_loop();
        }
    }
}

unsafe impl<T> Send for PoolInner<T> where T: Send {}
unsafe impl<T> Sync for PoolInner<T> where T: Sync {}

impl<T> Object<T> {
    fn new(
        position: Position,
        value: T,
        pool_inner: Arc<PoolInner<T>>,
    ) -> Self {
        Self { position, value: MaybeUninit::new(value), pool_inner }
    }
}

impl<T> Drop for Object<T> {
    fn drop(&mut self) {
        // SAFETY: The value is guaranteed to be initialised in `Self::new`.
        let value = unsafe { self.value.assume_init_read() };

        // SAFETY: The value is guaranteed to be initialised in `Self::new`, and
        // `Drop` is called only once.
        unsafe {
            self.value.assume_init_drop();
        }

        if self.pool_inner.try_release(&self.position, value).is_err() {
            hint::cold_path();

            unreachable!(
                "`Object` has an invalid position, API has been misused"
            );
        }
    }
}

impl<T> Deref for Object<T> {
    type Target = T;

    fn deref(&self) -> &Self::Target {
        // SAFETY: The value is guaranteed to be initialised in `Self::new`.
        unsafe { self.value.assume_init_ref() }
    }
}

impl<T> DerefMut for Object<T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        // SAFETY: The value is guaranteed to be initialised in `Self::new`.
        unsafe { self.value.assume_init_mut() }
    }
}

/// The position of an [`Object`] in a [`Pool`].
struct Position {
    block_index: usize,
    slot_index: SlotIndex,
}

impl Position {
    fn as_object_index(&self) -> usize {
        self.block_index * Slots::SLOTS_SIZE as usize + self.slot_index as usize
    }
}

/// The future returned by [`Pool::get`].
///
/// It resolves to `Some(`[`Object`]`)` once an object is available in the pool.
/// It resolves to `None` if the pool is closed with [`Pool::close`].
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct Get<T> {
    /// The inner value of the pool, so that it has access to
    /// [`PoolInner::try_get`] and [`PoolInner::get_wakers_sender`].
    ///
    /// It should ideally be using `MaybeUninit`. However, if `<Get as
    /// Future>::poll` is called after the future has been resolved, it will
    /// create an Undefined Behaviour, which is incorrect because `poll` is
    /// not marked as `unsafe`. So, in order to panic properly, we are
    /// using `Option`. (Alternatively, we could fuse the future and use
    /// `MaybeUninit`, but it would add an overhead).
    pool_inner: Option<Arc<PoolInner<T>>>,
}

impl<T> Get<T> {
    fn new(inner: &Arc<PoolInner<T>>) -> Self {
        Self { pool_inner: Some(inner.clone()) }
    }
}

impl<T> Future for Get<T> {
    type Output = Option<Object<T>>;

    fn poll(
        mut self: Pin<&mut Self>,
        context: &mut Context<'_>,
    ) -> Poll<Self::Output> {
        if self.pool_inner.is_none() {
            hint::cold_path();

            unreachable!(
                "The `Get` future has already been resolved: it is violating the contract of `Future`"
            );
        }

        // SAFETY: The `Option` is guaranteed to be `Some`, as checked at the
        // beginning of this method.
        unsafe {
            match self.pool_inner.as_ref().unwrap_unchecked().try_get() {
                Ok((position, object)) => Poll::Ready({
                    let pool_inner = self.pool_inner.take().unwrap_unchecked();

                    Some(Object::new(position, object, pool_inner))
                }),

                Err(AcquireError::Closed) => Poll::Ready(None),

                Err(AcquireError::AllSlotsOccupied) => {
                    let waker = context.waker().clone();
                    let _ = self
                        .pool_inner
                        .as_ref()
                        .unwrap_unchecked()
                        .get_wakers_sender
                        .send(waker);

                    Poll::Pending
                }
            }
        }
    }
}

/// The future returned by [`Pool::close`].
///
/// It owns the [`Pool`]. It transfers ownership back to the caller once
/// resolved. See [`Pool::close`] for an example.
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct Close<T> {
    /// The pool. This struct owns the [`Pool`], so that no one else can use
    /// it.
    ///
    /// It should ideally be using `MaybeUninit`. However, if `<Close as
    /// Future>::poll` is called after the future has been resolved, it will
    /// create an Undefined Behaviour, which is incorrect because `poll` is
    /// not marked as `unsafe`. So, in order to panic properly, we are
    /// using `Option`. (Alternatively, we could fuse the future and use
    /// `MaybeUninit`, but it would add an overhead).
    pool: Option<Pool<T>>,
}

impl<T> Close<T> {
    /// Close the pool and create a new [`Self`].
    ///
    /// This method is used by [`Pool::close`].
    fn new(pool: Pool<T>) -> Self {
        pool.inner.close();

        Self { pool: Some(pool) }
    }
}

impl<T> Future for Close<T> {
    type Output = Pool<T>;

    fn poll(
        mut self: Pin<&mut Self>,
        context: &mut Context<'_>,
    ) -> Poll<Self::Output> {
        if self.pool.is_none() {
            hint::cold_path();

            unreachable!(
                "The `Close` future has already been resolved: it is violating the contract of `Future`"
            );
        }

        // SAFETY: The `Option` is guaranteed to be `Some`, as checked at the
        // beginning of this method.
        unsafe {
            let mut closing_state = loop {
                if let Some(closing_state) = self
                    .pool
                    .as_ref()
                    .unwrap_unchecked()
                    .inner
                    .closing_state
                    .try_write_lock()
                {
                    break closing_state;
                }

                hint::cold_path();
                hint::spin_loop();
            };

            let all_vacants = self
                .pool
                .as_ref()
                .unwrap_unchecked()
                .inner
                .blocks
                .iter()
                .all(|slot| slot.is_all_vacants());

            if all_vacants {
                // At this step, we know that all slots are vacant. We also know
                // that, when `Close` has been created, the pool was closed,
                // which means that it's impossible for a slot
                // to be occupied once all slots have been
                // vacant. Finally, because `Pool::close` takes
                // ownership of the `Pool`, it's impossible to reopen it. So we
                // are good!

                *closing_state = ClosingState::Closed;
                drop(closing_state);

                Poll::Ready(self.pool.take().unwrap_unchecked())
            } else {
                // At this step, not all slots are vacant.

                // We can register the waker and mark the future as pending.
                let waker = context.waker().clone();

                *closing_state = ClosingState::Closing(Some(waker));
                drop(closing_state);

                Poll::Pending
            }
        }
    }
}

enum ClosingState {
    /// Pool is opened.
    Opened,

    /// [`Close`] has been built (if `None`) or is being resolved (if `Some`).
    Closing(Option<Waker>),

    /// [`Close`] has resolved.
    Closed,
}

/// Mimic `std::sync::mpsc::TryIter` for `loom`.
#[derive(Debug)]
struct TryIter<'a, T: 'a> {
    receiver: &'a Receiver<T>,
}

impl<'a, T> TryIter<'a, T> {
    fn new(receiver: &'a Receiver<T>) -> Self {
        Self { receiver }
    }
}

impl<'a, T> Iterator for TryIter<'a, T> {
    type Item = T;

    fn next(&mut self) -> Option<T> {
        self.receiver.try_recv().ok()
    }
}

#[cfg(test)]
mod tests_helper {
    #[cfg(not(loom))]
    pub fn run(test: impl Fn()) {
        test()
    }

    #[cfg(loom)]
    pub use loom::model as run;
}

#[cfg(test)]
mod tests {
    #[cfg(not(loom))]
    mod sys {
        pub use std::thread;

        pub use futures_executor::block_on;
    }

    #[cfg(loom)]
    mod sys {
        pub use loom::future::block_on;
        pub mod thread {
            pub use std::thread::sleep;

            pub use loom::thread::spawn;
        }
    }

    use std::{ops::Not, time::Duration};

    use self::sys::{block_on, thread};
    use super::{AcquireError, Pool, compute_size, tests_helper::run};

    struct T {
        x: u8,
    }

    fn thread_wait() {
        thread::sleep(Duration::from_millis(1));

        #[cfg(loom)]
        loom::thread::yield_now();
    }

    #[test]
    fn test_compute_size() {
        for desired_size in 0..=8 {
            assert_eq!(compute_size(desired_size), 8);
        }

        for desired_size in 9..=16 {
            assert_eq!(compute_size(desired_size), 16);
        }

        for desired_size in 17..=32 {
            assert_eq!(compute_size(desired_size), 32);
        }
    }

    #[test]
    fn test_try_get() {
        run(|| {
            let pool = Pool::new(8, || T { x: 0 });

            // Keep all objects alive until the end of the test.
            let _object_0 = pool.try_get().unwrap();
            let _object_1 = pool.try_get().unwrap();
            let _object_2 = pool.try_get().unwrap();
            let _object_3 = pool.try_get().unwrap();
            let _object_4 = pool.try_get().unwrap();
            let _object_5 = pool.try_get().unwrap();
            let _object_6 = pool.try_get().unwrap();
            let _object_7 = pool.try_get().unwrap();

            // 8 slots maximum: they are all occupied.
            assert!(pool.try_get().is_err());
        })
    }

    #[test]
    fn test_get() {
        run(|| {
            let pool = Pool::new(8, || T { x: 0 });

            block_on(async {
                let mut object_0 = pool.get().await.unwrap();
                object_0.x = 7;
            });

            block_on(async {
                assert_eq!(pool.get().await.unwrap().x, 7);
            });

            let _object_0 = pool.try_get().unwrap();
            let _object_1 = pool.try_get().unwrap();
            let _object_2 = pool.try_get().unwrap();
            let _object_3 = pool.try_get().unwrap();
            let mut object_4 = pool.try_get().unwrap();
            let _object_5 = pool.try_get().unwrap();
            let _object_6 = pool.try_get().unwrap();
            let _object_7 = pool.try_get().unwrap();

            // 8 slots maximum: they are all occupied.
            assert!(pool.try_get().is_err());

            let thread_handle = thread::spawn({
                move || {
                    block_on(async {
                        assert_eq!(pool.get().await.unwrap().x, 42);
                    })
                }
            });

            object_4.x = 42;

            // Give time to the thread to wait on the `Get` future.
            thread_wait();
            drop(object_4);

            thread_handle.join().unwrap();
        })
    }

    /// For a `Get` future to return `None`, the pool must be closed while the
    /// `Get` future is waiting to resolve.
    #[test]
    // Disable the test for `loom` because it doesn't play well with `sleep`. We
    // don't need loom to run on this test, a normal test is enough.
    #[cfg(not(loom))]
    fn test_get_returns_none() {
        run(|| {
            let pool = Pool::new(8, || T { x: 0 });

            // Fetch all objects, so that `pool.get()` will wait.
            let object_0 = pool.try_get().unwrap();
            let object_1 = pool.try_get().unwrap();
            let object_2 = pool.try_get().unwrap();
            let object_3 = pool.try_get().unwrap();
            let object_4 = pool.try_get().unwrap();
            let object_5 = pool.try_get().unwrap();
            let object_6 = pool.try_get().unwrap();
            let object_7 = pool.try_get().unwrap();

            // 8 slots maximum: they are all occupied.
            assert!(pool.try_get().is_err());

            let get_thread_handle = thread::spawn({
                let get_future = pool.get();

                move || {
                    block_on(async {
                        let maybe_object = get_future.await;

                        assert!(maybe_object.is_none());
                    })
                }
            });

            thread::spawn(move || {
                thread::sleep(Duration::from_millis(100));

                drop(object_0);
                drop(object_1);
                drop(object_2);
                drop(object_3);
                drop(object_4);
                drop(object_5);
                drop(object_6);
                drop(object_7);
            });

            // Give time to the threads to start.
            thread_wait();

            // - `Get` is waiting.
            // - `Close` closes the pool.
            // - `Get` is cancelled.
            // - 100ms later, all objects returns to the pool.
            // - `Close` can resolve.

            let pool = block_on(pool.close());

            #[cfg(not(loom))]
            assert!(get_thread_handle.is_finished());

            assert!(pool.is_closed());
        })
    }

    #[test]
    fn test_object_drop_releases_the_value() {
        run(|| {
            let pool = Pool::new(8, || T { x: 0 });

            let object_0 = pool.try_get().unwrap();
            let _object_1 = pool.try_get().unwrap();
            let _object_2 = pool.try_get().unwrap();
            let _object_3 = pool.try_get().unwrap();
            let object_4 = pool.try_get().unwrap();
            let _object_5 = pool.try_get().unwrap();
            let _object_6 = pool.try_get().unwrap();
            let _object_7 = pool.try_get().unwrap();

            assert!(pool.try_get().is_err());

            // Drop the object `object_0`.
            drop(object_0);

            // Now we can get a new object!
            let _object_0 = pool.try_get().unwrap();

            // But now all slots are occupied too.
            assert!(pool.try_get().is_err());

            // Drop the object `object_4`.
            drop(object_4);

            // Now we can get a new object!
            let _object_4 = pool.try_get().unwrap();

            // But now all slots are occupied too.
            assert!(pool.try_get().is_err());
        })
    }

    #[test]
    fn test_object_derefs() {
        run(|| {
            let pool = Pool::new(8, || T { x: 0 });

            {
                let object = pool.try_get().unwrap();
                assert_eq!(object.x, 0);
            }

            {
                let mut object = pool.try_get().unwrap();
                object.x = 42;

                assert_eq!(object.x, 42);
            }

            {
                let object0 = pool.try_get().unwrap();
                assert_eq!(object0.x, 42);

                let object1 = pool.try_get().unwrap();
                assert_eq!(object1.x, 0);
            }
        })
    }

    #[test]
    fn test_close_and_reopen() {
        run(|| {
            let pool = Pool::new(8, || T { x: 0 });

            // Get two objects.
            let mut object_0 = pool.try_get().unwrap();
            let object_1 = pool.try_get().unwrap();

            // Let's close the pool, and wait for all objects to return into the
            // pool to resolve the future.
            let thread_handle =
                thread::spawn(move || block_on(async { pool.close().await }));

            // Give time to the thread to wait on the `Close` future.
            thread_wait();

            // Mutate and drop one object. Still one in the wild, so `Close`
            // doesn't resolve.
            object_0.x = 42;
            drop(object_0);
            thread_wait();

            #[cfg(not(loom))]
            assert!(thread_handle.is_finished().not());

            // Drop the last object: `Close` can now resolve!
            drop(object_1);
            thread_wait();

            let pool = thread_handle.join().unwrap();

            // The pool is still closed!
            assert!(pool.is_closed());

            // Which means we can't get any object from it.
            assert!(matches!(pool.try_get(), Err(AcquireError::Closed)));

            // But we can reopen it!
            pool.reopen();

            // Not closed anymore.
            assert!(pool.is_closed().not());

            // And now, we can get a new object: this is `object_0` again.
            assert_eq!(pool.try_get().unwrap().x, 42);
        })
    }

    #[test]
    fn test_close_when_all_objects_are_wandering() {
        run(|| {
            let pool = Pool::new(8, || T { x: 0 });

            let object_0 = pool.try_get().unwrap();
            let object_1 = pool.try_get().unwrap();
            let object_2 = pool.try_get().unwrap();
            let object_3 = pool.try_get().unwrap();
            let object_4 = pool.try_get().unwrap();
            let object_5 = pool.try_get().unwrap();
            let object_6 = pool.try_get().unwrap();
            let object_7 = pool.try_get().unwrap();

            let thread_handle =
                thread::spawn(|| block_on(async { pool.close().await }));

            thread_wait();
            drop(object_0);
            drop(object_1);
            drop(object_2);
            drop(object_3);
            drop(object_4);
            drop(object_5);
            drop(object_6);
            drop(object_7);

            let pool = thread_handle.join().unwrap();
            assert!(pool.is_closed());
        })
    }

    #[test]
    fn test_reopen_on_non_closed_pool() {
        run(|| {
            let pool = Pool::new(8, || T { x: 0 });
            pool.reopen();
        })
    }
}