bottle 0.1.0

use std::sync::{Arc, Weak};
use std::sync::atomic::{self, Ordering::*};
use std::ptr::NonNull;
use std::ops::{CoerceUnsized, DispatchFromDyn};
use std::marker::Unsize;
use std::cell::UnsafeCell;
use std::alloc::{Global, Alloc, Layout, handle_alloc_error};

use crate::{
    thread, Thread, Signal, Handler, MutHandler, Future, Emitter, Expose, Subscription
};

mod actor;
use actor::Actor;

/// Check is the given pointer is dangling.
/// In our case, dangling pointers have value usize::MAX.
fn is_dangling<T: ?Sized>(ptr: NonNull<T>) -> bool {
    let address = ptr.as_ptr() as *mut () as usize;
    address == std::usize::MAX
}

/// A soft limit on the amount of references that may be made to an `Remote`.
///
/// Going above this limit will abort your program (although not necessarily) at
/// _exactly_ `MAX_REFCOUNT + 1` references.
const MAX_REFCOUNT: usize = (std::isize::MAX) as usize;

/// A thread-safe actor pointer.
///
/// ## Thread Safety
///
/// `Remote<T>` will implement [`Send`] and [`Sync`] even if `T` does not.
/// This is possible because in most cases, the underlying data is never accessed directly.
/// The only way to access the data directly is via [`Remote::read`] and [`Remote::write`],
/// but thoses two functions are only implemented when `T` implements [`Send`] and [`Sync`].
///
/// ## Accessing the underlying value
///
/// The value is indirectly accessed via message passing using the non-blocking methods
/// [`send`] and [`send_mut`].
///
/// ## Breaking cycles with `WeakRemote`
///
/// The [`downgrade`][downgrade] method can be used to create a non-owning
/// [`WeakRemote`][weak] pointer. A [`WeakRemote`][weak] pointer can be [`upgrade`][upgrade]d
/// to an `Remote`, but this will return [`None`] if the value has already been
/// dropped.
///
/// A cycle between `Remote` pointers will never be deallocated. For this reason,
/// [`WeakRemote`] is used to break cycles. For example, a tree could have
/// strong `Remote` pointers from parent nodes to children, and [`WeakRemote`]
/// pointers from children back to their parents.
///
/// ## Cloning references
///
/// Creating a new reference from an existing reference remote pointer is done using the
/// `Clone` trait implemented for [`Remote<T>`][arc] and [`WeakRemote<T>`][weak].
pub struct Remote<T: ?Sized> {
    ptr: NonNull<Inner<T>>
}

impl<T: ?Sized + Unsize<U>, U: ?Sized> CoerceUnsized<Remote<U>> for Remote<T> {}
impl<T: ?Sized + Unsize<U>, U: ?Sized> DispatchFromDyn<Remote<U>> for Remote<T> {}

unsafe impl<T: ?Sized> Sync for Remote<T> {}
unsafe impl<T: ?Sized> Send for Remote<T> {}

pub struct WeakRemote<T: ?Sized> {
    ptr: NonNull<Inner<T>>
}

impl<T: ?Sized + Unsize<U>, U: ?Sized> CoerceUnsized<WeakRemote<U>> for WeakRemote<T> {}
impl<T: ?Sized + Unsize<U>, U: ?Sized> DispatchFromDyn<WeakRemote<U>> for WeakRemote<T> {}

unsafe impl<T: ?Sized> Sync for WeakRemote<T> {}
unsafe impl<T: ?Sized> Send for WeakRemote<T> {}

impl<T: ?Sized> PartialEq for Remote<T> {
	fn eq(&self, other: &Remote<T>) -> bool {
		self.ptr == other.ptr
	}
}

impl<T: ?Sized> Eq for Remote<T> {
	// ...
}


/// A delivery is a callback function intended to call a method of an actor.
/// The callback does not need to be `Send`, however for it to be safe,
/// one must ensure that the only non-`Send` data used is the target actor, since we know that
/// it won't be touched outside of it's own thread.
struct Delivery<F: FnOnce() -> ()> {
    callback: Option<F>,
    receiver: Receiver
}

unsafe impl<F: FnOnce() -> ()> Send for Delivery<F> {}

impl<F: FnOnce() -> ()> thread::Query for Delivery<F> {
    fn process(&mut self) {
        let mut callback = None;
        std::mem::swap(&mut self.callback, &mut callback);
        match callback {
            Some(f) => (f)(),
            None => ()
        }
    }

    fn receiver(&self) -> Option<Receiver> {
        Some(self.receiver)
    }
}

impl<T: 'static + ?Sized> Remote<T> {
    /// Create from a constructor.
    /// Note that the value is not created right away. This actually sends a query to the target
    /// thread. The value will be initialized when the thread process the query.
    /// Until then, any direct access to the value through [`read`] or [`write`] will fail.
    /// If it is executed in the target thread, then value is directly initialized.
    pub fn from<F: 'static + FnOnce(Remote<T>) -> T + Send>(thread: &Arc<Thread>, constructor: F) -> Remote<T> where T: Sized {
        let mut inner : Box<Inner<T>> = Box::new(Inner {
            strong: atomic::AtomicUsize::new(1),
            weak: atomic::AtomicUsize::new(1),
            thread: Arc::downgrade(thread),
            actor: UnsafeCell::new(unsafe { Actor::uninitialized() })
        });

        let remote = Remote {
            ptr: Box::into_raw_non_null(inner)
        };

        let this = remote.clone();

        match Thread::current() {
            Some(current_thread) if Arc::ptr_eq(&current_thread, thread) => {
                // no need to send a query, since we are in the target thread, execute the constructor
                // directly.
                unsafe {
                    // It is safe since it is the very first thing that touches the actor.
                    let actor_mut: &mut Actor<T> = &mut *this.inner().actor.get();
                    actor_mut.initialize(constructor(this))
                }
            },
            _ => {
                thread.query(thread::Callback::new(move || {
                    unsafe {
                        // It is safe since it is the very first thing that touches the actor.
                        let actor_mut: &mut Actor<T> = &mut *this.inner().actor.get();
                        actor_mut.initialize(constructor(this))
                    }
                }));
            }
        }

        remote
    }

    /// Create a remote pointer on the given thread.
    pub fn new(thread: &Arc<Thread>, t: T) -> Remote<T> where T: Send + Sized {
        // Start the weak pointer count as 1 which is the weak pointer that's
        // held by all the strong pointers (kinda), see std/rc.rs for more info
        let mut inner : Box<Inner<T>> = Box::new(Inner {
            strong: atomic::AtomicUsize::new(1),
            weak: atomic::AtomicUsize::new(1),
            thread: Arc::downgrade(thread),
            actor: UnsafeCell::new(Actor::new(t))
        });

        Remote {
            ptr: Box::into_raw_non_null(inner)
        }
    }

    /// Create a remote pointer on the current thread.
    /// Return back the value as an error when there are no current thread.
    pub fn local(t: T) -> std::result::Result<Remote<T>, T> where T: Sized {
        match Thread::current() {
            Some(thread) => {
                // Start the weak pointer count as 1 which is the weak pointer that's
                // held by all the strong pointers (kinda), see std/rc.rs for more info
                let mut inner : Box<Inner<T>> = Box::new(Inner {
                    strong: atomic::AtomicUsize::new(1),
                    weak: atomic::AtomicUsize::new(1),
                    thread: Arc::downgrade(&thread),
                    actor: UnsafeCell::new(Actor::new(t))
                });

                Ok(Remote {
                    ptr: Box::into_raw_non_null(inner)
                })
            },
            None => Err(t)
        }
    }

	pub fn send<S: Signal + 'static>(&self, msg: S) -> Future<S::Response> where T: Handler<S> {
        match self.inner().thread.upgrade() {
            Some(thread) => {
                let (sender, receiver) = crossbeam_channel::bounded(1);

                let this = self.clone();
                let delivery = Delivery {
                    // Note that this function is not `Send` because of T.
                    // However in our case `Delivery` makes it `Send` for us, since it is safe
                    // in our case (c.f. `Delivery` to have more infos).
                    callback: Some(move || {
                        let actor = unsafe {
                            // since the remote pointer `this` is captured by the closure,
                            // we know it still lives, and that it is safe to dereference.
                            &*this.inner().actor.get()
                        };
                        sender.send(actor.data.handle(msg)).unwrap_or(())
                    }),
                    receiver: self.receiver()
                };

                thread.query(delivery);

                Future {
                    receiver: receiver
                }
            },
            None => panic!("thread died")
        }
	}

    pub fn send_mut<S: Signal + 'static>(&self, msg: S) -> Future<S::Response> where T: MutHandler<S> {
        match self.inner().thread.upgrade() {
            Some(thread) => {
                let (sender, receiver) = crossbeam_channel::bounded(1);

                let this = self.clone();
                let delivery = Delivery {
                    // Note that this function is not `Send` because of T.
                    // However in our case `Delivery` makes it `Send` for us, since it is safe
                    // in our case (c.f. `Delivery` to have more infos).
                    callback: Some(move || {
                        let actor = unsafe {
                            // since the remote pointer `this` is captured by the closure,
                            // we know it still lives, and that it is safe to dereference.
                            &mut *this.inner().actor.get()
                        };
                        sender.send(actor.data.handle_mut(msg)).unwrap_or(())
                    }),
                    receiver: self.receiver()
                };

                thread.query(delivery);

                Future {
                    receiver: receiver
                }
            },
            None => panic!("thread died")
        }
	}

	// pub fn thread(&self) -> &Weak<Thread> {
	// 	&self.thread
	// }

    // // ...
    // pub fn read(&self) -> LockResult<ActorReadGuard<T>> where T: Sync + Send {
    //     let (sender, receiver) = crossbeam_channel::bounded(1);
    //     // ...
    //     thread.query(thread::Callback::new(move || {
    //         unsafe {
    //             // It is safe since it is the very first thing that touches the actor.
    //             let actor_mut: &mut Actor<T> = &mut *this.inner().actor.get();
    //             actor_mut.initialize(constructor(this))
    //         }
    //     }));
    // }

    // pub fn subscribe_to<S: Signal, E: Emitter<S>>(&self, emitter: &E) where T: Handler<S> {
    //     let this = self.clone() as Remote<dyn Handler<S>>;
    //     emitter.subscribe(this)
    // }
    //
    // pub fn subscribe_mut_to<S: Signal, E: Emitter<S>>(&self, emitter: &E) where T: MutHandler<S> {
    //     let this = self.clone() as Remote<dyn MutHandler<S>>;
    //     emitter.subscribe_mut(this)
    // }
}

impl<T: ?Sized> Remote<T> {
    /// Return a pointer to the underlying value.
    pub fn ptr(&self) -> *const T {
        let actor = unsafe { &*self.inner().actor.get() };
        &actor.data as *const T
    }

    /// Return a mutable pointer to the underlying value.
    pub fn ptr_mut(&self) -> *mut T {
        let mut actor = unsafe { &mut *self.inner().actor.get() };
        &mut actor.data as *mut T
    }

    pub fn is_initialized(&self) -> bool {
        let actor = unsafe { &*self.inner().actor.get() };
        actor.is_initialized()
    }

    pub fn get<D>(&self) -> Option<D> where T: Expose<D> {
        if self.is_initialized() {
            match (self.inner().thread.upgrade(), Thread::current()) {
                (Some(thread), Some(current)) if Arc::ptr_eq(&thread, &current) => {
                    let actor = unsafe { &*self.inner().actor.get() };
                    Some(actor.data.get())
                },
                _ => None
            }
        } else {
            None
        }
    }

    pub fn receiver(&self) -> Receiver {
        let actor = unsafe { &*self.inner().actor.get() };
        Receiver {
            owner: &actor.data as *const T as *const (),
            inner: self.ptr.as_ptr() as *mut ()
        }
    }

    /// Creates a new [`WeakRemote`][weak] pointer to this value.
    pub fn downgrade(&self) -> WeakRemote<T> {
        // This Relaxed is OK because we're checking the value in the CAS
        // below.
        let mut cur = self.inner().weak.load(Relaxed);

        loop {
            // check if the weak counter is currently "locked"; if so, spin.
            if cur == std::usize::MAX {
                cur = self.inner().weak.load(Relaxed);
                continue;
            }

            // NOTE: this code currently ignores the possibility of overflow
            // into usize::MAX; in general both Rc and Arc need to be adjusted
            // to deal with overflow.

            // Unlike with Clone(), we need this to be an Acquire read to
            // synchronize with the write coming from `is_unique`, so that the
            // events prior to that write happen before this read.
            match self.inner().weak.compare_exchange_weak(cur, cur + 1, Acquire, Relaxed) {
                Ok(_) => {
                    // Make sure we do not create a dangling Weak
                    debug_assert!(!is_dangling(self.ptr));
                    return WeakRemote { ptr: self.ptr };
                }
                Err(old) => cur = old,
            }
        }
    }

    #[inline]
    pub(crate) fn inner(&self) -> &Inner<T> {
        // This unsafety is ok because while this remote is alive we're guaranteed
        // that the inner pointer is valid. Furthermore, we know that the
        // `Inner` structure itself is `Sync` because the inner data is
        // `Sync` as well, so we're ok loaning out an immutable pointer to these
        // contents.
        unsafe { self.ptr.as_ref() }
    }

    // Non-inlined part of `drop`.
    #[inline(never)]
    unsafe fn drop_slow(&mut self) {
        // Destroy the data at this time, even though we may not free the box
        // allocation itself (there may still be weak pointers lying around).
        std::ptr::drop_in_place(&mut self.ptr.as_mut().actor);

        if self.inner().weak.fetch_sub(1, Release) == 1 {
            atomic::fence(Acquire);
            std::alloc::Global.dealloc(self.ptr.cast(), Layout::for_value(self.ptr.as_ref()))
        }
    }
}

impl<T: 'static + ?Sized> Clone for Remote<T> {
    /// Makes a clone of the `Remote` pointer.
    ///
    /// This creates another pointer to the same inner value, increasing the
    /// strong reference count.
    #[inline]
    fn clone(&self) -> Remote<T> {
        // Using a relaxed ordering is alright here, as knowledge of the
        // original reference prevents other threads from erroneously deleting
        // the object.
        //
        // As explained in the [Boost documentation][1], Increasing the
        // reference counter can always be done with memory_order_relaxed: New
        // references to an object can only be formed from an existing
        // reference, and passing an existing reference from one thread to
        // another must already provide any required synchronization.
        //
        // [1]: (www.boost.org/doc/libs/1_55_0/doc/html/atomic/usage_examples.html)
        let old_size = self.inner().strong.fetch_add(1, Relaxed);

        // However we need to guard against massive refcounts in case someone
        // is `mem::forget`ing Arcs. If we don't do this the count can overflow
        // and users will use-after free. We racily saturate to `isize::MAX` on
        // the assumption that there aren't ~2 billion threads incrementing
        // the reference count at once. This branch will never be taken in
        // any realistic program.
        //
        // We abort because such a program is incredibly degenerate, and we
        // don't care to support it.
        if old_size > MAX_REFCOUNT {
            unsafe {
                std::process::abort();
            }
        }

        Remote {
            ptr: self.ptr
        }
    }
}

unsafe impl<#[may_dangle] T: ?Sized> Drop for Remote<T> {
    /// Drops the `Remote`.
    ///
    /// This will decrement the strong reference count. If the strong reference
    /// count reaches zero then the only other references (if any) are
    /// [`WeakRemote`], so we `drop` the inner value.
    #[inline]
    fn drop(&mut self) {
        // Because `fetch_sub` is already atomic, we do not need to synchronize
        // with other threads unless we are going to delete the object. This
        // same logic applies to the below `fetch_sub` to the `weak` count.
        if self.inner().strong.fetch_sub(1, Release) != 1 {
            return;
        }

        // This fence is needed to prevent reordering of use of the data and
        // deletion of the data.  Because it is marked `Release`, the decreasing
        // of the reference count synchronizes with this `Acquire` fence. This
        // means that use of the data happens before decreasing the reference
        // count, which happens before this fence, which happens before the
        // deletion of the data.
        //
        // As explained in the [Boost documentation][1],
        //
        // > It is important to enforce any possible access to the object in one
        // > thread (through an existing reference) to *happen before* deleting
        // > the object in a different thread. This is achieved by a "release"
        // > operation after dropping a reference (any access to the object
        // > through this reference must obviously happened before), and an
        // > "acquire" operation before deleting the object.
        //
        // In particular, while the contents of an Arc are usually immutable, it's
        // possible to have interior writes to something like a Mutex<T>. Since a
        // Mutex is not acquired when it is deleted, we can't rely on its
        // synchronization logic to make writes in thread A visible to a destructor
        // running in thread B.
        //
        // Also note that the Acquire fence here could probably be replaced with an
        // Acquire load, which could improve performance in highly-contended
        // situations. See [2].
        //
        // [1]: (www.boost.org/doc/libs/1_55_0/doc/html/atomic/usage_examples.html)
        // [2]: (https://github.com/rust-lang/rust/pull/41714)
        atomic::fence(Acquire);

        unsafe {
            self.drop_slow();
        }
    }
}

impl<S: Signal, T: 'static + ?Sized + Handler<Subscription<S>>> Emitter<S> for Remote<T> {
    /**
     * Subscribe to an emitter.
     */
    fn subscribe(&self, actor: Remote<dyn 'static + Handler<S>>) {
        self.send(Subscription::Const(actor));
    }

    /**
     * Subscribe mutabily to an emitter.
     */
    fn subscribe_mut(&self, actor: Remote<dyn 'static + MutHandler<S>>) {
        self.send(Subscription::Mut(actor));
    }
}

impl<T: 'static> WeakRemote<T> {
    /// Try to send a signal to the remote actor.
    /// Return None if the actor is dead.
    pub fn try_send<S: Signal + 'static>(&self, msg: S) -> Option<Future<S::Response>> where T: Handler<S> {
        match self.upgrade() {
            Some(remote) => Some(remote.send(msg)),
            None => None
        }
	}

    /// Try to send a signal to the remote (mutable) actor.
    /// Return None if the actor is dead.
    pub fn try_send_mut<S: Signal + 'static>(&self, msg: S) -> Option<Future<S::Response>> where T: MutHandler<S> {
        match self.upgrade() {
            Some(remote) => Some(remote.send_mut(msg)),
            None => None
        }
	}
}

impl<T: ?Sized> WeakRemote<T> {
    /// Constructs a new `WeakRemote<T>`, without allocating any memory.
    /// Calling [`upgrade`] on the return value always gives [`None`].
    pub fn new() -> WeakRemote<T> {
        // we use usize::MAX to mark the pointer as dangling.
        let dangling = std::usize::MAX;
        let ptr = &dangling as *const usize as (*const *mut Inner<T>);
        let dangling_ptr = unsafe { *ptr };

        WeakRemote {
            ptr: NonNull::new(dangling_ptr).expect("MAX is not 0"),
        }
    }

    /// Returns `None` when the pointer is dangling and there is no allocated `Inner`,
    /// (i.e., when this `WeakRemote` was created by `WeakRemote::new`).
    #[inline]
    fn inner(&self) -> Option<&Inner<T>> {
        if is_dangling(self.ptr) {
            None
        } else {
            Some(unsafe { self.ptr.as_ref() })
        }
    }

    /// Attempts to upgrade the `WeakRemote` pointer to a [`Remote`], extending
    /// the lifetime of the value if successful.
    ///
    /// Returns [`None`] if the value has since been dropped.
    pub fn upgrade(&self) -> Option<Remote<T>> {
        // We use a CAS loop to increment the strong count instead of a
        // fetch_add because once the count hits 0 it must never be above 0.
        let inner = self.inner()?;

        // Relaxed load because any write of 0 that we can observe
        // leaves the field in a permanently zero state (so a
        // "stale" read of 0 is fine), and any other value is
        // confirmed via the CAS below.
        let mut n = inner.strong.load(Relaxed);

        loop {
            if n == 0 {
                return None;
            }

            // See comments in `Remote::clone` for why we do this (for `mem::forget`).
            if n > MAX_REFCOUNT {
                unsafe {
                    std::process::abort();
                }
            }

            // Relaxed is valid for the same reason it is on Arc's Clone impl
            match inner.strong.compare_exchange_weak(n, n + 1, Relaxed, Relaxed) {
                Ok(_) => return Some(Remote {
                    // null checked above
                    ptr: self.ptr
                }),
                Err(old) => n = old,
            }
        }
    }
}

impl<T: ?Sized> Clone for WeakRemote<T> {
    /// Makes a clone of the `WeakRemote` pointer that points to the same value.
    #[inline]
    fn clone(&self) -> WeakRemote<T> {
        let inner = if let Some(inner) = self.inner() {
            inner
        } else {
            return WeakRemote { ptr: self.ptr };
        };
        // See comments in Arc::clone() for why this is relaxed.  This can use a
        // fetch_add (ignoring the lock) because the weak count is only locked
        // where are *no other* weak pointers in existence. (So we can't be
        // running this code in that case).
        let old_size = inner.weak.fetch_add(1, Relaxed);

        // See comments in Remote::clone() for why we do this (for mem::forget).
        if old_size > MAX_REFCOUNT {
            unsafe {
                std::process::abort();
            }
        }

        return WeakRemote { ptr: self.ptr };
    }
}

impl<T> Default for WeakRemote<T> {
    /// Constructs a new `WeakRemote<T>`, without allocating memory.
    /// Calling [`upgrade`] on the return value always
    /// gives [`None`].
    fn default() -> WeakRemote<T> {
        WeakRemote::new()
    }
}

impl<T: ?Sized> Drop for WeakRemote<T> {
    /// Drops the `WeakRemote` pointer.
    fn drop(&mut self) {
        // If we find out that we were the last weak pointer, then its time to
        // deallocate the data entirely. See the discussion in Arc::drop() about
        // the memory orderings
        //
        // It's not necessary to check for the locked state here, because the
        // weak count can only be locked if there was precisely one weak ref,
        // meaning that drop could only subsequently run ON that remaining weak
        // ref, which can only happen after the lock is released.
        let inner = if let Some(inner) = self.inner() {
            inner
        } else {
            return
        };

        if inner.weak.fetch_sub(1, Release) == 1 {
            atomic::fence(Acquire);
            unsafe {
                Global.dealloc(self.ptr.cast(), Layout::for_value(self.ptr.as_ref()))
            }
        }
    }
}

pub(crate) struct Inner<T: ?Sized> {
    strong: atomic::AtomicUsize,

    // the value usize::MAX acts as a sentinel for temporarily "locking" the
    // ability to upgrade weak pointers or downgrade strong ones; this is used
    // to avoid races in `make_mut` and `get_mut`.
    weak: atomic::AtomicUsize,

    // The actor's thread.
    thread: Weak<Thread>,

    pub(crate) actor: UnsafeCell<Actor<T>>,
}

/// A `Receiver` is an abstract pointer that can be turned into an actual `Remote` pointer only
/// by the pointed data itself.
/// It is used by signals receivers to build `Remote` pointers of themselves.
#[derive(Clone, Copy)]
pub struct Receiver {
    owner: *const (),
    inner: *mut ()
}

unsafe impl Sync for Receiver {}
unsafe impl Send for Receiver {}

impl Receiver {
    /// Try to upgrade the receiver with the given data.
    /// Is the `Receiver` is indeed a pointer to the given data,
    /// then it returns a `Remote` pointer to the data.
    /// Note that this function is UNSAFE since we must ensure that the data is indeed wrapped
    /// in a LIVING `Remote` pointer.
    /// It is the case when, for instance, we use it to retrive the target of a signal when it
    /// is processed, since the thread query (`Delivery`) contains a `Remote` pointer to the
    /// target.
    pub(crate) unsafe fn upgrade<T>(&self, data: &T) -> Option<Remote<T>> {
        if data as *const T as *const () == self.owner {
            let inner = &mut *(self.inner as *mut Inner<T>);
            let old_size = inner.strong.fetch_add(1, Relaxed);
            if old_size > MAX_REFCOUNT {
                unsafe {
                    std::process::abort();
                }
            }

            Some(Remote {
                ptr: NonNull::new(inner as *mut Inner<T>).unwrap()
            })
        } else {
            None
        }
    }
}