Queue.h

Go to the documentation of this file.

/*
 * Copyright (C) 1996-2023 The Squid Software Foundation and contributors
 *
 * Squid software is distributed under GPLv2+ license and includes
 * contributions from numerous individuals and organizations.
 * Please see the COPYING and CONTRIBUTORS files for details.
 */
 
#ifndef SQUID_IPC_QUEUE_H
#define SQUID_IPC_QUEUE_H
 
#include "base/InstanceId.h"
#include "debug/Stream.h"
#include "ipc/mem/FlexibleArray.h"
#include "ipc/mem/Pointer.h"
#include "util.h"
 
#include <algorithm>
#include <atomic>
 
class String;
 
namespace Ipc
{
 
class QueueReader
{
public:
    QueueReader(); // the initial state is "blocked without a signal"
 
    bool blocked() const { return popBlocked.load(); }
 
    bool signaled() const { return popSignal.load(); }
 
    void block() { popBlocked.store(true); }
 
    void unblock() { popBlocked.store(false); }
 
    bool raiseSignal() { return blocked() && !popSignal.exchange(true); }
 
    void clearSignal() { unblock(); popSignal.store(false); }
 
private:
    std::atomic<bool> popBlocked; 
    std::atomic<bool> popSignal; 
 
public:
    typedef std::atomic<int> Rate; 
    Rate rateLimit; 
 
    // we need a signed atomic type because balance may get negative
    typedef std::atomic<int> AtomicSignedMsec;
    typedef AtomicSignedMsec Balance;
    Balance balance;
 
    const InstanceId<QueueReader> id;
};
 
class QueueReaders
{
public:
    QueueReaders(const int aCapacity);
    size_t sharedMemorySize() const;
    static size_t SharedMemorySize(const int capacity);
 
    const int theCapacity; 
    Ipc::Mem::FlexibleArray<QueueReader> theReaders; 
};
 
class OneToOneUniQueue
{
public:
    // pop() and push() exceptions; TODO: use TextException instead
    class Full {};
    class ItemTooLarge {};
 
    OneToOneUniQueue(const unsigned int aMaxItemSize, const int aCapacity);
 
    unsigned int maxItemSize() const { return theMaxItemSize; }
    int size() const { return theSize; }
    int capacity() const { return theCapacity; }
    int sharedMemorySize() const { return Items2Bytes(theMaxItemSize, theCapacity); }
 
    bool empty() const { return !theSize; }
    bool full() const { return theSize == theCapacity; }
 
    static int Bytes2Items(const unsigned int maxItemSize, int size);
    static int Items2Bytes(const unsigned int maxItemSize, const int size);
 
    template<class Value> bool pop(Value &value, QueueReader *const reader = nullptr);
 
    template<class Value> bool push(const Value &value, QueueReader *const reader = nullptr);
 
    template<class Value> bool peek(Value &value) const;
 
    template<class Value> void statIn(std::ostream &, int localProcessId, int remoteProcessId) const;
    template<class Value> void statOut(std::ostream &, int localProcessId, int remoteProcessId) const;
 
private:
    void statOpen(std::ostream &, const char *inLabel, const char *outLabel, uint32_t count) const;
    void statClose(std::ostream &) const;
    template<class Value> void statSamples(std::ostream &, unsigned int start, uint32_t size) const;
    template<class Value> void statRange(std::ostream &, unsigned int start, uint32_t n) const;
 
    // optimization: these non-std::atomic data members are in shared memory,
    // but each is used only by one process (aside from obscured reporting)
    unsigned int theIn; 
    unsigned int theOut; 
 
    std::atomic<uint32_t> theSize; 
    const unsigned int theMaxItemSize; 
    const uint32_t theCapacity; 
 
    char theBuffer[];
};
 
class OneToOneUniQueues
{
public:
    OneToOneUniQueues(const int aCapacity, const unsigned int maxItemSize, const int queueCapacity);
 
    size_t sharedMemorySize() const;
    static size_t SharedMemorySize(const int capacity, const unsigned int maxItemSize, const int queueCapacity);
 
    const OneToOneUniQueue &operator [](const int index) const;
    inline OneToOneUniQueue &operator [](const int index);
 
private:
    inline const OneToOneUniQueue &front() const;
 
public:
    const int theCapacity; 
};
 
class BaseMultiQueue
{
public:
    BaseMultiQueue(const int aLocalProcessId);
    virtual ~BaseMultiQueue() {}
 
    void clearReaderSignal(const int remoteProcessId);
 
    void clearAllReaderSignals();
 
    template <class Value> bool pop(int &remoteProcessId, Value &value);
 
    template <class Value> bool push(const int remoteProcessId, const Value &value);
 
    template<class Value> bool peek(int &remoteProcessId, Value &value) const;
 
    template<class Value> void stat(std::ostream &) const;
 
    QueueReader::Balance &localBalance() { return localReader().balance; }
 
    const QueueReader::Balance &balance(const int remoteProcessId) const;
 
    QueueReader::Rate &localRateLimit() { return localReader().rateLimit; }
 
    const QueueReader::Rate &rateLimit(const int remoteProcessId) const;
 
    int inSize(const int remoteProcessId) const { return inQueue(remoteProcessId).size(); }
 
    int outSize(const int remoteProcessId) const { return outQueue(remoteProcessId).size(); }
 
protected:
    virtual const OneToOneUniQueue &inQueue(const int remoteProcessId) const = 0;
    OneToOneUniQueue &inQueue(const int remoteProcessId);
 
    virtual const OneToOneUniQueue &outQueue(const int remoteProcessId) const = 0;
    OneToOneUniQueue &outQueue(const int remoteProcessId);
 
    virtual const QueueReader &localReader() const = 0;
    QueueReader &localReader();
 
    virtual const QueueReader &remoteReader(const int remoteProcessId) const = 0;
    QueueReader &remoteReader(const int remoteProcessId);
 
    virtual int remotesCount() const = 0;
    virtual int remotesIdOffset() const = 0;
 
protected:
    const int theLocalProcessId; 
 
private:
    int theLastPopProcessId; 
};
 
class FewToFewBiQueue: public BaseMultiQueue
{
public:
    typedef OneToOneUniQueue::Full Full;
    typedef OneToOneUniQueue::ItemTooLarge ItemTooLarge;
 
private:
    struct Metadata {
        Metadata(const int aGroupASize, const int aGroupAIdOffset, const int aGroupBSize, const int aGroupBIdOffset);
        size_t sharedMemorySize() const { return sizeof(*this); }
        static size_t SharedMemorySize(const int, const int, const int, const int) { return sizeof(Metadata); }
 
        const int theGroupASize;
        const int theGroupAIdOffset;
        const int theGroupBSize;
        const int theGroupBIdOffset;
    };
 
public:
    class Owner
    {
    public:
        Owner(const String &id, const int groupASize, const int groupAIdOffset, const int groupBSize, const int groupBIdOffset, const unsigned int maxItemSize, const int capacity);
        ~Owner();
 
    private:
        Mem::Owner<Metadata> *const metadataOwner;
        Mem::Owner<OneToOneUniQueues> *const queuesOwner;
        Mem::Owner<QueueReaders> *const readersOwner;
    };
 
    static Owner *Init(const String &id, const int groupASize, const int groupAIdOffset, const int groupBSize, const int groupBIdOffset, const unsigned int maxItemSize, const int capacity);
 
    enum Group { groupA = 0, groupB = 1 };
    FewToFewBiQueue(const String &id, const Group aLocalGroup, const int aLocalProcessId);
 
    static int MaxItemsCount(const int groupASize, const int groupBSize, const int capacity);
 
    template<class Value> bool findOldest(const int remoteProcessId, Value &value) const;
 
protected:
    const OneToOneUniQueue &inQueue(const int remoteProcessId) const override;
    const OneToOneUniQueue &outQueue(const int remoteProcessId) const override;
    const QueueReader &localReader() const override;
    const QueueReader &remoteReader(const int processId) const override;
    int remotesCount() const override;
    int remotesIdOffset() const override;
 
private:
    bool validProcessId(const Group group, const int processId) const;
    int oneToOneQueueIndex(const Group fromGroup, const int fromProcessId, const Group toGroup, const int toProcessId) const;
    const OneToOneUniQueue &oneToOneQueue(const Group fromGroup, const int fromProcessId, const Group toGroup, const int toProcessId) const;
    int readerIndex(const Group group, const int processId) const;
    Group localGroup() const { return theLocalGroup; }
    Group remoteGroup() const { return theLocalGroup == groupA ? groupB : groupA; }
 
private:
    const Mem::Pointer<Metadata> metadata; 
    const Mem::Pointer<OneToOneUniQueues> queues; 
    const Mem::Pointer<QueueReaders> readers; 
 
    const Group theLocalGroup; 
};
 
class MultiQueue: public BaseMultiQueue
{
public:
    typedef OneToOneUniQueue::Full Full;
    typedef OneToOneUniQueue::ItemTooLarge ItemTooLarge;
 
private:
    struct Metadata {
        Metadata(const int aProcessCount, const int aProcessIdOffset);
        size_t sharedMemorySize() const { return sizeof(*this); }
        static size_t SharedMemorySize(const int, const int) { return sizeof(Metadata); }
 
        const int theProcessCount;
        const int theProcessIdOffset;
    };
 
public:
    class Owner
    {
    public:
        Owner(const String &id, const int processCount, const int processIdOffset, const unsigned int maxItemSize, const int capacity);
        ~Owner();
 
    private:
        Mem::Owner<Metadata> *const metadataOwner;
        Mem::Owner<OneToOneUniQueues> *const queuesOwner;
        Mem::Owner<QueueReaders> *const readersOwner;
    };
 
    static Owner *Init(const String &id, const int processCount, const int processIdOffset, const unsigned int maxItemSize, const int capacity);
 
    MultiQueue(const String &id, const int localProcessId);
 
protected:
    const OneToOneUniQueue &inQueue(const int remoteProcessId) const override;
    const OneToOneUniQueue &outQueue(const int remoteProcessId) const override;
    const QueueReader &localReader() const override;
    const QueueReader &remoteReader(const int remoteProcessId) const override;
    int remotesCount() const override;
    int remotesIdOffset() const override;
 
private:
    bool validProcessId(const int processId) const;
    const OneToOneUniQueue &oneToOneQueue(const int fromProcessId, const int toProcessId) const;
    const QueueReader &reader(const int processId) const;
 
private:
    const Mem::Pointer<Metadata> metadata; 
    const Mem::Pointer<OneToOneUniQueues> queues; 
    const Mem::Pointer<QueueReaders> readers; 
};
 
// OneToOneUniQueue
 
template <class Value>
bool
OneToOneUniQueue::pop(Value &value, QueueReader *const reader)
{
    if (sizeof(value) > theMaxItemSize)
        throw ItemTooLarge();
 
    // A writer might push between the empty test and block() below, so we do
    // not return false right after calling block(), but test again.
    if (empty()) {
        if (!reader)
            return false;
 
        reader->block();
        // A writer might push between the empty test and block() below,
        // so we must test again as such a writer will not signal us.
        if (empty())
            return false;
    }
 
    if (reader)
        reader->unblock();
 
    const unsigned int pos = (theOut++ % theCapacity) * theMaxItemSize;
    memcpy(&value, theBuffer + pos, sizeof(value));
    --theSize;
 
    return true;
}
 
template <class Value>
bool
OneToOneUniQueue::peek(Value &value) const
{
    if (sizeof(value) > theMaxItemSize)
        throw ItemTooLarge();
 
    if (empty())
        return false;
 
    // the reader may pop() before we copy; making this method imprecise
    const unsigned int pos = (theOut % theCapacity) * theMaxItemSize;
    memcpy(&value, theBuffer + pos, sizeof(value));
    return true;
}
 
template <class Value>
bool
OneToOneUniQueue::push(const Value &value, QueueReader *const reader)
{
    if (sizeof(value) > theMaxItemSize)
        throw ItemTooLarge();
 
    if (full())
        throw Full();
 
    const unsigned int pos = theIn++ % theCapacity * theMaxItemSize;
    memcpy(theBuffer + pos, &value, sizeof(value));
    const bool wasEmpty = !theSize++;
 
    return wasEmpty && (!reader || reader->raiseSignal());
}
 
template <class Value>
void
OneToOneUniQueue::statIn(std::ostream &os, const int localProcessId, const int remoteProcessId) const
{
    os << "  kid" << localProcessId << " receiving from kid" << remoteProcessId << ": ";
    // Nobody can modify our theOut so, after capturing some valid theSize value
    // in count, we can reliably report all [theOut, theOut+count) items that
    // were queued at theSize capturing time. We will miss new items push()ed by
    // the other side, but it is OK -- we report state at the capturing time.
    const auto count = theSize.load();
    statOpen(os, "other", "popIndex", count);
    statSamples<Value>(os, theOut, count);
    statClose(os);
}
 
template <class Value>
void
OneToOneUniQueue::statOut(std::ostream &os, const int localProcessId, const int remoteProcessId) const
{
    os << "  kid" << localProcessId << " sending to kid" << remoteProcessId << ": ";
    // Nobody can modify our theIn so, after capturing some valid theSize value
    // in count, we can reliably report all [theIn-count, theIn) items that were
    // queued at theSize capturing time. We may report items already pop()ed by
    // the other side, but that is OK because pop() does not modify items -- it
    // only increments theOut.
    const auto count = theSize.load();
    statOpen(os, "pushIndex", "other", count);
    statSamples<Value>(os, theIn - count, count); // unsigned offset underflow OK
    statClose(os);
}
 
template <class Value>
void
OneToOneUniQueue::statSamples(std::ostream &os, const unsigned int start, const uint32_t count) const
{
    if (!count) {
        os << " ";
        return;
    }
 
    os << ", items: [\n";
    // report a few leading and trailing items, without repetitions
    const auto sampleSize = std::min(3U, count); // leading (and max) sample
    statRange<Value>(os, start, sampleSize);
    if (sampleSize < count) { // the first sample did not show some items
        // The `start` offset aside, the first sample reported all items
        // below the sampleSize offset. The second sample needs to report
        // the last sampleSize items (i.e. starting at count-sampleSize
        // offset) except those already reported by the first sample.
        const auto secondSampleOffset = std::max(sampleSize, count - sampleSize);
        const auto secondSampleSize = std::min(sampleSize, count - sampleSize);
 
        // but first we print a sample separator, unless there are no items
        // between the samples or the separator hides the only unsampled item
        const auto bothSamples = sampleSize + secondSampleSize;
        if (bothSamples + 1U == count)
            statRange<Value>(os, start + sampleSize, 1);
        else if (count > bothSamples)
            os << "    # ... " << (count - bothSamples) << " items not shown ...\n";
 
        statRange<Value>(os, start + secondSampleOffset, secondSampleSize);
    }
    os << "  ]";
}
 
template <class Value>
void
OneToOneUniQueue::statRange(std::ostream &os, const unsigned int start, const uint32_t n) const
{
    assert(sizeof(Value) <= theMaxItemSize);
    auto offset = start;
    for (uint32_t i = 0; i < n; ++i) {
        // XXX: Throughout this C++ header, these overflow wrapping tricks work
        // only because theCapacity currently happens to be a power of 2 (e.g.,
        // the highest offset (0xF...FFF) % 3 is 0 and so is the next offset).
        const auto pos = (offset++ % theCapacity) * theMaxItemSize;
        Value value;
        memcpy(&value, theBuffer + pos, sizeof(value));
        os << "    { ";
        value.stat(os);
        os << " },\n";
    }
}
 
// OneToOneUniQueues
 
inline OneToOneUniQueue &
OneToOneUniQueues::operator [](const int index)
{
    return const_cast<OneToOneUniQueue &>((*const_cast<const OneToOneUniQueues *>(this))[index]);
}
 
inline const OneToOneUniQueue &
OneToOneUniQueues::front() const
{
    const char *const queue =
        reinterpret_cast<const char *>(this) + sizeof(*this);
    return *reinterpret_cast<const OneToOneUniQueue *>(queue);
}
 
// BaseMultiQueue
 
template <class Value>
bool
BaseMultiQueue::pop(int &remoteProcessId, Value &value)
{
    // iterate all remote processes, starting after the one we visited last
    for (int i = 0; i < remotesCount(); ++i) {
        if (++theLastPopProcessId >= remotesIdOffset() + remotesCount())
            theLastPopProcessId = remotesIdOffset();
        OneToOneUniQueue &queue = inQueue(theLastPopProcessId);
        if (queue.pop(value, &localReader())) {
            remoteProcessId = theLastPopProcessId;
            debugs(54, 7, "popped from " << remoteProcessId << " to " << theLocalProcessId << " at " << queue.size());
            return true;
        }
    }
    return false; // no process had anything to pop
}
 
template <class Value>
bool
BaseMultiQueue::push(const int remoteProcessId, const Value &value)
{
    OneToOneUniQueue &remoteQueue = outQueue(remoteProcessId);
    QueueReader &reader = remoteReader(remoteProcessId);
    debugs(54, 7, "pushing from " << theLocalProcessId << " to " << remoteProcessId << " at " << remoteQueue.size());
    return remoteQueue.push(value, &reader);
}
 
template <class Value>
bool
BaseMultiQueue::peek(int &remoteProcessId, Value &value) const
{
    // mimic FewToFewBiQueue::pop() but quit just before popping
    int popProcessId = theLastPopProcessId; // preserve for future pop()
    for (int i = 0; i < remotesCount(); ++i) {
        if (++popProcessId >= remotesIdOffset() + remotesCount())
            popProcessId = remotesIdOffset();
        const OneToOneUniQueue &queue = inQueue(popProcessId);
        if (queue.peek(value)) {
            remoteProcessId = popProcessId;
            return true;
        }
    }
    return false; // most likely, no process had anything to pop
}
 
template <class Value>
void
BaseMultiQueue::stat(std::ostream &os) const
{
    for (int processId = remotesIdOffset(); processId < remotesIdOffset() + remotesCount(); ++processId) {
        const auto &queue = inQueue(processId);
        queue.statIn<Value>(os, theLocalProcessId, processId);
    }
 
    os << "\n";
 
    for (int processId = remotesIdOffset(); processId < remotesIdOffset() + remotesCount(); ++processId) {
        const auto &queue = outQueue(processId);
        queue.statOut<Value>(os, theLocalProcessId, processId);
    }
 
    os << "\n";
 
    const auto &reader = localReader();
    os << "  kid" << theLocalProcessId << " reader flags: " <<
       "{ blocked: " << reader.blocked() << ", signaled: " << reader.signaled() << " }\n";
}
 
// FewToFewBiQueue
 
template <class Value>
bool
FewToFewBiQueue::findOldest(const int remoteProcessId, Value &value) const
{
    // we may be called before remote process configured its queue end
    if (!validProcessId(remoteGroup(), remoteProcessId))
        return false;
 
    // we need the oldest value, so start with the incoming, them-to-us queue:
    const OneToOneUniQueue &in = inQueue(remoteProcessId);
    debugs(54, 2, "peeking from " << remoteProcessId << " to " <<
           theLocalProcessId << " at " << in.size());
    if (in.peek(value))
        return true;
 
    // if the incoming queue is empty, check the outgoing, us-to-them queue:
    const OneToOneUniQueue &out = outQueue(remoteProcessId);
    debugs(54, 2, "peeking from " << theLocalProcessId << " to " <<
           remoteProcessId << " at " << out.size());
    return out.peek(value);
}
 
} // namespace Ipc
 
#endif // SQUID_IPC_QUEUE_H