libcore/include/sirikata/core/queue/ThreadSafeQueue.hpp

Go to the documentation of this file.

/*  Sirikata Utilities -- Sirikata Synchronization Utilities
 *  ThreadSafeQueue.hpp
 *
 *  Copyright (c) 2008, Patrick Reiter Horn
 *  All rights reserved.
 *
 *  Redistribution and use in source and binary forms, with or without
 *  modification, are permitted provided that the following conditions are
 *  met:
 *  * Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *  * Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 *  * Neither the name of Sirikata nor the names of its contributors may
 *    be used to endorse or promote products derived from this software
 *    without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
 * IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
 * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER
 * OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

#ifndef SIRIKATA_ThreadSafeQueue_HPP__
#define SIRIKATA_ThreadSafeQueue_HPP__


namespace Sirikata {
namespace ThreadSafeQueueNS{
class Lock;
class Condition;
SIRIKATA_EXPORT void lock(Lock*lok);
SIRIKATA_EXPORT void wait(Lock*lok,
          Condition* cond,
          bool(*check)(void*, void *),
          void* arg1, void* arg2);
SIRIKATA_EXPORT void notify(Condition* cond);
SIRIKATA_EXPORT void unlock(Lock*lok);
SIRIKATA_EXPORT Lock* lockCreate();
SIRIKATA_EXPORT Condition* condCreate();
SIRIKATA_EXPORT void lockDestroy(Lock*oldlock);
SIRIKATA_EXPORT void condDestroy(Condition*oldcond);
}

template <typename T>
class ThreadSafeQueue {
  protected:
    typedef std::deque<T> ListType;
    ThreadSafeQueueNS::Lock* mLock;
    ListType mList;
    ThreadSafeQueueNS::Condition* mCond;
  private:
    ThreadSafeQueue& operator= (const ThreadSafeQueue &other){
        if (this<&other) {
            ThreadSafeQueueNS::lock(mLock);
            ThreadSafeQueueNS::lock(other.mLock);
        }else {
            ThreadSafeQueueNS::lock(other.mLock);
            ThreadSafeQueueNS::lock(mLock);
        }
        try {
            mList=other.mList;
        }catch (...) {
            if (this<&other) {
                ThreadSafeQueueNS::unlock(other.mLock);
                ThreadSafeQueueNS::unlock(mLock);
            }else {
                ThreadSafeQueueNS::unlock(mLock);
                ThreadSafeQueueNS::unlock(other.mLock);
            }
            throw;
        }
        if (this<&other) {
            ThreadSafeQueueNS::unlock(other.mLock);
            ThreadSafeQueueNS::unlock(mLock);
        }else {
            ThreadSafeQueueNS::unlock(mLock);
            ThreadSafeQueueNS::unlock(other.mLock);
        }

        return *this;
    }
    ThreadSafeQueue(const ThreadSafeQueue &other){
        mLock=ThreadSafeQueueNS::lockCreate();
        mCond=ThreadSafeQueueNS::condCreate();
        ThreadSafeQueueNS::lock(other.mLock);
        try {
            mList=other.mList;
            ThreadSafeQueueNS::unlock(other.mLock);
        }catch (...) {
            ThreadSafeQueueNS::unlock(other.mLock);
            throw;
        }
    }

    static bool waitCheck(void * thus, void*vretval) {
        T* retval=reinterpret_cast<T*>(vretval);

        if (reinterpret_cast <ThreadSafeQueue* >(thus)->mList.empty())
            return true;
        *retval=reinterpret_cast <ThreadSafeQueue* >(thus)->mList.front();
        reinterpret_cast <ThreadSafeQueue* >(thus)->mList.pop_front();
        return false;
    }

public:
    class NodeIterator {
    private:
        // Noncopyable
        NodeIterator(const NodeIterator &other);
        void operator=(const NodeIterator &other);

        T *mNext;
        ListType mSwappedList;

    public:
        NodeIterator(ThreadSafeQueue<T> &queue) : mNext(NULL) {
            queue.swap(mSwappedList);
        }

        T *next() {
            if (mNext) {
                mSwappedList.pop_front();
            }
            if (mSwappedList.empty()) {
                return NULL;
            }
            mNext = &(mSwappedList.front());
            return mNext;
        }
    };
    friend class NodeIterator;

    ThreadSafeQueue() {
        mLock=ThreadSafeQueueNS::lockCreate();
        mCond=ThreadSafeQueueNS::condCreate();
    }
    ~ThreadSafeQueue(){
        ThreadSafeQueueNS::lockDestroy(mLock);
        ThreadSafeQueueNS::condDestroy(mCond);
    }

    void swap(std::deque<T> &swapWith) {
        ThreadSafeQueueNS::lock(mLock);
        mList.swap(swapWith);
        ThreadSafeQueueNS::unlock(mLock);
    }

    void popAll(std::deque<T> *popResults) {
        popResults->resize(0);
        swap(*popResults);
    }

    int32 push(const T &value) {
        int32 new_size;
        ThreadSafeQueueNS::lock(mLock);
        try {
            mList.push_back(value);
            new_size = mList.size();
            ThreadSafeQueueNS::notify(mCond);
        } catch (...) {
            ThreadSafeQueueNS::unlock(mLock);
            throw;
        }
        ThreadSafeQueueNS::unlock(mLock);
        return new_size;
    }

    int32 pushMultiple(const std::deque<T> &values) {
        int32 new_size;
        ThreadSafeQueueNS::lock(mLock);
        try {
            while(!values.empty()) {
                T& value = values.front();
                mList.push_back(value);
                values.pop();
            }
            new_size = mList.size();
            ThreadSafeQueueNS::notify(mCond);
        } catch (...) {
            ThreadSafeQueueNS::unlock(mLock);
            throw;
        }
        ThreadSafeQueueNS::unlock(mLock);
        return new_size;
    }

    bool pop(T& ret) {
        ThreadSafeQueueNS::lock(mLock);
        if (mList.empty()) {
            ThreadSafeQueueNS::unlock(mLock);
            return false;
        } else {
            try {
                ret = mList.front();
                mList.pop_front();
            }catch (...) {
                ThreadSafeQueueNS::unlock(mLock);
                throw;
            }
            ThreadSafeQueueNS::unlock(mLock);
            return true;
        }
    }

    void blockingPop(T& retval) {
        ThreadSafeQueueNS::wait(mLock, mCond, &ThreadSafeQueue<T>::waitCheck, this, &retval);
    }

    bool probablyEmpty() {
        return mList.empty();
    }

    int32 size() {
        return mList.size();
    }
};

}

#endif