ts_queue.hpp

/*
    Thread safe wrapper for queue-like containers
    Copyright (C) 2013  Michał Walenciak <Kicer86@gmail.com>

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/


#ifndef OPENLIBRARY_PALGORITHM_TS_QUEUE
#define OPENLIBRARY_PALGORITHM_TS_QUEUE

#include <cassert>

#include <atomic>
#include <condition_variable>
#include <deque>
#include <mutex>

#include "utils/optional.hpp"

namespace ol
{

    template<typename T>
    class TS_Queue
    {
        public:
            TS_Queue(size_t max_size):
                m_queue(),
                m_is_not_full(),
                m_is_not_empty(),
                m_queue_mutex(),
                m_max_size(max_size),
                m_stopped(false)
            {

            }


            virtual ~TS_Queue()
            {
                stop();

                std::unique_lock<std::mutex> lock(m_queue_mutex);
                assert(m_queue.empty());
            }


            void push(const T &item)
            {
                assert(m_stopped == false);

                if (m_stopped == false)
                {
                    std::unique_lock<std::mutex> lock(m_queue_mutex);

                    m_is_not_full.wait(lock, [&] { return m_queue.size() < m_max_size; } );  //wait for conditional_variable if there is no place in queue
                    m_queue.push_back(item);
                    m_is_not_empty.notify_one();
                }
            }
            
            [[deprecated]]
            void push_front(const T &item)
            {
                push(item);
            }


            void push(T&& item)
            {
                assert(m_stopped == false);

                if (m_stopped == false)
                {
                    std::unique_lock<std::mutex> lock(m_queue_mutex);

                    m_is_not_full.wait(lock, [&] { return m_queue.size() < m_max_size; } );  //wait for conditional_variable if there is no place in queue
                    m_queue.push_back(std::move(item));
                    m_is_not_empty.notify_one();
                }
            }
            
            [[deprecated]]
            void push_back(T&& item)
            {
                push(std::move(item));
            }


            Optional<T> pop()
            {
                std::unique_lock<std::mutex> lock(m_queue_mutex);
                Optional<T> result;

                wait_for_data(lock);

                if (m_queue.empty() == false)
                {
                    result = std::move( *(m_queue.begin()) );
                    m_queue.pop_front();
                    m_is_not_full.notify_one();
                }

                return std::move(result);
            }
            
            [[deprecated]]
            Optional<T> pop_front()
            {
                return pop();
            }
            

            Optional<T> pop_for(const std::chrono::milliseconds& timeout)
            {
                std::unique_lock<std::mutex> lock(m_queue_mutex);
                Optional<T> result;

                const bool status = wait_for_data(lock, timeout);

                if (status && m_queue.empty() == false)
                {
                    result = std::move( *(m_queue.begin()) );
                    m_queue.pop_front();
                    m_is_not_full.notify_one();
                }

                return std::move(result);
            }

            size_t size() const
            {
                std::unique_lock<std::mutex> lock(m_queue_mutex);
                const size_t result = m_queue.size();
                return result;
            }

            bool empty() const
            {
                std::unique_lock<std::mutex> lock(m_queue_mutex);
                const bool result = m_queue.empty();
                return result;
            }


            void stop()
            {
                m_stopped = true;
                m_is_not_empty.notify_all();
            }

            void wait_for_data()
            {
                std::unique_lock<std::mutex> lock(m_queue_mutex);
                Optional<T> result;

                wait_for_data(lock);
            }

        private:
            std::deque<T> m_queue;
            std::condition_variable m_is_not_full;
            std::condition_variable m_is_not_empty;
            mutable std::mutex m_queue_mutex;
            size_t m_max_size;
            std::atomic<bool> m_stopped;

            void wait_for_data(std::unique_lock<std::mutex>& lock)
            {
                auto precond = [&]
                {
                    const bool condition = m_stopped == false && m_queue.empty();
                    return !condition;
                };

                m_is_not_empty.wait(lock, precond);
            }

            bool wait_for_data(std::unique_lock<std::mutex>& lock, const std::chrono::milliseconds& timeout)
            {
                auto precond = [&]
                {
                    const bool condition = m_stopped == false && m_queue.empty();
                    return !condition;
                };

                const bool status = m_is_not_empty.wait_for(lock, timeout, precond);
                
                return status;
            }
    };

}
#endif