atomic_buffer_queue.h

Go to the documentation of this file.

// Evo C++ Library
/* Copyright 2019 Justin Crowell
Distributed under the BSD 2-Clause License -- see included file LICENSE.txt for details.
*/

#pragma once
#ifndef INCL_evo_atomic_buffer_queue_h
#define INCL_evo_atomic_buffer_queue_h

#include "atomic.h"

namespace evo {


template<class T, class TSize=SizeT>
class AtomicBufferQueue {
public:
    typedef AtomicBufferQueue<T,TSize> This;    
    typedef TSize Size;                         
    typedef T     Item;                         

    static const Size DEFAULT_SIZE = 128;       

    AtomicBufferQueue(Size size=DEFAULT_SIZE) {
        size = size_pow2(size);
        buf_       = new Item[size];
        size_      = size;
        size_mask_ = size - 1;
        next_pos_.store(1);
        read_pos_.store(1);
    }

    ~AtomicBufferQueue() {
        delete [] buf_;
    }

    Size size() const {
        return size_;
    }

    Size used() const {
        const uint64 cursor = cursor_pos_.load(EVO_ATOMIC_ACQUIRE);
        const uint64 read   = read_pos_.load(EVO_ATOMIC_ACQUIRE);
        return (Size)(cursor < read ? 0 : cursor + 1 - read);
    }

    bool empty() const {
        return (cursor_pos_.load(EVO_ATOMIC_ACQUIRE) < read_pos_.load(EVO_ATOMIC_ACQUIRE));
    }

    bool full() const {
        return (used() >= size_);
    }

    void clear() {
        next_pos_.store(1, EVO_ATOMIC_RELEASE);
        cursor_pos_.store(0, EVO_ATOMIC_RELEASE);
        read_pos_.store(1, EVO_ATOMIC_RELEASE);
    }

    void add(typename DataCopy<Item>::PassType item) {
        // Claim a slot and wait for available capacity
        const uint64 seq = next_pos_.fetch_add(1, EVO_ATOMIC_ACQ_REL);
        while (seq - read_pos_.load(EVO_ATOMIC_ACQUIRE) >= size_)
            sleepus(1);

        // Store event in queue
        EVO_EXCEPTION_GUARD_START
        EVO_ATOMIC_FENCE(EVO_ATOMIC_ACQUIRE);
        buf_[seq & size_mask_] = item;
        EVO_ATOMIC_FENCE(EVO_ATOMIC_RELEASE);
        EVO_EXCEPTION_GUARD_END

        // Wait for cursor to reach previous slot, then increment cursor to commit the write
        const uint64 prev_seq = seq - 1;
        while (!cursor_pos_.compare_set(prev_seq, seq, EVO_ATOMIC_ACQ_REL, EVO_ATOMIC_ACQUIRE))
            sleepus(1);
    }

    Item& add_start(uint64& seq) {
        // Claim a slot and wait for available capacity
        seq = next_pos_.fetch_add(1, EVO_ATOMIC_ACQ_REL);
        while (seq - read_pos_.load(EVO_ATOMIC_ACQUIRE) >= size_)
            sleepus(1);

        // Return event, user will set it
        EVO_ATOMIC_FENCE(EVO_ATOMIC_ACQUIRE);
        return buf_[seq & size_mask_];
    }

    void add_commit(uint64 seq) {
        // Wait for cursor to reach previous slot, then increment cursor to commit the write
        EVO_ATOMIC_FENCE(EVO_ATOMIC_RELEASE);
        const uint64 prev_seq = seq - 1;
        while (!cursor_pos_.compare_set(prev_seq, seq, EVO_ATOMIC_ACQ_REL, EVO_ATOMIC_ACQUIRE))
            sleepus(1);
    }

    bool pop(Item& item) {
        uint64 seq;
        seq = read_pos_.load(EVO_ATOMIC_ACQUIRE);
        if (seq <= cursor_pos_.load(EVO_ATOMIC_ACQUIRE)) {
            item = buf_[seq & size_mask_];
            read_pos_.fetch_add(1, EVO_ATOMIC_RELEASE);
            return true;
        }
        return false;
    }

private:
    // Disable copying
    AtomicBufferQueue(const This&);
    This& operator=(const This&);

    T*   buf_;
    Size size_;                 // Must be a power of 2 for mask to work
    Size size_mask_;            // Mask for faster modulus

    // Positions increase to infinity (index = pos % ringbuf_size_), would take hundreds of years to max out 64 bits
    AtomicUInt64 cursor_pos_;   // Position of latest item committed to queue
    AtomicUInt64 next_pos_;     // Next write position in queue (cursor + 1 when no add() in progress)
    AtomicUInt64 read_pos_;     // Position of next item to read from queue (cursor + 1 when queue is empty)
};


}
#endif