#![deny(missing_docs)]
//! Global reference-counted, thread-sharable, immutable heap

use std::any::Any;
use std::cell::UnsafeCell;
use std::marker::PhantomData;
use std::ops::Deref;
use std::sync::atomic::{self, AtomicUsize, Ordering};

use bunch::Bunch;
use crossbeam_channel::{self, Receiver, Sender};
use lazy_static::lazy_static;

lazy_static! {
    static ref GARBO: Garbo = { Garbo::new() };
}

/// Put a value on the heap, returning a Handle
pub fn put<T: 'static>(t: T) -> Handle<T> {
    GARBO.put(t)
}

enum SlotInner {
    Free,
    Occupied {
        contains: Box<dyn Any>,
        refcount: AtomicUsize,
    },
}

struct Slot(UnsafeCell<SlotInner>);

#[derive(Clone)]
struct SlotRef(*const Slot);

unsafe impl Send for SlotRef {}
unsafe impl<T> Send for Handle<T> {}

impl SlotRef {
    fn new(slot: &Slot) -> Self {
        SlotRef(slot)
    }

    fn inner(&self) -> &mut SlotInner {
        unsafe { &mut *(*self.0).0.get() }
    }
}

struct Garbo {
    slots: Bunch<Slot>,
    receiver: Receiver<SlotRef>,
    sender: Sender<SlotRef>,
}

/// A handle pointing at a value in the heap
pub struct Handle<T> {
    slot: SlotRef,
    sender: Sender<SlotRef>,
    _marker: PhantomData<T>,
}

impl<T> Clone for Handle<T> {
    fn clone(&self) -> Self {
        if let SlotInner::Occupied { ref refcount, .. } = self.slot.inner() {
            refcount.fetch_add(1, Ordering::Relaxed);
            Handle {
                slot: self.slot.clone(),
                sender: self.sender.clone(),
                _marker: PhantomData,
            }
        } else {
            unreachable!("Attempt at cloning an empty slot")
        }
    }
}

impl<T> Drop for Handle<T> {
    fn drop(&mut self) {
        if let SlotInner::Occupied { refcount, .. } = self.slot.inner() {
            if refcount.fetch_sub(1, Ordering::Release) != 0 {
                return;
            }

            // The fence is neccesary to prevent reordering of use of the data and
            // deletion of the data.
            // See the drop implementation of `std::sync::Arc` for a more elaborate
            // explanation.

            atomic::fence(Ordering::Acquire);

            *self.slot.inner() = SlotInner::Free;

            // Here we send a message to the main Garbo instance that this slot is
            // now up for grabs

            self.sender.send(self.slot.clone()).expect("broken channel");
        } else {
            // Dropping a Free slot is a no-op
        }
    }
}

impl<T: 'static> Deref for Handle<T> {
    type Target = T;

    fn deref(&self) -> &Self::Target {
        if let SlotInner::Occupied { contains, .. } = self.slot.inner() {
            contains.downcast_ref().expect("invalid type conversion")
        } else {
            panic!("Deref of freed slot")
        }
    }
}

impl Slot {
    fn new<T: 'static>(t: T) -> Self {
        Slot(UnsafeCell::new(SlotInner::Occupied {
            contains: Box::new(t),
            refcount: AtomicUsize::new(0),
        }))
    }
}

impl Default for Garbo {
    fn default() -> Self {
        let (sender, receiver) = crossbeam_channel::unbounded();
        Garbo {
            slots: Default::default(),
            sender,
            receiver,
        }
    }
}

impl Garbo {
    fn new() -> Self {
        Default::default()
    }

    pub fn put<T: 'static>(&self, t: T) -> Handle<T> {
        // check if any slots have reported themselves empty
        match self.receiver.try_recv() {
            Ok(slot_ref) => {
                let mut_slot = slot_ref.inner();
                *mut_slot = SlotInner::Occupied {
                    contains: Box::new(t),
                    refcount: AtomicUsize::new(0),
                };

                Handle {
                    slot: slot_ref.clone(),
                    sender: self.sender.clone(),
                    _marker: PhantomData,
                }
            }
            Err(_) => {
                let slot = Slot::new(t);
                let slot_ref = SlotRef::new(self.slots.push(slot));
                Handle {
                    slot: slot_ref,
                    sender: self.sender.clone(),
                    _marker: PhantomData,
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use crate as garbo;
    use std::sync::Arc;

    #[test]
    fn it_works() {
        let arc = Arc::new(3);
        {
            let handle = garbo::put(arc.clone());
            {
                assert_eq!(Arc::strong_count(&arc), 2);

                let another_handle = handle.clone();

                // cloning the handle does not increase the strong count of the arc
                assert_eq!(Arc::strong_count(&arc), 2);

                assert_eq!(**handle, 3);
                assert_eq!(**another_handle, 3);
            }
            assert_eq!(**handle, 3);

            assert_eq!(Arc::strong_count(&arc), 2);
        }
        // all handles out of scope, the arc in the shared heap should be free
        assert_eq!(Arc::strong_count(&arc), 1);
    }

    #[test]
    fn stress() {
        let n = 1_000_000;

        let mut handles = vec![];

        // thread constantly allocating and deallocating

        handles.push(std::thread::spawn(move || {
            for i in 0..n {
                let handle = garbo::put(i);
                assert_eq!(*handle, i)
            }
        }));

        // write in one thread, read in the other

        let (sender, receiver) = crossbeam_channel::unbounded();

        handles.push(std::thread::spawn(move || {
            for i in 0..n {
                let handle = garbo::put(i);
                sender.send((handle, i)).unwrap()
            }
        }));

        handles.push(std::thread::spawn(move || {
            for _ in 0..n {
                match receiver.recv() {
                    Ok((handle, i)) => assert_eq!(*handle, i),
                    Err(_) => panic!(),
                }
            }
        }));

        for thread in handles {
            thread.join().unwrap()
        }
    }
}