#![cfg(loom)]
#[allow(dead_code)]
#[path = "../examples/semaphore.rs"]
mod semaphore;

use std::{
    future::{Future, poll_fn},
    sync::{Arc, atomic::Ordering::SeqCst},
    task::Poll,
};

use loom::{future::block_on, sync::atomic::AtomicUsize, thread};
use semaphore::Semaphore;

#[test]
fn basic_usage() {
    const NUM: usize = 2;

    struct Shared {
        semaphore: Semaphore,
        active: AtomicUsize,
    }

    async fn actor(shared: Arc<Shared>) {
        let _permit = shared.semaphore.acquire().await.unwrap();
        let actual = shared.active.fetch_add(1, SeqCst);
        assert!(actual <= NUM - 1);

        let actual = shared.active.fetch_sub(1, SeqCst);
        assert!(actual <= NUM);
    }

    loom::model(|| {
        let shared = Arc::new(Shared {
            semaphore: Semaphore::new(NUM),
            active: AtomicUsize::new(0),
        });

        for _ in 0..NUM {
            let shared = shared.clone();

            thread::spawn(move || {
                block_on(actor(shared));
            });
        }

        block_on(actor(shared));
    });
}

#[test]
fn release() {
    loom::model(|| {
        let semaphore = Arc::new(Semaphore::new(1));

        {
            let semaphore = semaphore.clone();
            thread::spawn(move || {
                block_on(semaphore.acquire()).unwrap();
            });
        }

        block_on(semaphore.acquire()).unwrap();
    });
}

#[test]
fn basic_closing() {
    const NUM: usize = 2;

    loom::model(|| {
        let semaphore = Arc::new(Semaphore::new(1));

        for _ in 0..NUM {
            let semaphore = semaphore.clone();

            thread::spawn(move || {
                for _ in 0..2 {
                    block_on(semaphore.acquire()).map_err(|_| ())?;
                }

                Ok::<(), ()>(())
            });
        }

        semaphore.close();
    });
}

#[test]
fn concurrent_close() {
    const NUM: usize = 3;

    loom::model(|| {
        let semaphore = Arc::new(Semaphore::new(1));

        for _ in 0..NUM {
            let semaphore = semaphore.clone();

            thread::spawn(move || {
                block_on(semaphore.acquire()).map_err(|_| ())?;
                semaphore.close();

                Ok::<(), ()>(())
            });
        }
    });
}

#[test]
fn concurrent_cancel() {
    async fn poll_and_cancel(semaphore: Arc<Semaphore>) {
        let mut acquire1 = Some(semaphore.acquire());
        let mut acquire2 = Some(semaphore.acquire());
        poll_fn(|cx| {
            // poll the acquire future once, and then immediately throw
            // it away. this simulates a situation where a future is
            // polled and then cancelled, such as by a timeout.
            if let Some(acquire) = acquire1.take() {
                tokio::pin!(acquire);
                let _ = acquire.poll(cx);
            }
            if let Some(acquire) = acquire2.take() {
                tokio::pin!(acquire);
                let _ = acquire.poll(cx);
            }
            Poll::Ready(())
        })
        .await
    }

    loom::model(|| {
        let semaphore = Arc::new(Semaphore::new(0));
        let t1 = {
            let semaphore = semaphore.clone();
            thread::spawn(move || block_on(poll_and_cancel(semaphore)))
        };
        let t2 = {
            let semaphore = semaphore.clone();
            thread::spawn(move || block_on(poll_and_cancel(semaphore)))
        };
        let t3 = {
            let semaphore = semaphore.clone();
            thread::spawn(move || block_on(poll_and_cancel(semaphore)))
        };

        t1.join().unwrap();
        semaphore.add_permits(10);
        t2.join().unwrap();
        t3.join().unwrap();
    });
}

#[test]
fn batch() {
    let mut b = loom::model::Builder::new();
    b.preemption_bound = Some(1);

    b.check(|| {
        let semaphore = Arc::new(Semaphore::new(10));
        let active = Arc::new(AtomicUsize::new(0));
        let mut threads = vec![];

        for _ in 0..2 {
            let semaphore = semaphore.clone();
            let active = active.clone();

            threads.push(thread::spawn(move || {
                for n in &[4, 10, 8] {
                    let _permits = block_on(semaphore.acquire_many(*n)).unwrap();

                    active.fetch_add(*n as usize, SeqCst);

                    let num_active = active.load(SeqCst);
                    assert!(num_active <= 10);

                    thread::yield_now();

                    active.fetch_sub(*n as usize, SeqCst);
                }
            }));
        }

        for thread in threads.into_iter() {
            thread.join().unwrap();
        }

        assert_eq!(10, semaphore.available_permits());
    });
}

#[test]
fn release_during_acquire() {
    loom::model(|| {
        let semaphore = Arc::new(Semaphore::new(10));
        let permits = semaphore
            .try_acquire_many(8)
            .expect("try_acquire should succeed; semaphore uncontended");
        let semaphore2 = semaphore.clone();
        let thread = thread::spawn(move || block_on(semaphore2.acquire_many(4)).unwrap().forget());

        drop(permits);
        thread.join().unwrap();
        semaphore.add_permits(4);
        assert_eq!(10, semaphore.available_permits());
    })
}

#[test]
fn concurrent_permit_updates() {
    loom::model(move || {
        let semaphore = Arc::new(Semaphore::new(5));
        let t1 = {
            let semaphore = semaphore.clone();
            thread::spawn(move || semaphore.add_permits(3))
        };
        let t2 = {
            let semaphore = semaphore.clone();
            thread::spawn(move || {
                semaphore
                    .try_acquire()
                    .expect("try_acquire should succeed")
                    .forget();
            })
        };
        let t3 = {
            let semaphore = semaphore.clone();
            thread::spawn(move || semaphore.forget_permits(2))
        };

        t1.join().unwrap();
        t2.join().unwrap();
        t3.join().unwrap();
        assert_eq!(semaphore.available_permits(), 5);
    })
}
