log_consumer.h

Go to the documentation of this file.

#ifndef __LOG_CONSUMER_H
#define __LOG_CONSUMER_H

#include <bitset>

#include "worker_thread.h"
#include "ringbuffer.h"
#include "log_storage.h"

class LogScanner {
public:
    bool nextLogrec(char* src, size_t& pos, logrec_t*& lr,
                    lsn_t* nextLSN = nullptr, lsn_t* stopLSN = nullptr,
                    int* lrLength = nullptr);

    bool hasPartialLogrec();

    void reset();

    LogScanner(size_t blockSize)
            : truncCopied(0),
              truncMissing(0),
              toSkip(0),
              blockSize(blockSize) {
        // maximum logrec size = 3 pages
        truncBuf = new char[3 * log_storage::BLOCK_SIZE];
    }

    ~LogScanner() {
        delete[] truncBuf;
    }

    size_t getBlockSize() {
        return blockSize;
    }

    void setIgnore(logrec_t::kind_t type) {
        ignore.set(type);
    }

    void ignoreAll() {
        ignore.set();
    }

    void unsetIgnore(logrec_t::kind_t type) {
        ignore.reset(type);
    }

    bool isIgnored(logrec_t::kind_t type) {
        return ignore[type];
    }

private:
    size_t truncCopied;

    size_t truncMissing;

    size_t toSkip;

    const size_t blockSize;

    char* truncBuf;

    bitset<logrec_t::t_max_logrec> ignore;
};

struct ArchiverControl {
    pthread_mutex_t mutex;

    pthread_cond_t activateCond;

    lsn_t endLSN;

    bool activated;

    bool listening;

    std::atomic<bool>* shutdownFlag;

    ArchiverControl(std::atomic<bool>* shutdown);

    ~ArchiverControl();

    bool activate(bool wait, lsn_t lsn = lsn_t::null);

    bool waitForActivation();
};

class ReaderThread : public log_worker_thread_t {
protected:
    uint nextPartition;

    rc_t openPartition();

    AsyncRingBuffer* buf;

    int currentFd;

    off_t pos;

    lsn_t localEndLSN;

public:
    virtual void do_work();

    ReaderThread(AsyncRingBuffer* readbuf, lsn_t startLSN);

    virtual ~ReaderThread() {}

    size_t getBlockSize() {
        return buf->getBlockSize();
    }
};

class LogConsumer {
public:
    LogConsumer(lsn_t startLSN, size_t blockSize, bool ignore = true);

    virtual ~LogConsumer();

    void shutdown();

    void open(lsn_t endLSN, bool readWholeBlocks = false);

    bool next(logrec_t*& lr);

    lsn_t getNextLSN() {
        return nextLSN;
    }

    static void initLogScanner(LogScanner* logScanner);

private:
    AsyncRingBuffer* readbuf;

    ReaderThread* reader;

    LogScanner* logScanner;

    lsn_t nextLSN;

    lsn_t endLSN;

    char* currentBlock;

    size_t blockSize;

    size_t pos;

    bool readWholeBlocks;

    bool nextBlock();
};

#endif // __LOG_CONSUMER_H