/**  \file           atomic_shared.h
 *
 *  \author         George T. Talbot
 *  \author         Peter Dimov
 *
 *  \brief          This file defines a template class that can wrap
 *                  boost::shared_ptr and provide thread-safe atomic
 *                  read/write operations.
 *
 *  (C) Copyright George Talbot, Peter Dimov 2006.
 *
 *  Distributed under the Boost Software License, Version 1.0. (See
 *  accompanying file LICENSE_1_0.txt or copy at
 *  http://www.boost.org/LICENSE_1_0.txt)
 */
#ifndef ATOMIC_SHARED_H
#define ATOMIC_SHARED_H

#include <boost/shared_ptr.hpp>
#include <pthread.h>
#include <stdexcept>
#include <cerrno>

#ifdef ATOMIC_ABORT_ON_ERROR
#include <cassert>
#define ATOMIC_SHARED_ABORT abort();
#else
#define ATOMIC_SHARED_ABORT
#endif

/** A wrapper for shared_ptr that implements atomic assignments and compare_and_swap. */
template<class T>
class atomic_shared
{
    typedef boost::shared_ptr<T>    base_ptr;

    /** Lock a spinlock protecting an atomic pointer.
     *
     *  @param  l               Spinlock to lock.
     *  @param  should_throw    If true, thrown an exception if lock fails.
     */
    static void lock(pthread_spinlock_t& l, bool should_throw = true)
    {
        int rv = ::pthread_spin_lock(&l);
        
        if (rv)
        {
            errno   = rv;

            ATOMIC_SHARED_ABORT

            if (should_throw)
            {
                throw std::logic_error("can't lock spinlock");
            }
        }
    }

    /** Unlock a spinlock protecting an atomic pointer. */
    static void unlock(pthread_spinlock_t& l)
    {
        int rv = ::pthread_spin_unlock(&l);
        
        if (rv)
        {
            ATOMIC_SHARED_ABORT
            errno   = rv;
            throw std::logic_error("can't lock spinlock");
        }
    }

    /** Initialize a spinlock protecting an atomic pointer. */
    static void init(pthread_spinlock_t& l)
    {
        int rv = ::pthread_spin_init(&l, PTHREAD_PROCESS_PRIVATE);
        
        if (rv)
        {
            ATOMIC_SHARED_ABORT
            errno   = rv;
            throw std::logic_error("can't initialize spinlock");
        }
    }

    /** Lock the current spinlock and another spinlock.  This is used during
    *   operations that read some pointer and write this.
     */
    template<class Y>
    void lock_both(const atomic_shared<Y>& other)
    {
        // TODO:  Is this the right way to do this in deadlock-free fashion?
        if ((void*)&other < (void*)this)
        {
            lock(other.spinlock);
            lock(spinlock);
        }
        else
        {
            lock(spinlock);
            lock(other.spinlock);
        }
    }

    /** Unlock the current spinlock and another spinlock.  This is used during
     *  operations that read some pointer and write this.
     */
    template<class Y>
    void unlock_both(const atomic_shared<Y>& other)
    {
        if ((void*)&other < (void*)this)
        {
            unlock(spinlock);
            unlock(other.spinlock);
        }
        else
        {
            unlock(other.spinlock);
            unlock(spinlock);
        }
    }

public:
    typedef typename base_ptr::element_type  element_type;   ///< Type of element to which pointer points.
    typedef typename base_ptr::value_type    value_type;     ///< Same as element_type.
    typedef typename base_ptr::pointer       pointer;        ///< Pointer to refcounted object.
    typedef typename base_ptr::reference     reference;      ///< Reference to refcounted object.

    /** Initialize an empty spinlock. */
    atomic_shared()
    {
        init(spinlock);
    }

    /** Create a managed pointer from a bare one. */
    template<class Y>
    explicit atomic_shared(Y* p)
        : ptr(p)
    {
        init(spinlock);
    }

    /** Create a managed pointer from a bare one, and a custom deleter. */
    template<class Y, class D>
    atomic_shared(Y * p, D d)
        : ptr(p, d)
    {
        init(spinlock);
    }

    atomic_shared(const atomic_shared<T>& p)
    {
        init(spinlock);
        lock(p.spinlock);
        ptr = p.ptr;
        unlock(p.spinlock);
    }

    atomic_shared<T>& operator=(const atomic_shared<T>& p)
    {
        if (&p != this)
        {
            lock_both(p);
            ptr = p.ptr;
            unlock_both(p);
        }
        return *this;
    }

    template<class Y>
    atomic_shared(const atomic_shared<Y>& p)
    {
        init(spinlock);
        lock(p.spinlock);
        ptr = p.ptr;
        unlock(p.spinlock);
    }

    template<class Y>
    atomic_shared<T>& operator=(const atomic_shared<Y>& p)
    {
        lock_both(p);
        ptr = p.ptr;
        unlock_both(p);

        return *this;
    }

    ~atomic_shared()
    {
        // Lock the spinlock to hold off all other copiers.
        lock(spinlock, false);
        
        // Destroy the spinlock--this should cause all other copiers to
        // either assert or throw.
        int rv = ::pthread_spin_destroy(&spinlock);
        
        // For debugging, drop us into the debugger if we can't destroy the
        // spinlock.
        if (rv)
        {
            ATOMIC_SHARED_ABORT
        }
    }

    /** Reset the pointer to an empty pointer. */
    void    reset()
    {
        lock(spinlock);
        ptr.reset();
        unlock(spinlock);
    }

    /** Reset the pointer and replace it with a new bare pointer. */
    template<class Y>
    void    reset(Y* p)
    {
        lock(spinlock);
        ptr.reset(p);
        unlock(spinlock);
    }
    
    /** Reset the pointer and replace it with a new bare pointer with custom deleter. */
    template<class Y, class D>
    void    reset(Y* p, D d)
    {
        lock(spinlock);
        ptr.reset(p, d);
        unlock(spinlock);
    }

    reference   operator*() const
    {
        return ptr.operator*();
    }

    T*          operator->() const
    {
        return ptr.operator->();
    }

    T*          get() const
    {
        return ptr.get();
    }

    operator bool() const
    {
        return ptr;
    }

    bool    operator!() const
    {
        return !ptr;
    }

    bool    unique() const
    {
        return ptr.unique();
    }

    long    use_count() const
    {
        return ptr.use_count();
    }

    /** Atomically swap the pointer with another pointer. */
    void    swap(const atomic_shared<T>& other)
    {
        lock_both(other);
        ptr.swap(other.ptr);
        unlock_both(other);
    }

    /** Attempt to set the value of the pointer with another pointer, but only if the
     *  pointer is the same as a previously sampled value of the pointer.
     *
     *  @return true if the swap was successful.
     */
    bool    compare_and_set(const atomic_shared<T>& cmp, atomic_shared<T> xchg)    // Note that xchg is a local copy.
    {
        int rv = ::pthread_spin_trylock(&spinlock);

        switch (rv)
        {
        case EBUSY:
            if ((void*)&xchg < (void*)this)   unlock(xchg.spinlock);
            return false;

        case 0:
            break;

        default:
            ATOMIC_SHARED_ABORT
            errno   = rv;
            throw std::logic_error("can't pthread_spin_trylock");
        }

        bool    r = ptr == cmp.ptr;
        
        if (r) ptr.swap(xchg.ptr);

        unlock(spinlock);

        return r;
    }

private:
    template<class Y> friend class  atomic_shared;

    base_ptr                    ptr;
    mutable pthread_spinlock_t  spinlock;

};  // atomic_shared
        
#endif