event.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_event_h
#define INCL_evo_event_h

#include "atomic.h"

#if defined(EVO_CPP11)
    #include <functional>
#endif

namespace evo {


struct Event {
    Event()
        { }

    virtual ~Event()
        { }

    virtual bool operator()() = 0;
};


#if defined(EVO_CPP11)

class EventLambda : public Event {
public:
    typedef std::function<bool()> Lambda;   

    EventLambda(const Lambda& lambda) : lambda_(lambda) {
    }

    EventLambda(const EventLambda& src) : lambda_(src.lambda_) {
    }

    EventLambda& operator=(const EventLambda& src) {
        lambda_ = src.lambda_;
        return *this;
    }

    // Doc by parent
    bool operator()() {
        return lambda_();
    }

private:
    Lambda lambda_;
};

#endif


template<class T=Event>
class EventQueue {
public:
    typedef T EventT;       
    typedef uint Size;      

    static const Size DEFAULT_SIZE = 256;   

    EventQueue(Size size=DEFAULT_SIZE) {
        ringbuf_size_ = adjust_size(size);
        ringbuf_ = new T*[ringbuf_size_];
        ringbuf_size_mask_ = ringbuf_size_ - 1; // mask: set all bits up to size
        memset(ringbuf_, 0, sizeof(T*) * ringbuf_size_);
        next_pos_.store(1);
        read_pos_.store(1);
    }

    ~EventQueue() {
        if (read_pos_.load() <= cursor_pos_.load()) {
            assert( false ); // Queue should be empty
        }
        delete [] ringbuf_;
    }

    void add(T* event, ulongl spinwait_ns=1) {
        // 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) >= ringbuf_size_)
            sleepns(spinwait_ns);

        // Store event in queue
        EVO_ATOMIC_FENCE(EVO_ATOMIC_ACQUIRE);
        ringbuf_[seq & ringbuf_size_mask_] = event;
        EVO_ATOMIC_FENCE(EVO_ATOMIC_RELEASE);

        // 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))
            sleepns(spinwait_ns);
    }

    template<class U>
    void notify_multiwait(U& condmutex) {
        if (condmutex.trylock()) { // non-blocking
            condmutex.notify();
            condmutex.unlock();
        }
    }

    bool process() {
        T* event;
        uint64 start, seq;
        start = seq = read_pos_.load(EVO_ATOMIC_ACQUIRE);
        while (seq <= cursor_pos_.load(EVO_ATOMIC_ACQUIRE)) {
            event = ringbuf_[seq & ringbuf_size_mask_];
            seq = read_pos_.fetch_add(1, EVO_ATOMIC_ACQ_REL) + 1;
            if ((*event)())
                delete event;
        }
        return (seq > start);
    }

    template<class U>
    bool process_multi(U& mutex) {
        T* event;
        uint64 seq, count = 0;
        typename U::Lock lock(mutex);
        seq = read_pos_.load(EVO_ATOMIC_RELAXED);
        for (; seq <= cursor_pos_.load(EVO_ATOMIC_ACQUIRE); ++count) {
            event = ringbuf_[seq & ringbuf_size_mask_];
            read_pos_.fetch_add(1, EVO_ATOMIC_RELEASE);
            lock.unlock();
            if ((*event)())
                delete event;
            lock.lock();
            seq = read_pos_.load(EVO_ATOMIC_RELAXED);
        }
        lock.unlock();
        return (count > 0);
    }

    template<class U>
    void process_multiwait(U& condmutex, AtomicInt& stopflag, ulong waitms=1) {
        T* event;
        uint64 seq;
        typename U::Lock lock(condmutex);
        for (;;) {
            seq = read_pos_.load(EVO_ATOMIC_RELAXED);
            for (; seq <= cursor_pos_.load(EVO_ATOMIC_ACQUIRE);) {
                event = ringbuf_[seq & ringbuf_size_mask_];
                read_pos_.fetch_add(1, EVO_ATOMIC_RELEASE);
                lock.unlock();
                if ((*event)())
                    delete event;
                lock.lock();
                seq = read_pos_.load(EVO_ATOMIC_RELAXED);
            }
            if (stopflag.load(EVO_ATOMIC_RELAXED))
                break;
            condmutex.wait(waitms, true);
        }
        lock.unlock();
    }

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

    // Ring buffer
    T**  ringbuf_;
    Size ringbuf_size_;         // Must be a power of 2 for mask to work
    Size ringbuf_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)

    // Make sure size is within min/max and is a power of 2
    static Size adjust_size(Size size) {
        const Size MIN_SIZE = 16;
        const Size MAX_SIZE = (std::numeric_limits<Size>::max() >> 1) + 1;
        if (size <= MIN_SIZE)
            size = MIN_SIZE;
        else if (size >= MAX_SIZE)
            size = MAX_SIZE;
        else
            size = next_pow2(size);
        return size;
    }
};


}
#endif