ringbuffer.h

Go to the documentation of this file.

#ifndef __RINGBUFFER_H
#define __RINGBUFFER_H

#include "w_defines.h"
#include "basics.h"
#include "w_debug.h"
#include "vec_t.h"
#include "smthread.h"
#include "sm_base.h"

class AsyncRingBuffer {
public:
    char* producerRequest();

    void producerRelease();

    char* consumerRequest();

    void consumerRelease();

    bool isFull() {
        return begin == end && bparity != eparity;
    }

    bool isEmpty() {
        return begin == end && bparity == eparity;
    }

    size_t getBlockSize() {
        return blockSize;
    }

    size_t getBlockCount() {
        return blockCount;
    }

    void set_finished(bool f = true) {
        finished = f;
    }

    bool* get_finished() {
        return &finished;
    } // not thread-safe
    bool isFinished(); // thread-safe


    AsyncRingBuffer(size_t bsize, size_t bcount)
            : begin(0),
              end(0),
              bparity(true),
              eparity(true),
              finished(false),
              blockSize(bsize),
              blockCount(bcount) {
        buf = new char[blockCount * blockSize];
        DO_PTHREAD(pthread_mutex_init(&mutex, nullptr));
        DO_PTHREAD(pthread_cond_init(&notEmpty, nullptr));
        DO_PTHREAD(pthread_cond_init(&notFull, nullptr));
    }

    ~AsyncRingBuffer() {
        delete buf;
        DO_PTHREAD(pthread_mutex_destroy(&mutex));
        DO_PTHREAD(pthread_cond_destroy(&notEmpty));
        DO_PTHREAD(pthread_cond_destroy(&notFull));
    }

private:
    char* buf;

    int begin;

    int end;

    bool bparity;

    bool eparity;

    bool finished;

    const size_t blockSize;

    const size_t blockCount;

    pthread_mutex_t mutex;

    pthread_cond_t notEmpty;

    pthread_cond_t notFull;

    bool wait(pthread_cond_t*);

    void increment(int& p, bool& parity) {
        p = (p + 1) % blockCount;
        if (p == 0) {
            parity = !parity;
        }
    }
};

inline bool AsyncRingBuffer::wait(pthread_cond_t* cond) {
    struct timespec timeout;
    smthread_t::timeout_to_timespec(100, timeout); // 100ms
    // caller must have locked mutex!
    int code = pthread_cond_timedwait(cond, &mutex, &timeout);
    if (code == ETIMEDOUT) {
        //DBGTHRD(<< "Wait timed out -- try again");
        if (finished && isEmpty()) {
            DBGTHRD(<< "Wait aborted: finished flag is set");
            return false;
        }
    }
    DO_PTHREAD_TIMED(code);
    return true;
}

inline char* AsyncRingBuffer::producerRequest() {
    CRITICAL_SECTION(cs, mutex);
    while (isFull()) {
        DBGTHRD(<< "Waiting for condition notFull ...");
        if (!wait(&notFull)) {
            DBGTHRD(<< "Produce request failed!");
            return nullptr;
        }
    }
    DBGTHRD(<< "Producer request: block " << end);
    return buf + (end * blockSize);
}

inline void AsyncRingBuffer::producerRelease() {
    CRITICAL_SECTION(cs, mutex);
    bool wasEmpty = isEmpty();
    increment(end, eparity);
    DBGTHRD(<< "Producer release, new end is " << end);

    if (wasEmpty) {
        DBGTHRD(<< "Signal buffer not empty");
        DO_PTHREAD(pthread_cond_signal(&notEmpty));
    }
}

inline char* AsyncRingBuffer::consumerRequest() {
    CRITICAL_SECTION(cs, mutex);
    while (isEmpty()) {
        DBGTHRD(<< "Waiting for condition notEmpty ...");
        if (!wait(&notEmpty)) {
            DBGTHRD(<< "Consume request failed!");
            return nullptr;
        }
    }
    DBGTHRD(<< "Consumer request: block " << begin);
    return buf + (begin * blockSize);
}

inline void AsyncRingBuffer::consumerRelease() {
    CRITICAL_SECTION(cs, mutex);
    bool wasFull = isFull();
    increment(begin, bparity);
    DBGTHRD(<< "Consumer release, new begin is " << begin);

    if (wasFull) {
        DBGTHRD(<< "Signal buffer not full");
        DO_PTHREAD(pthread_cond_signal(&notFull));
    }
}

inline bool AsyncRingBuffer::isFinished() {
    /*
     * Acquiring the mutex here is done to ensure consistency
     * of modifications to the finished flag. It creates a memory
     * fence to ensure the correct order of operations in the case
     * where one thread reading the flag after another one has set it.
     *
     * Caution! This does not mean that there are no blocks left for
     * consumption---just that someone set the finished flag. The former
     * case must be checked by calling producerRequest()
     */
    CRITICAL_SECTION(cs, mutex);
    return finished;
}

#endif // __RINGBUFFER_H