//! `bimodal_array` provides dual-mode access to a contiguous array. Access
//! is granted through two handler types:
//!
//! 1. [`ElementHandle<T>`] grants mutable access to a single element.
//! 2. [`ArrayHandle<T>`] grants exclusive mutable access to the entire array.
//!
//! Handles represent the ability to *attempt* a lock. The actual lock is held
//! by the returned guards:
//!
//! 1. when `ElementHandle::lock()` succeeds, it produces an [`ElementGuard<T>`]
//! 2. when `ArrayHandle::lock()` succeeds, it produces an [`ArrayGuard<T>`].
//! 3. when [`ArrayHandle::lock_map()`] succeeds, it produces an
//!    [`ArrayMap<T>`].
//!
//! Each [`ElementHandle`] is bound to a specific element. Guards for different
//! elements may be held concurrently. An [`ArrayGuard<T>`] provides
//! exclusive access to the entire array and cannot coexist with any
//! [`ElementGuard<T>`]. [`ArrayHandle::lock_map()`] can be used to derive
//! reference-carrying views while keeping the array lock held for the returned
//! mapped wrapper.
//!
//! This pattern is useful for scatter–gather style workloads where independent
//! workers update elements in parallel, followed by an orchestration step that
//! processes the entire array. Internally, lock state is coordinated via a
//! single `AtomicUsize`. As a result, `bimodal_array` may not scale well under
//! high contention.
//!
//! # Examples
//!
//! ```no_run
//! # use bimodal_array::bimodal_array;
//! # use std::thread;
//! const WORKER_COUNT: usize = 8;
//! let data = vec![0usize; WORKER_COUNT];
//! let (mut array_handle, element_handles) = bimodal_array(data);
//! let threads: Vec<_> = element_handles
//!     .into_iter()
//!     .enumerate()
//!     .map(|(idx, mut element)| {
//!         thread::spawn(move || {
//!             let mut guard = element.lock().unwrap();
//!             *guard = 10 - idx;
//!         })
//!     })
//!     .collect();
//! threads.into_iter().for_each(|t| t.join().unwrap());
//! let mut array_guard = array_handle.lock().unwrap();
//! array_guard.sort();
//! ```

use core::slice;
use std::collections::TryReserveError;
use std::mem::ManuallyDrop;
use std::ops::Deref;
use std::ops::DerefMut;
use std::ptr::NonNull;

#[cfg(feature = "loom")]
use loom::sync::atomic::{AtomicUsize, Ordering, fence};
#[cfg(not(feature = "loom"))]
use std::sync::atomic::{AtomicUsize, Ordering, fence};

/// Errors that can occur when constructing or using a `bimodal_array`.
#[derive(Debug)]
pub enum BimodalArrayError {
    /// The provided collection length is not supported.
    ///
    /// Internally, `bimodal_array` reserves certain length values for state
    /// encoding. Currently, a length of `usize::MAX` is rejected.
    UnsupportedLength,

    /// Allocation failed.
    ///
    /// This typically occurs when reserving memory for the element handle
    /// storage. The contained [`TryReserveError`] provides additional details
    /// about whether the failure was due to capacity overflow or allocator
    /// refusal.
    AllocationFailed(TryReserveError),

    /// The array-wide lock could not be acquired.
    ///
    /// This happens when one or more [`ElementGuard<T>`] values are currently
    /// active.
    CouldNotAcquireArrayLock,

    /// The element-level lock could not be acquired.
    ///
    /// This happens when an [`ArrayGuard<T>`] is currently active.
    CouldNotAcquireElementLock,
}

/// The LockState represents how the BimodalArray is currently used
#[repr(transparent)]
struct LockState(AtomicUsize);

/// This represents when a GLOBAL_LOCK_STATE is in effect
const ARRAY_LOCK_STATE: usize = usize::MAX;

impl LockState {
    fn new() -> Self {
        Self(AtomicUsize::new(0))
    }

    /// Releases the global lock that is currently in place. Should only be
    /// called when dropping a GlobalLock.
    fn release_array_lock(&self) {
        let old_state = self.0.fetch_xor(ARRAY_LOCK_STATE, Ordering::Release);
        debug_assert!(
            old_state == ARRAY_LOCK_STATE,
            "An accessor was active when trying to release the global lock"
        )
    }

    /// Releases an accessor lock. Should only be called when dropping the
    /// accessor lock.
    fn release_element_lock(&self) {
        let old_state = self.0.fetch_sub(1, Ordering::Release);
        debug_assert!(
            (old_state != ARRAY_LOCK_STATE),
            "Global lock was in place when trying to release accessor"
        )
    }

    /// Acquires an accessor lock. Will only work if it can successfully
    /// increase the number of accessors and there is no Global lock in
    /// place
    fn acquire_element_lock(&self) -> Result<&Self, BimodalArrayError> {
        let mut current_state = self.0.load(Ordering::Relaxed);
        loop {
            if current_state == ARRAY_LOCK_STATE {
                return Err(BimodalArrayError::CouldNotAcquireElementLock);
            }
            match self.0.compare_exchange(
                current_state,
                current_state + 1,
                Ordering::Acquire,
                Ordering::Relaxed,
            ) {
                Ok(_) => return Ok(self),
                Err(new_state) => current_state = new_state,
            }
        }
    }

    /// Acquires a global lock. Will only work if there are no more accessors in
    /// place
    fn acquire_array_lock(&self) -> Result<&Self, BimodalArrayError> {
        match self
            .0
            .compare_exchange(0, ARRAY_LOCK_STATE, Ordering::Acquire, Ordering::Relaxed)
        {
            Ok(_) => Ok(self),
            Err(_) => Err(BimodalArrayError::CouldNotAcquireArrayLock),
        }
    }
}

struct BimodalArrayInner<T> {
    state: LockState,
    owner_count: AtomicUsize,
    len: usize,
    capacity: usize,
    data: NonNull<T>,
}

// SAFETY: The inner allocation is only accessed through ArrayHandle and
// ElementHandle, whose guards enforce the aliasing rules. Moving the owner
// record between threads is sound when the stored elements can be sent.
unsafe impl<T: Send> Send for BimodalArrayInner<T> {}

fn allocate_inner<T>(
    len: usize,
    capacity: usize,
    data_ptr: NonNull<T>,
) -> NonNull<BimodalArrayInner<T>> {
    let inner = BimodalArrayInner::new(len, capacity, data_ptr);
    let leaked_inner = Box::leak(Box::new(inner));
    leaked_inner.into()
}

impl<T> BimodalArrayInner<T> {
    fn new(len: usize, capacity: usize, data: NonNull<T>) -> Self {
        let owner_count = len + 1;
        Self {
            state: LockState::new(),
            owner_count: AtomicUsize::new(owner_count),
            len,
            capacity,
            data,
        }
    }

    fn to_raw_parts(&self) -> (*mut T, usize, usize) {
        (self.data.as_ptr(), self.len, self.capacity)
    }
}

fn inner_free<T>(inner: NonNull<BimodalArrayInner<T>>) {
    let p = inner.as_ptr();
    // SAFETY: All handles are created from a live Box::leak allocation and
    // hold the same non-null inner pointer until their Drop calls this
    // function. The allocation is freed only when the refcount reaches zero.
    let prev = unsafe { (*p).owner_count.fetch_sub(1, Ordering::Release) };
    if prev == 1 {
        fence(Ordering::Acquire);

        // SAFETY: This is the final owner. The data pointer, length, and
        // capacity came from a Vec allocation, and all constructors expose the
        // inner only after all len elements are initialized.
        unsafe {
            let (data_ptr, len, capacity) = (*p).to_raw_parts();
            drop(Vec::from_raw_parts(data_ptr, len, capacity));
            drop(Box::from_raw(p));
        }
    }
}

/// A handle granting access to a single element of a `bimodal_array`.
///
/// An `ElementHandle` represents the capability to acquire a lock for a
/// specific element. The actual lock is held by the returned
/// [`ElementGuard`], which provides mutable access to the element.
///
/// # Locking Semantics
///
/// - Calling [`lock`](Self::lock) attempts to acquire element-level access.
/// - If successful, it returns an [`ElementGuard`] providing `&mut T`.
/// - If an [`ArrayGuard`] is currently active, lock acquisition fails with
///   [`BimodalArrayError::CouldNotAcquireElementLock`].
///
/// Multiple element guards for *different* elements may coexist.
/// Element-level exclusivity is enforced through Rust's borrowing rules,
/// as `lock` requires `&mut self`.
pub struct ElementHandle<T> {
    inner: NonNull<BimodalArrayInner<T>>,
    data_ptr: NonNull<T>,
}

// SAFETY: An ElementHandle owns only raw pointers into the shared allocation.
// Access to the element requires &mut self and an ElementGuard, and moving the
// handle to another thread is sound when the element type can be sent.
unsafe impl<T: Send> Send for ElementHandle<T> {}
// SAFETY: Shared references to ElementHandle do not provide mutable element
// access because lock requires &mut self. Sharing the handle is sound when
// shared references to the element type are sound.
unsafe impl<T: Sync> Sync for ElementHandle<T> {}

impl<T> ElementHandle<T> {
    fn new(inner: NonNull<BimodalArrayInner<T>>, data_ptr: NonNull<T>) -> Self {
        Self { inner, data_ptr }
    }

    /// Attempts to acquire mutable access to this element.
    ///
    /// Returns an [`ElementGuard`] on success. If an [`ArrayGuard`] is
    /// currently active, this method returns
    /// [`BimodalArrayError::CouldNotAcquireElementLock`].
    ///
    /// Lock acquisition is non-blocking.
    pub fn lock<'a>(&'a mut self) -> Result<ElementGuard<'a, T>, BimodalArrayError> {
        // SAFETY: ElementHandle values are created with an inner pointer from
        // allocate_inner. The refcount keeps the inner allocation alive for the
        // lifetime of this handle.
        let inner = unsafe { &*self.inner.as_ptr() };
        let state = inner.state.acquire_element_lock()?;
        Ok(ElementGuard {
            state,
            data_ptr: self.data_ptr,
        })
    }
}

impl<T> Drop for ElementHandle<T> {
    fn drop(&mut self) {
        inner_free(self.inner);
    }
}

/// A guard providing mutable access to a single element.
///
/// An `ElementGuard` is returned by [`ElementHandle::lock`]. While an
/// `ElementGuard` is alive:
///
/// - No [`ArrayGuard<T>`] can be acquired.
/// - No other `ElementGuard` for the same element can exist.
///
/// Guards for *different* elements may coexist concurrently.
///
/// The guard dereferences to `&mut T`, allowing direct mutation of the
/// underlying element.
///
/// # RAII Semantics
///
/// The element-level lock is released automatically when this guard is
/// dropped. This ensures that access is tied to the lifetime of the guard.
///
/// # Examples
///
/// ```no_run
/// # use bimodal_array::bimodal_array;
/// let data = vec![0];
/// let (_, mut elements) = bimodal_array(data);
///
/// let mut guard = elements[0].lock().unwrap();
/// *guard = 42;
/// // Lock is released here when `guard` is dropped.
/// ```
pub struct ElementGuard<'a, T> {
    state: &'a LockState,
    data_ptr: NonNull<T>,
}

impl<'a, T> Deref for ElementGuard<'a, T> {
    type Target = T;

    fn deref(&self) -> &Self::Target {
        // SAFETY: ElementGuard is only constructed after acquiring an element
        // lock. The handle's &mut self requirement prevents a second guard for
        // the same element, and array-wide access is excluded by LockState.
        unsafe { self.data_ptr.as_ref() }
    }
}

impl<'a, T> AsRef<T> for ElementGuard<'a, T> {
    fn as_ref(&self) -> &T {
        self.deref()
    }
}

impl<'a, T> AsMut<T> for ElementGuard<'a, T> {
    fn as_mut(&mut self) -> &mut T {
        self.deref_mut()
    }
}

impl<'a, T> DerefMut for ElementGuard<'a, T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        // SAFETY: ElementGuard uniquely owns element access for its lifetime:
        // no ArrayGuard can coexist, and another guard for this element cannot
        // be created without another mutable borrow of the same handle.
        unsafe { self.data_ptr.as_mut() }
    }
}

impl<'a, T> Drop for ElementGuard<'a, T> {
    fn drop(&mut self) {
        self.state.release_element_lock();
    }
}

/// A handle granting array-wide access to a `bimodal_array`.
///
/// An `ArrayHandle` represents the capability to acquire an exclusive lock for
/// the entire collection. The actual lock is held by the returned
/// [`ArrayGuard`], which provides mutable access to the full slice `&mut [T]`.
/// It can also produce an [`ArrayMap`] via [`lock_map`](Self::lock_map), which
/// stores mapped values while keeping the array lock held for the lifetime of
/// the returned wrapper.
///
/// # Locking Semantics
///
/// - Calling [`lock`](Self::lock) attempts to acquire array-wide exclusive
///   access.
/// - If successful, it returns an [`ArrayGuard`] providing `&mut [T]`.
/// - If one or more [`ElementGuard<T>`] values are currently active, lock
///   acquisition fails with [`BimodalArrayError::CouldNotAcquireArrayLock`].
///
/// Lock acquisition is non-blocking.
pub struct ArrayHandle<T> {
    inner: NonNull<BimodalArrayInner<T>>,
}

// SAFETY: ArrayHandle owns a raw pointer to the shared allocation. Moving it to
// another thread is sound when the contained elements can be sent.
unsafe impl<T: Send> Send for ArrayHandle<T> {}
// SAFETY: Shared references to ArrayHandle do not grant array access because
// lock requires &mut self. Sharing the handle is sound when shared references
// to the element type are sound.
unsafe impl<T: Sync> Sync for ArrayHandle<T> {}

impl<T> ArrayHandle<T> {
    /// Attempts to acquire exclusive mutable access to the entire array.
    ///
    /// Returns an [`ArrayGuard`] on success. If any element guards are
    /// currently active, this method returns
    /// [`BimodalArrayError::CouldNotAcquireArrayLock`].
    ///
    /// Lock acquisition is non-blocking.
    pub fn lock<'a>(&'a mut self) -> Result<ArrayGuard<'a, T>, BimodalArrayError> {
        let p = self.inner.as_ptr();
        // SAFETY: The inner pointer comes from allocate_inner and is kept alive
        // by this handle. Acquiring the array lock excludes all element guards.
        let state = unsafe { (*p).state.acquire_array_lock()? };
        // SAFETY: The inner allocation remains live for this handle, and the
        // array lock acquired above gives exclusive access to the full slice.
        let (data_ptr, len, _) = unsafe { (*p).to_raw_parts() };
        let data_ptr = NonNull::new(data_ptr).unwrap();
        Ok(ArrayGuard {
            state,
            data_ptr,
            len,
        })
    }

    /// Acquires the array-wide lock and maps each element into a new
    /// collection-backed wrapper.
    ///
    /// This method is intended for deriving views that may borrow from the
    /// underlying array. The array lock remains held until the returned
    /// [`ArrayMap`] is dropped.
    ///
    /// Returns an [`ArrayMap`] on success. If any element guards are currently
    /// active, this method returns
    /// [`BimodalArrayError::CouldNotAcquireArrayLock`].
    pub fn lock_map<'a, S, F: Fn(&'a mut T) -> S>(
        &'a mut self,
        mapper: F,
    ) -> Result<ArrayMap<'a, S>, BimodalArrayError> {
        let p = self.inner.as_ptr();
        // SAFETY: The inner pointer comes from allocate_inner and is kept alive
        // by this handle. Acquiring the array lock excludes all element guards.
        let state = unsafe { (*p).state.acquire_array_lock()? };
        // SAFETY: The inner allocation remains live for this handle, and the
        // array lock acquired above gives exclusive access to the full slice.
        let (data_ptr, len, _) = unsafe { (*p).to_raw_parts() };
        // SAFETY: data_ptr and len describe the initialized array allocation,
        // and the array lock provides unique mutable access for this scope.
        let data = unsafe { slice::from_raw_parts_mut(data_ptr, len) };
        let data = data
            .iter_mut()
            .map(mapper)
            .collect::<Vec<_>>()
            .into_boxed_slice();
        Ok(ArrayMap {
            state,
            data: Some(data),
        })
    }
}

impl<T> Drop for ArrayHandle<T> {
    fn drop(&mut self) {
        inner_free(self.inner);
    }
}

/// A guard providing exclusive access to the entire array.
///
/// An `ArrayGuard` is returned by [`ArrayHandle::lock`]. While an
/// `ArrayGuard` is alive:
///
/// - No [`ElementGuard<T>`] can be acquired.
/// - No other `ArrayGuard` can exist.
///
/// The guard dereferences to `&mut [T]`, allowing full mutable access to
/// the underlying slice.
///
/// # RAII Semantics
///
/// The array-wide lock is released automatically when this guard is dropped.
/// This ensures that exclusivity is tied to the lifetime of the guard.
///
/// # Examples
///
/// ```no_run
/// # use bimodal_array::bimodal_array;
/// let data = vec![3, 1, 2];
/// let (mut array_handle, _) = bimodal_array(data);
///
/// let mut guard = array_handle.lock().unwrap();
/// guard.sort();
/// // Lock is released here when `guard` is dropped.
/// ```
pub struct ArrayGuard<'a, T> {
    state: &'a LockState,
    data_ptr: NonNull<T>,
    len: usize,
}

impl<'a, T> Deref for ArrayGuard<'a, T> {
    type Target = [T];

    fn deref(&self) -> &Self::Target {
        // SAFETY: ArrayGuard is only constructed after acquiring the array
        // lock, and it stores the initialized allocation pointer and length.
        unsafe { slice::from_raw_parts(self.data_ptr.as_ptr(), self.len) }
    }
}

impl<'a, T> DerefMut for ArrayGuard<'a, T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        // SAFETY: The array lock held by this guard excludes all element
        // guards and other array guards, so mutable access to the slice is
        // unique for the lifetime of this guard.
        unsafe { slice::from_raw_parts_mut(self.data_ptr.as_ptr(), self.len) }
    }
}

impl<'a, T> AsRef<[T]> for ArrayGuard<'a, T> {
    fn as_ref(&self) -> &[T] {
        self.deref()
    }
}

impl<'a, T> AsMut<[T]> for ArrayGuard<'a, T> {
    fn as_mut(&mut self) -> &mut [T] {
        self.deref_mut()
    }
}

impl<'a, T> Drop for ArrayGuard<'a, T> {
    fn drop(&mut self) {
        self.state.release_array_lock();
    }
}

/// A mapped wrapper returned by [`ArrayHandle::lock_map`].
///
/// `ArrayMap` stores mapped values derived from the locked array while keeping
/// the array-wide lock held for its entire lifetime. This makes it suitable
/// for reference-carrying views that must not outlive exclusive access to the
/// underlying array.
///
/// The mapped data can be accessed as a slice, but the wrapper does not expose
/// ownership-transfer operations for the stored values.
pub struct ArrayMap<'a, T> {
    state: &'a LockState,
    data: Option<Box<[T]>>,
}

impl<'a, T> AsRef<[T]> for ArrayMap<'a, T> {
    fn as_ref(&self) -> &[T] {
        self.data.as_deref().unwrap()
    }
}

impl<'a, T> AsMut<[T]> for ArrayMap<'a, T> {
    fn as_mut(&mut self) -> &mut [T] {
        self.data.as_mut().unwrap()
    }
}

impl<'a, T> Drop for ArrayMap<'a, T> {
    fn drop(&mut self) {
        drop(self.data.take());
        self.state.release_array_lock();
    }
}

/// Constructs a bimodal array from an owned `Vec<T>`.
///
/// This is the fallible constructor. On success, it returns:
///
/// - an [`ArrayHandle<T>`], which can be used to acquire an [`ArrayGuard<T>`]
///   for exclusive access to the full slice, and
/// - a `Vec<ElementHandle<T>>`, containing one handle per element, each of
///   which can be used to acquire an [`ElementGuard<T>`] for element-level
///   access.
///
/// The returned handles enable a scatter–gather access pattern: independent
/// workers can lock and update different elements concurrently, followed by an
/// orchestration step that locks the entire array for consolidation.
///
/// # Errors
///
/// Returns [`BimodalArrayError::UnsupportedLength`] if the length of `data` is
/// not supported by the internal state encoding (currently `usize::MAX`).
///
/// Returns [`BimodalArrayError::AllocationFailed`] if allocating fails.
pub fn try_bimodal_array<T>(
    data: Vec<T>,
) -> Result<(ArrayHandle<T>, Vec<ElementHandle<T>>), BimodalArrayError> {
    let len = data.len();
    if len == usize::MAX {
        return Err(BimodalArrayError::UnsupportedLength);
    }
    let mut element_handles = Vec::new();
    element_handles
        .try_reserve_exact(len)
        .map_err(BimodalArrayError::AllocationFailed)?;
    let (data_ptr, _len, capacity) = data.into_raw_parts();
    let data_ptr = NonNull::new(data_ptr).unwrap();
    let inner = allocate_inner(len, capacity, data_ptr);
    let array_handle = ArrayHandle { inner };
    element_handles.extend((0..len).map(|idx| {
        // SAFETY: idx is produced by 0..len, and data_ptr points to a Vec
        // allocation containing len initialized elements.
        let elem_ptr = unsafe { data_ptr.add(idx) };
        ElementHandle::new(inner, elem_ptr)
    }));
    Ok((array_handle, element_handles))
}

/// Constructs a bimodal array from an owned `Vec<T>`.
///
/// This is a convenience wrapper around [`try_bimodal_array`] that panics on
/// error. Prefer [`try_bimodal_array`] if construction failure should be
/// handled gracefully.
///
/// # Panics
///
/// Panics if [`try_bimodal_array`] returns an error (e.g. unsupported length or
/// allocation failure).
pub fn bimodal_array<T>(data: Vec<T>) -> (ArrayHandle<T>, Vec<ElementHandle<T>>) {
    try_bimodal_array(data).unwrap_or_else(|e| panic!("bimodal_array construction failed: {e:?}"))
}

/// A worker that owns the lifecycle of one element in a worker-backed
/// `bimodal_array`.
///
/// Each `ElementWorker` passed to [`bimodal_array_with_worker`] is moved into
/// its own thread. The worker first constructs the initial value for its array
/// slot with [`build`](Self::build). After that value has been written into the
/// array, the worker receives an [`ElementHandle`] for the same slot and
/// continues with [`work`](Self::work).
pub trait ElementWorker: Send + 'static {
    type T;

    /// Builds the initial value for this worker's array element.
    fn build(&self) -> Self::T;

    /// Runs worker logic after the element has been initialized.
    ///
    /// The provided handle points to the element produced by this worker's
    /// [`build`](Self::build) call.
    fn work(&self, handle: ElementHandle<Self::T>);
}

// Helper to move the pointer between threads.
struct SendablePtr<T>(NonNull<T>);

impl<T> Clone for SendablePtr<T> {
    fn clone(&self) -> Self {
        SendablePtr(self.0)
    }
}

impl<T> SendablePtr<T> {
    fn new(ptr: NonNull<T>) -> Self {
        Self(ptr)
    }

    fn into_ptr(self) -> NonNull<T> {
        self.0
    }
}

// SAFETY: SendablePtr only moves an address between worker threads. The caller
// is responsible for ensuring the pointed-to allocation outlives the transfer
// and that any accesses through cloned pointers are synchronized or disjoint.
unsafe impl<T> Send for SendablePtr<T> {}

/// Constructs a bimodal array whose elements are initialized by worker threads.
///
/// This constructor accepts one [`ElementWorker`] per array element. For each
/// worker, it spawns a thread, calls [`ElementWorker::build`] on that thread,
/// writes the returned value into the worker's array slot, and then calls
/// [`ElementWorker::work`] with an [`ElementHandle`] for that same slot.
///
/// The returned [`ArrayHandle`] is produced only after every worker has
/// completed its `build` step and every array slot has been initialized. Worker
/// `work` methods may still be running after this function returns, so attempts
/// to acquire the array-wide lock can fail while workers hold element guards.
///
/// Element order matches worker order: the worker at index `i` initializes and
/// receives the handle for element `i`.
///
/// # Panics
///
/// Panics if a worker thread fails to report that its element was initialized.
///
/// # Examples
///
/// ```no_run
/// use bimodal_array::bimodal_array_with_worker;
/// use bimodal_array::ElementHandle;
/// use bimodal_array::ElementWorker;
///
/// struct Worker {
///     value: usize,
/// }
///
/// impl ElementWorker for Worker {
///     type T = usize;
///
///     fn build(&self) -> Self::T {
///         self.value
///     }
///
///     fn work(&self, mut handle: ElementHandle<Self::T>) {
///         let mut guard = handle.lock().unwrap();
///         *guard += 1;
///     }
/// }
///
/// let workers = vec![Worker { value: 1 }, Worker { value: 2 }];
/// let mut array = bimodal_array_with_worker(workers);
///
/// // Workers may still be running, so production code should handle lock
/// // acquisition failure according to its scheduling model.
/// if let Ok(guard) = array.lock() {
///     assert_eq!(&*guard, &[2, 3]);
/// }
/// ```
pub fn bimodal_array_with_worker<W: ElementWorker>(builder: Vec<W>) -> ArrayHandle<W::T> {
    let len = builder.len();
    let (data_initialized_tx, data_initialized_rx) = std::sync::mpsc::channel::<()>();
    let mut data_vector = ManuallyDrop::new(Vec::<W::T>::with_capacity(len));
    let capacity = data_vector.capacity();
    let data_ptr = NonNull::new(data_vector.as_mut_ptr()).unwrap();
    let inner = allocate_inner(len, capacity, data_ptr);
    let inner_ptr = SendablePtr::new(inner);
    let data_ptr = SendablePtr::new(data_ptr);
    for (idx, builder) in builder.into_iter().enumerate() {
        let inner_ptr = inner_ptr.clone();
        let data_ptr = data_ptr.clone();
        let data_initialized_tx = data_initialized_tx.clone();
        std::thread::spawn(move || {
            let data = builder.build();
            let inner = inner_ptr.into_ptr();
            let data_ptr = data_ptr.into_ptr();
            // SAFETY: idx is unique for this worker and within 0..len. The
            // backing Vec was allocated with capacity len and has been kept
            // alive with ManuallyDrop, so writing exactly one element to this
            // slot initializes it without reading old contents.
            let slot_ptr: NonNull<W::T> = unsafe {
                let slot_ptr = data_ptr.add(idx);
                slot_ptr.as_ptr().write(data);
                slot_ptr
            };
            data_initialized_tx.send(()).unwrap();
            let element_handle: ElementHandle<W::T> = ElementHandle::new(inner, slot_ptr);
            builder.work(element_handle);
        });
    }
    (0..len).for_each(|_| data_initialized_rx.recv().unwrap());
    let array_handle: ArrayHandle<W::T> = ArrayHandle {
        inner: inner_ptr.into_ptr(),
    };
    array_handle
}
