//! [![crates.io version](https://img.shields.io/crates/v/safina-sync.svg)](https://crates.io/crates/safina-sync)
//! [![license: Apache 2.0](https://gitlab.com/leonhard-llc/safina-rs/-/raw/main/license-apache-2.0.svg)](http://www.apache.org/licenses/LICENSE-2.0)
//! [![unsafe forbidden](https://gitlab.com/leonhard-llc/safina-rs/-/raw/main/unsafe-forbidden-success.svg)](https://github.com/rust-secure-code/safety-dance/)
//! [![pipeline status](https://gitlab.com/leonhard-llc/safina-rs/badges/main/pipeline.svg)](https://gitlab.com/leonhard-llc/safina-rs/-/pipelines)
//!
//! This is a safe Rust library for sharing or sending data between async tasks.
//!
//! It is part of [`safina`](https://crates.io/crates/safina), a safe async runtime.
//!
//! # Features
//! - `forbid(unsafe_code)`
//! - Depends only on `std`
//! - Good test coverage (100%)
//! - Works with [`safina-executor`](https://crates.io/crates/safina-executor) or any async executor
//!
//! # Limitations
//! - Allocates
//!
//! # Documentation
//! https://docs.rs/safina-sync
//!
//! # Examples
//! ```rust
//! use std::sync::Arc;
//! use safina_async_test::async_test;
//! use safina_sync::Mutex;
//! # async fn some_async_fn() {}
//! # fn get_shared_data() -> Arc<Mutex<u32>> {
//! #     Arc::new(Mutex::new(0))
//! # }
//! # #[async_test]
//! # async fn test1() {
//! #     let addr = String::new();
//! let shared_counter: Arc<Mutex<u32>> = get_shared_data();
//! {
//!     let mut counter_guard = shared_counter.lock().await;
//!     *counter_guard += 1;
//!     // some_async_fn().await; // Cannot await while holding a MutexGuard.
//! }
//! some_async_fn().await; // Await is ok after releasing MutexGuard.
//! # }
//! ```
//!
//! # TO DO - Alternatives
//! - [async-lock](https://crates.io/crates/async-lock)
//!   - Contains a little `unsafe` code
//! - [futures-locks](https://crates.io/crates/futures-locks)
//!   - Contains a little `unsafe` code
//! - [futures-util](https://crates.io/crates/futures-util)
//!   - Very popular
//!   - Full of `unsafe`
//! - [tokio-sync](https://crates.io/crates/tokio-sync)
//!   - Very popular
//!   - Fast
//!   - Internally incredibly complicated
//!   - Full of `unsafe`
//!
//! # Changelog
//! - v0.1.0 - First published version
//!
//! # TO DO
//! - DONE - Implement `Mutex` with tests & docs
//! - DONE - Publish on crates.io
//! - Add `Barrier`
//! - Add `RwLock`
//! - Add `WaitableBool`
//! - Add `Promise`
//! - Add `Channel` (single receiver)
//! - Add `UnboundedChannel`
//! - Add `WaitableQueue` (multiple receivers)
//! - Add `UnboundedWaitableQueue`
//! - Add `Topic` (copies message to every receiver)
//!
//! # Release Process
//! 1. Edit `Cargo.toml` and bump version number.
//! 1. Run `./release.sh`
#![forbid(unsafe_code)]

use core::future::Future;
use core::ops::{Deref, DerefMut};
use core::pin::Pin;
use core::task::{Context, Poll};
use std::collections::VecDeque;
use std::sync::TryLockError;
use std::task::Waker;

/// An [RAII](https://doc.rust-lang.org/rust-by-example/scope/raii.html)
/// scoped lock of a [`Mutex`](struct.Mutex.html).
/// It automatically unlocks the mutex when dropped (falls out of scope).
///
/// You can access the data in the mutex through this guard's
/// [`Deref`](https://doc.rust-lang.org/stable/std/ops/trait.Deref.html)
/// and
/// [`DerefMut`](https://doc.rust-lang.org/stable/std/ops/trait.DerefMut.html)
/// implementations.
///
/// The struct is not
/// [`Send`](https://doc.rust-lang.org/stable/std/marker/trait.Send.html)
/// so you cannot await while holding it.
///
/// If a task panics while holding the struct, the underlying mutex becomes
/// ["poisoned"](https://doc.rust-lang.org/stable/std/sync/struct.Mutex.html#poisoning)
/// and subsequent calls to `lock` will panic.
pub struct MutexGuard<'a, T> {
    mutex: &'a Mutex<T>,
    value_guard: Option<std::sync::MutexGuard<'a, T>>,
}
impl<'a, T> MutexGuard<'a, T> {
    pub fn new(
        mutex: &'a Mutex<T>,
        value_guard: std::sync::MutexGuard<'a, T>,
    ) -> MutexGuard<'a, T> {
        let mut inner_guard = mutex.inner.lock().unwrap();
        assert!(!inner_guard.locked);
        inner_guard.locked = true;
        MutexGuard {
            mutex,
            value_guard: Some(value_guard),
        }
    }
}
impl<'a, T> Drop for MutexGuard<'a, T> {
    fn drop(&mut self) {
        let mut inner_guard = self.mutex.inner.lock().unwrap();
        assert!(inner_guard.locked);
        inner_guard.locked = false;
        self.value_guard.take();
        while let Some(waker) = inner_guard.wakers.pop_front() {
            waker.wake();
        }
    }
}
impl<'a, T> Deref for MutexGuard<'a, T> {
    type Target = T;
    fn deref(&self) -> &Self::Target {
        self.value_guard.as_ref().unwrap().deref()
    }
}
impl<'a, T> DerefMut for MutexGuard<'a, T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        self.value_guard.as_mut().unwrap().deref_mut()
    }
}

pub struct LockFuture<'a, T> {
    mutex: &'a Mutex<T>,
}
impl<'a, T> Future for LockFuture<'a, T> {
    type Output = MutexGuard<'a, T>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        loop {
            match self.mutex.value.try_lock() {
                Ok(guard) => return Poll::Ready(MutexGuard::new(&self.mutex, guard)),
                Err(TryLockError::Poisoned(e)) => panic!("{}", e),
                Err(TryLockError::WouldBlock) => {}
            }
            let mut guard = self.mutex.inner.lock().unwrap();
            if guard.locked {
                // Mutex is locked.  Add our waker and sleep.
                guard.wakers.push_back(cx.waker().clone());
                return Poll::Pending;
            }
            // Mutex is now unlocked.  Try to acquire it again.
        }
    }
}

struct Inner {
    wakers: VecDeque<Waker>,
    locked: bool,
}

/// A wrapper around [std::sync::Mutex](https://doc.rust-lang.org/stable/std/sync/struct.Mutex.html)
/// with an async [`lock`](#method.lock) method.
pub struct Mutex<T> {
    inner: std::sync::Mutex<Inner>,
    value: std::sync::Mutex<T>,
}

impl<T> Mutex<T> {
    pub fn new(value: T) -> Mutex<T> {
        Self {
            inner: std::sync::Mutex::new(Inner {
                wakers: VecDeque::new(),
                locked: false,
            }),
            value: std::sync::Mutex::new(value),
        }
    }

    /// Acquires the mutex, sleeping the current task until it is able to do so.
    ///
    /// After this function returns, is the only task holding the lock.
    ///
    /// The returned guard is an
    /// [RAII](https://doc.rust-lang.org/rust-by-example/scope/raii.html)
    /// scoped lock.
    /// It automatically unlocks the mutex when dropped (falls out of scope).
    ///
    /// The returned guard is not
    /// [`Send`](https://doc.rust-lang.org/stable/std/marker/trait.Send.html)
    /// so you cannot `await` while holding it.
    ///
    /// If a task panics while holding the guard, the underlying mutex becomes
    /// ["poisoned"](https://doc.rust-lang.org/stable/std/sync/struct.Mutex.html#poisoning)
    /// and subsequent calls to `lock` will panic.
    pub async fn lock(&self) -> MutexGuard<'_, T> {
        LockFuture { mutex: self }.await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use core::ops::Range;
    use core::time::Duration;
    use safina_async_test::async_test;
    use safina_timer::sleep_for;
    use std::sync::Arc;
    use std::time::Instant;

    pub fn expect_elapsed(before: Instant, range_ms: Range<u64>) {
        if range_ms.is_empty() {
            panic!("invalid range {:?}", range_ms)
        }
        let elapsed = before.elapsed();
        let duration_range =
            Duration::from_millis(range_ms.start)..Duration::from_millis(range_ms.end);
        if !duration_range.contains(&elapsed) {
            panic!("{:?} elapsed, out of range {:?}", elapsed, duration_range);
        }
    }

    #[async_test]
    async fn should_unlock_immediately() {
        safina_executor::increase_threads_to(1);
        let before = Instant::now();
        let counter = Mutex::new(0u8);
        {
            let mut guard = counter.lock().await;
            *guard += 1;
        }
        let guard = counter.lock().await;
        assert_eq!(1u8, *guard);
        expect_elapsed(before, 0..1);
    }

    #[async_test]
    async fn should_unlocking_wakes_waiters() {
        safina_executor::increase_threads_to(1);
        safina_timer::start_timer_thread();
        let before = Instant::now();
        let counter = Arc::new(Mutex::new(0u8));
        let counter_clone = counter.clone();
        safina_executor::spawn(async move {
            let mut guard = counter_clone.lock().await;
            *guard += 1;
        });
        // Delay so other task can get the lock
        sleep_for(Duration::from_millis(100)).await;
        let guard = counter.lock().await;
        assert_eq!(1u8, *guard);
        expect_elapsed(before, 100..200);
    }

    #[async_test]
    async fn should_handle_many_waiters() {
        safina_executor::increase_threads_to(10);
        safina_timer::start_timer_thread();
        let before = Instant::now();
        const NUM_WAITERS: u16 = 10000;
        let counter = Arc::new(Mutex::new(0u16));
        let mut receivers = Vec::new();
        for _ in 0..NUM_WAITERS {
            let counter_clone = counter.clone();
            let (sender, receiver) = std::sync::mpsc::channel::<()>();
            receivers.push(receiver);
            safina_executor::spawn(async move {
                let mut guard = counter_clone.lock().await;
                *guard += 1;
                sender.send(()).unwrap();
            });
        }
        for receiver in receivers {
            receiver.recv().unwrap();
        }
        let guard = counter.lock().await;
        assert_eq!(NUM_WAITERS, *guard);
        expect_elapsed(before, 0..1000);
    }
}
