suspendable_queue.hpp

Go to the documentation of this file.

// SPDX-FileCopyrightText: 2006-2024 Knut Reinert & Freie Universität Berlin
// SPDX-FileCopyrightText: 2016-2024 Knut Reinert & MPI für molekulare Genetik
// SPDX-License-Identifier: BSD-3-Clause
 
#pragma once
 
#include <algorithm>
#include <cassert>
#include <condition_variable>
#include <iterator>
#include <mutex>
#include <ranges>
#include <span>
#include <thread>
#include <vector>
 
#include <seqan3/core/platform.hpp>
 
namespace seqan3::contrib
{
 
// ============================================================================
// Forwards
// ============================================================================
 
// ============================================================================
// Classes
// ============================================================================
 
// ----------------------------------------------------------------------------
// Class ConcurrentQueue
// ----------------------------------------------------------------------------
 
template <typename TSpec = void>
struct Tag
{};
 
template <typename TSpec = Tag<void>>
struct Suspendable;
 
template <typename TValue, typename TSpec = Suspendable<>>
class ConcurrentQueue;
 
struct Limit_;
using Limit = Tag<Limit_>;
 
template <typename TValue, typename TSpec>
class ConcurrentQueue<TValue, Suspendable<TSpec>>
{
public:
    typedef std::vector<TValue> TString;
    typedef typename TString::size_type TSize;
 
    size_t readerCount;
    size_t writerCount;
 
    TString data;
    TSize occupied;
    TSize back;
    TSize front;
 
    std::mutex cs;
    std::condition_variable more;
 
    bool virgin;
 
    ConcurrentQueue() : readerCount(0), writerCount(0), occupied(0), back(0), front(0), virgin(true)
    {}
 
    ~ConcurrentQueue()
    {
        assert(writerCount == 0u);
 
        // wait for all pending readers to finish
        while (readerCount != 0u)
        {}
    }
};
 
template <typename TValue>
class ConcurrentQueue<TValue, Suspendable<Limit>> : public ConcurrentQueue<TValue, Suspendable<>>
{
public:
    typedef ConcurrentQueue<TValue, Suspendable<>> TBase;
    typedef typename TBase::TString TString;
    typedef typename TBase::TSize TSize;
 
    std::condition_variable less;
 
    ConcurrentQueue(TSize maxSize) : TBase()
    {
        this->data.resize(maxSize);
        // reserve(this->data, maxSize, Exact());
        // _setLength(this->data, maxSize);
    }
 
    ConcurrentQueue(ConcurrentQueue const & other) : TBase((TBase const &)other)
    {}
};
 
template <typename TValue, typename TSpec>
inline void lockReading(ConcurrentQueue<TValue, Suspendable<TSpec>> &)
{}
 
template <typename TValue, typename TSpec>
inline void unlockReading(ConcurrentQueue<TValue, Suspendable<TSpec>> & me)
{
    {
        std::lock_guard<std::mutex> lock(me.cs);
        if (--me.readerCount != 0u)
            return;
    }
    me.less.notify_all(); // publish the condition that reader count is 0.
}
 
template <typename TValue, typename TSpec>
inline void lockWriting(ConcurrentQueue<TValue, Suspendable<TSpec>> &)
{}
 
template <typename TValue, typename TSpec>
inline void unlockWriting(ConcurrentQueue<TValue, Suspendable<TSpec>> & me)
{
    {
        std::lock_guard<std::mutex> lk(me.cs);
        if (--me.writerCount != 0u)
            return;
    }
    me.more.notify_all(); // publish the condition, that writer count is 0.
}
 
template <typename TValue, typename TSize, typename TSpec>
inline void setReaderCount(ConcurrentQueue<TValue, Suspendable<TSpec>> & me, TSize readerCount)
{
    std::unique_lock<std::mutex> lock(me.cs);
    me.readerCount = readerCount;
}
 
template <typename TValue, typename TSize, typename TSpec>
inline void setWriterCount(ConcurrentQueue<TValue, Suspendable<TSpec>> & me, TSize writerCount)
{
    std::unique_lock<std::mutex> lock(me.cs);
    me.writerCount = writerCount;
}
 
template <typename TValue, typename TSize1, typename TSize2, typename TSpec>
inline void
setReaderWriterCount(ConcurrentQueue<TValue, Suspendable<TSpec>> & me, TSize1 readerCount, TSize2 writerCount)
{
    std::unique_lock<std::mutex> lock(me.cs);
    me.readerCount = readerCount;
    me.writerCount = writerCount;
}
 
template <typename TValue, typename TSize, typename TSpec>
inline bool waitForMinSize(ConcurrentQueue<TValue, Suspendable<TSpec>> & me, TSize minSize)
{
    std::unique_lock<std::mutex> lock(me.cs);
    while (me.occupied < minSize && me.writerCount > 0u)
        me.more.wait(lock);
    return me.occupied >= minSize;
}
 
template <typename TValue, typename TSpec>
inline bool empty(ConcurrentQueue<TValue, Suspendable<TSpec>> const & me)
{
    return me.occupied == 0;
}
 
template <typename TValue, typename TSpec>
inline typename ConcurrentQueue<TValue, Suspendable<TSpec>>::SizeType
length(ConcurrentQueue<TValue, Suspendable<TSpec>> const & me)
{
    return me.occupied;
}
 
template <typename TValue, typename TSpec>
inline bool
_popFront(TValue & result, ConcurrentQueue<TValue, Suspendable<TSpec>> & me, std::unique_lock<std::mutex> & lk)
{
    typedef ConcurrentQueue<TValue, Suspendable<TSpec>> TQueue;
    typedef typename TQueue::TString TString;
    typedef typename TString::size_type TSize;
 
    TSize cap = me.data.size();
 
    while (me.occupied == 0u && me.writerCount > 0u)
        me.more.wait(lk);
 
    if (me.occupied == 0u)
        return false;
 
    assert(me.occupied > 0u);
 
    // extract value and destruct it in the data string
    // TIter it = me.data.begin() + me.front;
    result = std::ranges::iter_move(std::ranges::next(me.data.begin(), me.front));
    // std::swap(result, *it);
    // valueDestruct(it);
 
    me.front = (me.front + 1) % cap;
    me.occupied--;
 
    /* now: either me.occupied > 0 and me.nextout is the index
       of the next occupied slot in the buffer, or
       me.occupied == 0 and me.nextout is the index of the next
       (empty) slot that will be filled by a producer (such as
       me.nextout == me.nextin) */
 
    return true;
}
 
template <typename TValue, typename TSpec>
inline bool
_popBack(TValue & result, ConcurrentQueue<TValue, Suspendable<TSpec>> & me, std::unique_lock<std::mutex> & lk)
{
    typedef ConcurrentQueue<TValue, Suspendable<TSpec>> TQueue;
    typedef typename TQueue::TString TString;
    typedef typename TString::size_type TSize;
 
    TSize cap = me.data.size();
 
    while (me.occupied == 0u && me.writerCount > 0u)
        me.more.wait(lk);
 
    if (me.occupied == 0u)
        return false;
 
    assert(me.occupied > 0u);
 
    me.back = (me.back + cap - 1) % cap;
 
    // extract value and destruct it in the data string
    // TIter it = me.data.begin() + me.back;
    result = std::ranges::iter_move(std::ranges::next(me.data.begin(), me.back));
    // std::swap(result, *it);
    // valueDestruct(it);
 
    me.occupied--;
 
    /* now: either me.occupied > 0 and me.nextout is the index
       of the next occupied slot in the buffer, or
       me.occupied == 0 and me.nextout is the index of the next
       (empty) slot that will be filled by a producer (such as
       me.nextout == me.nextin) */
 
    return true;
}
 
template <typename TValue, typename TSpec>
inline bool popFront(TValue & result, ConcurrentQueue<TValue, Suspendable<TSpec>> & me)
{
    std::unique_lock<std::mutex> lock(me.cs);
    return _popFront(result, me, lock);
}
 
template <typename TValue>
inline bool popFront(TValue & result, ConcurrentQueue<TValue, Suspendable<Limit>> & me)
{
    {
        std::unique_lock<std::mutex> lk(me.cs);
        if (!_popFront(result, me, lk))
            return false;
    }
    me.less.notify_all();
    return true;
}
 
template <typename TValue, typename TSpec>
inline bool popBack(TValue & result, ConcurrentQueue<TValue, Suspendable<TSpec>> & me)
{
    std::unique_lock<std::mutex> lk(me.cs);
    return _popBack(result, me, lk);
}
 
template <typename TValue>
inline bool popBack(TValue & result, ConcurrentQueue<TValue, Suspendable<Limit>> & me)
{
    {
        std::unique_lock<std::mutex> lk(me.cs);
        if (!_popBack(result, me, lk))
            return false;
    }
    me.less.notify_all();
    return true;
}
 
template <typename TValue, typename TValue2, typename TSpec, typename TExpand>
inline bool
appendValue(ConcurrentQueue<TValue, Suspendable<TSpec>> & me, TValue2 && val, [[maybe_unused]] Tag<TExpand> expandTag)
{
    typedef ConcurrentQueue<TValue, Suspendable<TSpec>> TQueue;
    typedef typename TQueue::TString TString;
    typedef typename TString::size_type TSize;
 
    {
        std::lock_guard<std::mutex> lock(me.cs);
        TSize cap = me.data.size();
 
        if (me.occupied >= cap)
        {
            // increase capacity
            // _setLength(me.data, cap);
            // reserve(me.data, cap + 1, expandTag);
            me.data.resize(cap + 1);
            TSize delta = me.data.size() - cap;
            assert(delta == 1);
 
            // create a gap of delta many values between tail and head
            // Why?
            // _clearSpace(me.data, delta, me.back, me.back, expandTag);
            std::ranges::move_backward(std::span{me.data.data() + me.front, me.data.data() + cap},
                                       me.data.data() + me.data.size());
            if (me.occupied != 0 && me.back <= me.front)
                me.front += delta;
 
            cap += delta;
        }
 
        // valueConstruct(begin(me.data, Standard()) + me.back, val);
        *std::ranges::next(me.data.begin(), me.back) = std::forward<TValue2>(val);
        me.back = (me.back + 1) % cap;
 
        ++me.occupied;
    }
 
    /* now: either me.occupied < BSIZE and me.nextin is the index
     of the next empty slot in the buffer, or
     me.occupied == BSIZE and me.nextin is the index of the
     next (occupied) slot that will be emptied by a consumer
     (such as me.nextin == me.nextout) */
    me.more.notify_all();
    return true;
}
 
template <typename TValue, typename TValue2, typename TSpec, typename TExpand>
inline bool appendValue(ConcurrentQueue<TValue, Suspendable<Limit>> & me, TValue2 && val, Tag<TExpand> expandTag);
 
template <typename TValue, typename TValue2>
inline bool appendValue(ConcurrentQueue<TValue, Suspendable<Limit>> & me, TValue2 && val, Limit)
{
    typedef ConcurrentQueue<TValue, Suspendable<Limit>> TQueue;
    typedef typename TQueue::TString TString;
    typedef typename TString::size_type TSize;
 
    {
        std::unique_lock<std::mutex> lock(me.cs);
        TSize cap = me.data.size();
 
        while (me.occupied >= cap && me.readerCount > 0u)
            me.less.wait(lock);
 
        if (me.occupied >= cap)
            return false;
 
        assert(me.occupied < cap);
 
        // valueConstruct(begin(me.data, Standard()) + me.back, val);
        *std::ranges::next(me.data.begin(), me.back) = std::forward<TValue2>(val);
        me.back = (me.back + 1) % cap;
        me.occupied++;
    }
 
    /* now: either me.occupied < BSIZE and me.nextin is the index
     of the next empty slot in the buffer, or
     me.occupied == BSIZE and me.nextin is the index of the
     next (occupied) slot that will be emptied by a consumer
     (such as me.nextin == me.nextout) */
    me.more.notify_all();
    return true;
}
 
template <typename TValue, typename TValue2, typename TSpec>
inline bool appendValue(ConcurrentQueue<TValue, Suspendable<TSpec>> & me, TValue2 && val)
{
    return appendValue(me, std::forward<TValue2>(val), TSpec{});
}
 
} // namespace seqan3::contrib