flaque 0.2.0

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

//! A fast, concurrent, lock-free, and sometimes wait-free, pool implementation.
//!
//! ## Quick overview
//!
//! A _pool_ is a limited-sized collection of _objects_. 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 increases by one. Being as fast as possible,
//! avoiding blocking, and, whenever possible, avoiding waiting, is the name of
//! the game.
//!
//! All objects are lazily created with a _builder_, which can be synchronous or
//! asynchronous, fallible or infallible. 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,
//! }
//!
//! impl Default for T {
//!     fn default() -> Self {
//!         Self { x: 0 }
//!     }
//! }
//!
//! // Let's create the pool: capacity of 8, and the builder for the `T`s.
//! let pool = Pool::new(8, T::default);
//!
//! // Let's try to get an object from the pool.
//! let object_0 = pool.try_get().unwrap();
//!
//! // Or, let's wait to get an object from the pool.
//! # futures_executor::block_on(async {
//! let object_1 = pool.get().await;
//! # });
//! ```
//!
//! The main methods are:
//!
//! - `Pool::new` to build a `Pool` with a synchronous builder,
//! - `Pool::new_with_async_builder` to build a `Pool` with an asynchronous
//!   builder,
//! - `Pool::try_get` if one wants to synchronously, atomically get an object
//!   from the pool,
//! - `Pool::get` if one wants to asynchronously get an object from the pool,
//!   waiting for one to become available.
//!
//! An object is represented by the `Object<T, _>` type, which `deref`s and
//! `deref_mut`s to `T`. When it drops, the value returns immediately 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 `Ok(object)` once one is
//! available in the pool, otherwise it returns to `Err(GetError)` if the pool
//! is closed or if the builder failed to build an object.
//!
//! `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
//! # use flaque::Pool;
//! #
//! # futures_executor::block_on(async {
//! # let pool = Pool::new(8, bool::default);
//! 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 builder;
mod cache_padded;
mod close;
mod get;
mod slots;
mod try_lock;

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

pub use builder::{Async, AsyncBuilder, IntoResult, Sync, SyncBuilder};
pub use close::Close;
use close::ClosingState;
pub use get::{AsyncGet, Get, GetError};

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 TryGetError<B> {
    /// 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,

    /// A slot has been acquired successfully but the builder has failed to
    /// build an object. The slot is released before this error is returned,
    /// i.e. it can acquired later.
    Builder(B),
}

/// Error returned by [`PoolInner::try_acquire`].
#[derive(Debug, PartialEq)]
enum TryAcquireError {
    /// 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 TryReleaseError {
    /// The given position is invalid.
    InvalidPosition,
}

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

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

/// The inner type of the [`Pool`].
struct PoolInner<T, B> {
    /// 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: B,

    /// 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, B> {
    /// 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, B>>,
}

impl<T, B> Pool<T, B> {
    /// Create a new [`Pool`] with a synchronous object builder.
    ///
    /// 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
    /// [`Pool::size`] to get the actual size of the pool. The minimum size is
    /// also 8.
    ///
    /// Because the object builder is synchronous (i.e. it implements
    /// [`SyncBuilder`]), the [`Pool::try_get`] method is sync.
    ///
    /// ```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(desired_size: usize, builder: B) -> Pool<T, Sync<B>>
    where
        B: SyncBuilder<T>,
    {
        Pool::new_inner(desired_size, Sync(builder))
    }

    /// Create a new [`Pool`] with an asynchronous object builder.
    ///
    /// 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
    /// [`Pool::size`] to get the actual size of the pool. The minimum size is
    /// also 8.
    ///
    /// Because the object builder is synchronous (i.e. it implements
    /// [`AsyncBuilder`]), the [`Pool::try_get`](#method.try_get-1) method is
    /// async.
    ///
    /// ```rust
    /// use std::future::ready;
    ///
    /// 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_with_async_builder(10, || ready(T::default));
    /// assert_eq!(pool.size(), 16);
    ///
    /// // The smallest size is 8.
    /// let smallest_pool = Pool::new_with_async_builder(1, || ready(T::default));
    /// assert_eq!(smallest_pool.size(), 8);
    /// ```
    pub fn new_with_async_builder(
        desired_size: usize,
        builder: B,
    ) -> Pool<T, Async<B>>
    where
        B: AsyncBuilder<T>,
    {
        Pool::new_inner(desired_size, Async(builder))
    }

    fn new_inner(desired_size: usize, builder: B) -> Self {
        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,
                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, B> {
        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();
    }

    /// Like [`Pool::try_get`] but will wait on an object to be available, i.e.
    /// to be in the pool.
    ///
    /// [`Get`] resolves to `Err` if this [`Pool`] has been closed while this
    /// future was waiting —see [`Pool::close`]—, or if the builder failed to
    /// build an object, otherwise it resolves to `Ok(`[`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 Ok(object) = pool.get().await {
    ///     // Yepee!
    /// } else {
    ///     // Oh no, the pool has been closed while awaiting…
    /// }
    /// # })
    /// ```
    pub fn get(&self) -> Get<T, B> {
        Get::new(&self.inner)
    }
}

impl<T, B> Pool<T, Sync<B>>
where
    B: SyncBuilder<T>,
{
    /// Try to synchronously get an [`Object`] from this pool.
    ///
    /// It errors with [`TryGetError`] if:
    ///
    /// - all objects are outside the pool,
    /// - the pool is closed,
    /// - the builder fails to build an object.
    ///
    /// ```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, Sync<B>>, TryGetError<B::Error>> {
        let (position, maybe_object_value) = match self.inner.try_acquire() {
            Ok(result) => result,
            Err(TryAcquireError::Closed) => {
                return Err(TryGetError::Closed);
            }
            Err(TryAcquireError::AllSlotsOccupied) => {
                return Err(TryGetError::AllSlotsOccupied);
            }
        };

        let object_value = self
            .inner
            .build_object_value_if_necessary(&position, maybe_object_value)
            .map_err(TryGetError::Builder)?;

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

impl<T, B> Pool<T, Async<B>>
where
    B: AsyncBuilder<T>,
{
    /// Try to synchronously get an [`Object`] from this pool.
    ///
    /// This method is marked as `async` because the builder is asynchronous.
    ///
    /// It errors with [`TryGetError`] if:
    ///
    /// - all objects are outside the pool,
    /// - the pool is closed,
    /// - the builder fails to build an object.
    ///
    /// ```rust
    /// # use flaque::Pool;
    /// use std::future::ready;
    /// #
    /// # type T = bool;
    /// #
    /// # futures_executor::block_on(async {
    /// // Create a pool.
    /// let pool = Pool::new_with_async_builder(8, || ready(T::default));
    ///
    /// // Try to get an object.
    /// if let Ok(object) = pool.try_get().await {
    ///     // Yepee!
    /// }
    /// # })
    /// ```
    ///
    /// See [`Pool::get`] for the asynchronous flavour.
    pub async fn try_get(
        &self,
    ) -> Result<Object<T, Async<B>>, TryGetError<B::Error>> {
        let (position, maybe_object_value) = match self.inner.try_acquire() {
            Ok(result) => result,
            Err(TryAcquireError::Closed) => {
                return Err(TryGetError::Closed);
            }
            Err(TryAcquireError::AllSlotsOccupied) => {
                return Err(TryGetError::AllSlotsOccupied);
            }
        };

        let object_value = self
            .inner
            .build_object_value_if_necessary(position, maybe_object_value)
            .await
            .map_err(TryGetError::Builder)?;

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

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

impl<T, B> PoolInner<T, B> {
    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_acquire(&self) -> Result<(Position, Option<T>), TryAcquireError> {
        if self.is_closed() {
            return Err(TryAcquireError::Closed);
        }

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

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

            let maybe_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()
                })
            };

            return Ok((object_position, maybe_object_value));
        }

        Err(TryAcquireError::AllSlotsOccupied)
    }

    fn try_release(
        &self,
        object_position: &Position,
        object_value: T,
    ) -> Result<(), TryReleaseError> {
        let Some(slots) = self.blocks.get(object_position.block_index) else {
            return Err(TryReleaseError::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));
                }
            });
        };

        self.release_slot(object_position, slots);

        Ok(())
    }

    fn release_slot(&self, object_position: &Position, slots: &Slots) {
        // Now we can release the slot.
        slots.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();
    }

    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();
        }
    }
}

impl<T, B> PoolInner<T, Sync<B>>
where
    B: SyncBuilder<T>,
{
    fn build_object_value_if_necessary(
        &self,
        position: &Position,
        maybe_object_value: Option<T>,
    ) -> Result<T, B::Error> {
        Ok(match maybe_object_value {
            Some(object_value) => object_value,
            None => match self.builder.build_object() {
                Ok(object_value) => object_value,
                Err(builder_error) => {
                    self.release_slot(
                        position,
                        &self.blocks[position.block_index],
                    );

                    return Err(builder_error);
                }
            },
        })
    }
}

impl<T, B> PoolInner<T, Async<B>>
where
    B: AsyncBuilder<T>,
{
    async fn build_object_value_if_necessary(
        &self,
        position: Position,
        maybe_object_value: Option<T>,
    ) -> Result<T, B::Error> {
        Ok(match maybe_object_value {
            Some(object_value) => object_value,
            None => match self.builder.build_object().await {
                Ok(object_value) => object_value,
                Err(builder_error) => {
                    self.release_slot(
                        &position,
                        &self.blocks[position.block_index],
                    );

                    return Err(builder_error);
                }
            },
        })
    }
}

unsafe impl<T, B> std::marker::Send for PoolInner<T, B>
where
    T: std::marker::Send,
    B: std::marker::Send,
{
}

unsafe impl<T, B> std::marker::Sync for PoolInner<T, B>
where
    T: std::marker::Sync,
    B: std::marker::Sync,
{
}

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

impl<T, B> Drop for Object<T, B> {
    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, B> Deref for Object<T, B> {
    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, B> DerefMut for Object<T, B> {
    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`].
#[derive(Clone, Copy)]
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
    }
}

/// 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::{future::ready, ops::Not, time::Duration};

    use self::sys::{block_on, thread};
    use super::{GetError, Pool, TryGetError, 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_try_get_builder_fails() {
        use std::sync::{
            Arc,
            atomic::{AtomicBool, Ordering},
        };

        run(|| {
            let fail_now = Arc::new(AtomicBool::new(false));
            let builder = {
                let fail_now = fail_now.clone();

                move || {
                    if fail_now.load(Ordering::SeqCst) {
                        Err(())
                    } else {
                        Ok(T { x: 0 })
                    }
                }
            };
            let pool = Pool::<T, _>::new(8, builder);

            let _object_0 = pool.try_get().unwrap();
            let _object_1 = pool.try_get().unwrap();
            let _object_2 = pool.try_get().unwrap();

            fail_now.store(true, Ordering::SeqCst);

            assert!(matches!(pool.try_get(), Err(TryGetError::Builder(())),));

            fail_now.store(false, Ordering::SeqCst);

            // We can fetch all the 5 remaining objects, because the previous
            // `try_get` released the slot.

            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();

            // Impossible to acquire a new slot.
            assert!(matches!(
                pool.try_get(),
                Err(TryGetError::AllSlotsOccupied),
            ));

            // Release an object. The builder will fail to build a new one, but
            // the builder will not be called in that case because `object_3` is
            // already initialised!
            drop(object_3);
            fail_now.store(true, Ordering::SeqCst);
            let _object_3 = pool.try_get().unwrap();
        })
    }

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

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

                // 8 slots maximum: they are all occupied.
                assert!(pool.try_get().await.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;
            });

            let _object_0 = block_on(async {
                let object = pool.get().await.unwrap();

                assert_eq!(object.x, 7);

                object
            });

            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();
        })
    }

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

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

                let object_0 = pool.get().await.unwrap();
                assert_eq!(object_0.x, 7);

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

                // 8 slots maximum: they are all occupied.
                assert!(pool.try_get().await.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();
            })
        })
    }

    /// `Get` can only return `AcquireError::Builder` if a new object must be
    /// created, so if there is at least one vacant slot. In case all slots are
    /// occupied, `Get` will never return `AcquireError::Builder` as no object
    /// need to be built. It simplifies this test: we don't need to spawn
    /// particular thread to get concurrency. All we need is to ensure `get`
    /// returns the correct error in the first case.
    #[test]
    fn test_get_builder_fails() {
        use std::sync::{
            Arc,
            atomic::{AtomicBool, Ordering},
        };

        run(|| {
            let fail_now = Arc::new(AtomicBool::new(false));
            let builder = {
                let fail_now = fail_now.clone();

                move || {
                    if fail_now.load(Ordering::SeqCst) {
                        Err(())
                    } else {
                        Ok(T { x: 0 })
                    }
                }
            };
            let pool = Arc::new(Pool::<T, _>::new(8, builder));

            block_on(async {
                let _object_0 = pool.get().await.unwrap();
                let _object_1 = pool.get().await.unwrap();
                let _object_2 = pool.get().await.unwrap();

                fail_now.store(true, Ordering::SeqCst);

                assert!(
                    matches!(pool.get().await, Err(GetError::Builder(())),)
                );

                fail_now.store(false, Ordering::SeqCst);

                // We can fetch all the 5 remaining objects, because the
                // previous `get` released the slot.

                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();

                // Release an object. The builder will fail to build a new one,
                // but the builder will not be called in that case because
                // `object_3` is already initialised!
                drop(object_3);
                fail_now.store(true, Ordering::SeqCst);
                let _object_3 = pool.get().await.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_err());
                    })
                }
            });

            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(TryGetError::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();
        })
    }
}