reader.h

Go to the documentation of this file.

/*
 *  rtmlib is a Real-Time Monitoring Library.
 *
 *    Copyright (C) 2018-2020 André Pedro
 *
 *  This file is part of rtmlib.
 *
 *  This program is free software: you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation, either version 3 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#ifndef _RTEML_READER_H_
#define _RTEML_READER_H_

#include <array>
#include <time.h>
#include <tuple>
#include <utility>

#include "circularbuffer.h"
#include "event.h"

template <typename B> class RTML_reader {
protected:
  const B &buffer;

  size_t top;

  size_t bottom;

  timespanw timestamp;

public:
  typedef B buffer_t;

  typedef enum { AVAILABLE = 0, UNAVAILABLE, OVERFLOW, BUFFER_READ } error_t;

  RTML_reader(const B &);

  error_t pull(typename B::event_t &);

  error_t pop(typename B::event_t &);

  bool synchronize();

  bool gap() const;

  size_t length() const;

  void debug() const;
};

template <typename B>
RTML_reader<B>::RTML_reader(const B &_buffer)
    : buffer(_buffer), top(0), bottom(0), timestamp(0) {}

template <typename B>
typename RTML_reader<B>::error_t
RTML_reader<B>::pull(typename B::event_t &event) {

  typename B::event_t event_next;

  if (length() > 0) {
    // read in an atomic way
    if (buffer.read(event, bottom) != buffer.OK)
      return BUFFER_READ;

    // update local bottom
    bottom = (size_t)(bottom + 1) % (buffer.size + 0);
    if (buffer.read(event_next, bottom) == buffer.OK)
      timestamp = event_next.getTime();
  }

  DEBUGV3("pull-> %d (%d,%d)\n", length(), bottom, top);

  if (gap())
    return OVERFLOW;
  else
    return (length() > 0) ? AVAILABLE : UNAVAILABLE;
}

template <typename B>
typename RTML_reader<B>::error_t
RTML_reader<B>::pop(typename B::event_t &event) {

  if (length() > 0) {
    if (top - 1 > 0)
      buffer.read(event, --top);
    else if (top - 1 <= 0) {
      top = buffer.size;
      buffer.read(event, top);
    }
  }

  DEBUGV3("pop-> %d (%d,%d)\n", length(), bottom, top);

  if (gap())
    return OVERFLOW;
  else
    return (length() > 0) ? AVAILABLE : UNAVAILABLE;
}

template <typename B> bool RTML_reader<B>::synchronize() {
  /*
   * The synchronization depends on the buffer state. Any event that is
   * overwritten without being read is a gap for the reader.
   */

  size_t b, t;
  timespanw ts;
  buffer.state(b, t, ts);

  if (timestamp > ts) {
    // no gaps, but the reader cannot be further ahead
    top = t; // set new top
    return false;
  } else {
    // update reader state
    top = t;
    bottom = b;
    timestamp = ts;

    // there is a gap
    return true;
  }
}

template <typename B> bool RTML_reader<B>::gap() const {
  typename B::event_t event;

  // detect a gap
  buffer.read(event, bottom);
  if (timestamp < event.getTime())
    return true;

  return false;
}

template <typename B> size_t RTML_reader<B>::length() const {
  return (top >= bottom) ? top - bottom : (buffer.size + 1) - (bottom - top);
}

template <typename B> void RTML_reader<B>::debug() const {
  DEBUGV("bottom:%d top:%d timestamp:%lu\n", bottom, top, timestamp);
}

#endif //_RTEML_READER_H_