threadsafe_queue.hpp

Go to the documentation of this file.

#ifndef GENESIS_UTILS_THREADING_THREADSAFE_QUEUE_H_
#define GENESIS_UTILS_THREADING_THREADSAFE_QUEUE_H_
 
/*
    Genesis - A toolkit for working with phylogenetic data.
    Copyright (C) 2014-2024 Lucas Czech
 
    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/>.
 
    Contact:
    Lucas Czech <lczech@carnegiescience.edu>
    Department of Plant Biology, Carnegie Institution For Science
    260 Panama Street, Stanford, CA 94305, USA
*/
 
/*
    The code below is adapted from https://github.com/anthonywilliams/ccia_code_samples
    and was published under the Boost Software License:
 
    Boost Software License - Version 1.0 - August 17th, 2003
 
    Permission is hereby granted, free of charge, to any person or organization
    obtaining a copy of the software and accompanying documentation covered by
    this license (the "Software") to use, reproduce, display, distribute,
    execute, and transmit the Software, and to prepare derivative works of the
    Software, and to permit third-parties to whom the Software is furnished to
    do so, all subject to the following:
 
    The copyright notices in the Software and this entire statement, including
    the above license grant, this restriction and the following disclaimer,
    must be included in all copies of the Software, in whole or in part, and
    all derivative works of the Software, unless such copies or derivative
    works are solely in the form of machine-executable object code generated by
    a source language processor.
 
    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
    FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
    SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
    FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
    ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
    DEALINGS IN THE SOFTWARE.
 */
 
#include <atomic>
#include <condition_variable>
#include <future>
#include <memory>
#include <mutex>
#include <queue>
 
namespace genesis {
namespace utils {
 
// =================================================================================================
//     Threadsafe Queue
// =================================================================================================
 
template <typename T>
class ThreadsafeQueue
{
public:
 
    // -------------------------------------------------------------------------
    //     Constructor and Rule of Five
    // -------------------------------------------------------------------------
 
    ThreadsafeQueue() = default;
    ~ThreadsafeQueue() = default;
 
    ThreadsafeQueue( ThreadsafeQueue const& other )
    {
        std::lock_guard<std::mutex> lk( other.mutex_ );
        data_queue_ = other.data_queue_;
    }
 
    ThreadsafeQueue(ThreadsafeQueue&& other) noexcept
    {
        std::lock_guard<std::mutex> lk(other.mutex_);
        data_queue_ = std::move( other.data_queue_ );
    }
 
    ThreadsafeQueue& operator=( ThreadsafeQueue const& other )
    {
        if( &other == this ) {
            return *this;
        }
 
        // Lock both mutexes without deadlock, C++11.
        // In C++17, the next three lines could simply be this one instead:
        // std::scoped_lock lk( mutex_, other.mutex_ );
        std::lock(mutex_, other.mutex_);
        std::lock_guard<std::mutex> lhs_lk(mutex_, std::adopt_lock);
        std::lock_guard<std::mutex> rhs_lk(other.mutex_, std::adopt_lock);
 
        data_queue_ = other.data_queue_;
        return *this;
    }
 
    ThreadsafeQueue& operator=( ThreadsafeQueue&& other ) noexcept
    {
        if( &other == this ) {
            return *this;
        }
 
        // See above, copy assignment. Same here.
        std::lock(mutex_, other.mutex_);
        std::lock_guard<std::mutex> lhs_lk(mutex_, std::adopt_lock);
        std::lock_guard<std::mutex> rhs_lk(other.mutex_, std::adopt_lock);
        data_queue_ = std::move( other.data_queue_ );
        return *this;
    }
 
    // -------------------------------------------------------------------------
    //     Member Functions
    // -------------------------------------------------------------------------
 
    void push( T new_value )
    {
        std::lock_guard<std::mutex> lk(mutex_);
        data_queue_.push( std::move( new_value ));
        data_cond_.notify_one();
    }
 
    void wait_and_pop( T& value )
    {
        std::unique_lock<std::mutex> lk(mutex_);
        data_cond_.wait(lk, [this] { return !data_queue_.empty(); });
        value = std::move( data_queue_.front() );
        data_queue_.pop();
    }
 
    std::shared_ptr<T> wait_and_pop()
    {
        std::unique_lock<std::mutex> lk(mutex_);
        data_cond_.wait(lk, [this] { return !data_queue_.empty(); });
        std::shared_ptr<T> res(std::make_shared<T>(data_queue_.front()));
        data_queue_.pop();
        return res;
    }
 
    bool try_pop( T& value )
    {
        std::lock_guard<std::mutex> lk(mutex_);
        if( data_queue_.empty() ) {
            return false;
        }
        value = std::move( data_queue_.front() );
        data_queue_.pop();
        return true;
    }
 
    std::shared_ptr<T> try_pop()
    {
        std::lock_guard<std::mutex> lk(mutex_);
        if( data_queue_.empty() ) {
            return std::shared_ptr<T>();
        }
        std::shared_ptr<T> res(std::make_shared<T>(data_queue_.front()));
        data_queue_.pop();
        return res;
    }
 
    bool empty() const
    {
        std::lock_guard<std::mutex> lk(mutex_);
        return data_queue_.empty();
    }
 
private:
 
    mutable std::mutex mutex_;
    std::queue<T> data_queue_;
    std::condition_variable data_cond_;
};
 
// =================================================================================================
//     Threadsafe Queue
// =================================================================================================
 
// The below code is another threadsafe queue implementation from the same book,
// but it seems broken, see https://github.com/anthonywilliams/ccia_code_samples/issues/53
// We hence do not use it here, but keep it here for later reference, should that issue get resolved.
 
// /**
//  * @brief Thread-safe queue with fine-grained locking.
//  *
//  * This implementation is from the excellent "C++ Concurrency in Action" book by Anthony Williams,
//  * second edition, chapter 6, listings 6.7 to 6.10.
//  * The code is published at https://github.com/anthonywilliams/ccia_code_samples under
//  * the [Boost Software License](https://www.boost.org/LICENSE_1_0.txt).
//  */
// template<typename T>
// class ThreadsafeQueue
// {
// private:
//
//     // -------------------------------------------------------------------------
//     //     Member Types
//     // -------------------------------------------------------------------------
//
//     struct Node
//     {
//         std::shared_ptr<T> data;
//         std::unique_ptr<Node> next;
//     };
//
//     // -------------------------------------------------------------------------
//     //     Constructor and Rule of Five
//     // -------------------------------------------------------------------------
//
// public:
//
//     ThreadsafeQueue()
//         : head_( new Node )
//         , tail_( head_.get() )
//     {}
//
//     ~ThreadsafeQueue() = default;
//
//     ThreadsafeQueue( ThreadsafeQueue const& ) = delete;
//     ThreadsafeQueue( ThreadsafeQueue&& ) = default;
//
//     ThreadsafeQueue& operator=( ThreadsafeQueue const& ) = delete;
//     ThreadsafeQueue& operator=( ThreadsafeQueue&& ) = default;
//
//     // -------------------------------------------------------------------------
//     //     Member Functions
//     // -------------------------------------------------------------------------
//
//     bool empty()
//     {
//         std::lock_guard<std::mutex> head_lock( head_mutex_ );
//         return ( head_.get() == get_tail_() );
//     }
//
//     void push( T new_value )
//     {
//         // Create the new elements
//         std::shared_ptr<T> new_data(
//             std::make_shared<T>( std::move( new_value ))
//         );
//         std::unique_ptr<Node> new_node( new Node );
//
//         // Lock scope to do the push
//         {
//             std::lock_guard<std::mutex> tail_lock( tail_mutex_ );
//             tail_->data = new_data;
//             Node* const new_tail = new_node.get();
//             tail_->next = std::move( new_node );
//             tail_ = new_tail;
//         }
//
//         // Notify any waiting pop
//         data_cond_.notify_one();
//     }
//
//     std::shared_ptr<T> try_pop()
//     {
//         std::unique_ptr<Node> const old_head = try_pop_head_();
//         return old_head ? old_head->data : std::shared_ptr<T>();
//     }
//
//     bool try_pop( T& value )
//     {
//         std::unique_ptr<Node> const old_head = try_pop_head_( value );
//         return old_head;
//     }
//
//     std::shared_ptr<T> wait_and_pop()
//     {
//         std::unique_ptr<Node> const old_head = wait_pop_head_();
//         return old_head->data;
//     }
//
//     void wait_and_pop( T& value )
//     {
//         std::unique_ptr<Node> const old_head = wait_pop_head_( value );
//     }
//
// private:
//
//     // -------------------------------------------------------------------------
//     //     Members Functions
//     // -------------------------------------------------------------------------
//
//
//     Node* get_tail_()
//     {
//         std::lock_guard<std::mutex> tail_lock( tail_mutex_ );
//         return tail_;
//     }
//
//     std::unique_ptr<Node> pop_head_()
//     {
//         std::unique_ptr<Node> old_head = std::move( head_ );
//         head_ = std::move( old_head->next );
//         return old_head;
//     }
//
//     std::unique_ptr<Node> try_pop_head_()
//     {
//         std::lock_guard<std::mutex> head_lock( head_mutex_ );
//         if( head_.get() == get_tail_() ) {
//             return std::unique_ptr<Node>();
//         }
//         return pop_head_();
//     }
//
//     std::unique_ptr<Node> try_pop_head_( T& value )
//     {
//         std::lock_guard<std::mutex> head_lock( head_mutex_ );
//         if( head_.get() == get_tail_() ) {
//             return std::unique_ptr<Node>();
//         }
//         value = std::move( *head_->data );
//         return pop_head_();
//     }
//
//     std::unique_lock<std::mutex> wait_for_data_()
//     {
//         std::unique_lock<std::mutex> head_lock( head_mutex_ );
//         // The following wait can block indefinitely...
//         // See https://github.com/anthonywilliams/ccia_code_samples/issues/53
//         data_cond_.wait(
//             head_lock,
//             [&]{
//                 return head_.get() != get_tail_();
//             }
//         );
//
//         // This is a potential workaround, but of course makes the queue slower.
//         // while( head.get() == get_tail() ) {
//         //     data_cond.wait_for(
//         //         head_lock,
//         //         std::chrono::milliseconds(1)
//         //     );
//         // }
//
//         // Bug in the listing and book,
//         // see https://github.com/anthonywilliams/ccia_code_samples/issues/52
//         // return std::move( head_lock );
//         return head_lock;
//     }
//
//     std::unique_ptr<Node> wait_pop_head_()
//     {
//         std::unique_lock<std::mutex> head_lock( wait_for_data_() );
//         return pop_head_();
//     }
//
//     std::unique_ptr<Node> wait_pop_head_(T& value)
//     {
//         std::unique_lock<std::mutex> head_lock( wait_for_data_() );
//         value = std::move( *head_->data );
//         return pop_head_();
//     }
//
//     // -------------------------------------------------------------------------
//     //     Member Variables
//     // -------------------------------------------------------------------------
//
//     // Access mutexes
//     std::mutex head_mutex_;
//     std::mutex tail_mutex_;
//
//     // Queue, with ownership controlled by the head
//     std::unique_ptr<Node> head_;
//     Node* tail_;
//
//     // Condition for wait_and_pop()
//     std::condition_variable data_cond_;
// };
 
} // namespace utils
} // namespace genesis
 
#endif // include guard