Lane.h

Go to the documentation of this file.

#ifndef CASACORE_AOCOMMON_LANE_11_H_
#define CASACORE_AOCOMMON_LANE_11_H_
 
#include <condition_variable>
#include <cstring>
#include <deque>
#include <mutex>

 
// #define LANE_DEBUG_MODE
 
#ifdef LANE_DEBUG_MODE
#include <cmath>
#include <iostream>
#include <sstream>
#include <string>
#endif
 
namespace casacore::aocommon {
 
#ifdef LANE_DEBUG_MODE
#define set_lane_debug_name(lane, str) (lane).setDebugName(str)
#define LANE_REGISTER_DEBUG_INFO registerDebugInfo()
#define LANE_REGISTER_DEBUG_WRITE_WAIT registerDebugWriteWait()
#define LANE_REGISTER_DEBUG_READ_WAIT registerDebugReadWait()
#define LANE_REPORT_DEBUG_INFO reportDebugInfo()
 
#else
 
#define set_lane_debug_name(lane, str)
#define LANE_REGISTER_DEBUG_INFO
#define LANE_REGISTER_DEBUG_WRITE_WAIT
#define LANE_REGISTER_DEBUG_READ_WAIT
#define LANE_REPORT_DEBUG_INFO
 
#endif

template <typename Tp>
class Lane {
 public:
  typedef std::size_t size_type;

  typedef Tp value_type;

  Lane() noexcept
      : _buffer(nullptr),
        _capacity(0),
        _write_position(0),
        _free_write_space(0),
        _status(status_normal) {}

  explicit Lane(size_t capacity)
      : _buffer(new Tp[capacity]),
        _capacity(capacity),
        _write_position(0),
        _free_write_space(_capacity),
        _status(status_normal) {}
 
  Lane(const Lane<Tp>& source) = delete;

  Lane(Lane<Tp>&& source) noexcept
      : _buffer(nullptr),
        _capacity(0),
        _write_position(0),
        _free_write_space(0),
        _status(status_normal) {
    swap(source);
  }

  ~Lane() {
    LANE_REPORT_DEBUG_INFO;
    delete[] _buffer;
  }
 
  Lane<Tp>& operator=(const Lane<Tp>& source) = delete;

  Lane<Tp>& operator=(Lane<Tp>&& source) noexcept {
    swap(source);
    return *this;
  }

  void swap(Lane<Tp>& other) noexcept {
    std::swap(_buffer, other._buffer);
    std::swap(_capacity, other._capacity);
    std::swap(_write_position, other._write_position);
    std::swap(_free_write_space, other._free_write_space);
    std::swap(_status, other._status);
  }

  void clear() noexcept {
    _write_position = 0;
    _free_write_space = _capacity;
    _status = status_normal;
  }

  void write(const value_type& element) {
    std::unique_lock<std::mutex> lock(_mutex);
    LANE_REGISTER_DEBUG_INFO;
 
    while (_free_write_space == 0 && _status == status_normal) {
      LANE_REGISTER_DEBUG_WRITE_WAIT;
      _writing_possible_condition.wait(lock);
    }
    if (_status == status_normal) {
      _buffer[_write_position] = element;
      _write_position = (_write_position + 1) % _capacity;
      --_free_write_space;
      // Now that there is less free write space, there is more free read
      // space and thus readers may continue.
      _reading_possible_condition.notify_all();
    }
  }

  template <typename... Args>
  void emplace(Args&&... args) {
    std::unique_lock<std::mutex> lock(_mutex);
    LANE_REGISTER_DEBUG_INFO;
 
    while (_free_write_space == 0 && _status == status_normal) {
      LANE_REGISTER_DEBUG_WRITE_WAIT;
      _writing_possible_condition.wait(lock);
    }
 
    if (_status == status_normal) {
      _buffer[_write_position] = value_type(std::forward<Args>(args)...);
      _write_position = (_write_position + 1) % _capacity;
      --_free_write_space;
      // Now that there is less free write space, there is more free read
      // space and thus readers can possibly continue.
      _reading_possible_condition.notify_all();
    }
  }

  void write(value_type&& element) {
    std::unique_lock<std::mutex> lock(_mutex);
    LANE_REGISTER_DEBUG_INFO;
 
    while (_free_write_space == 0 && _status == status_normal) {
      LANE_REGISTER_DEBUG_WRITE_WAIT;
      _writing_possible_condition.wait(lock);
    }
    if (_status == status_normal) {
      _buffer[_write_position] = std::move(element);
      _write_position = (_write_position + 1) % _capacity;
      --_free_write_space;
      // Now that there is less free write space, there is more free read
      // space and thus readers can possibly continue.
      _reading_possible_condition.notify_all();
    }
  }
 
  void write(const value_type* elements, size_t n) { write_generic(elements, n); }
 
  void move_write(value_type* elements, size_t n) { write_generic(elements, n); }
 
  bool read(value_type& destination) {
    std::unique_lock<std::mutex> lock(_mutex);
    LANE_REGISTER_DEBUG_INFO;
    while (free_read_space() == 0 && _status == status_normal) {
      LANE_REGISTER_DEBUG_READ_WAIT;
      _reading_possible_condition.wait(lock);
    }
    if (free_read_space() == 0)
      return false;
    else {
      destination = std::move(_buffer[read_position()]);
      ++_free_write_space;
      // Now that there is more free write space, writers can possibly continue.
      _writing_possible_condition.notify_all();
      return true;
    }
  }
 
  size_t read(value_type* destinations, size_t n) {
    size_t n_left = n;
 
    std::unique_lock<std::mutex> lock(_mutex);
    LANE_REGISTER_DEBUG_INFO;
 
    size_t free_space = free_read_space();
    size_t read_size = free_space > n ? n : free_space;
    immediate_read(destinations, read_size);
    n_left -= read_size;
 
    while (n_left != 0 && _status == status_normal) {
      destinations += read_size;
 
      do {
        LANE_REGISTER_DEBUG_READ_WAIT;
        _reading_possible_condition.wait(lock);
      } while (free_read_space() == 0 && _status == status_normal);
 
      free_space = free_read_space();
      read_size = free_space > n_left ? n_left : free_space;
      immediate_read(destinations, read_size);
      n_left -= read_size;
    }
    return n - n_left;
  }

  size_t discard(size_t n) {
    size_t n_left = n;
 
    std::unique_lock<std::mutex> lock(_mutex);
    LANE_REGISTER_DEBUG_INFO;
 
    size_t free_space = free_read_space();
    size_t read_size = free_space > n ? n : free_space;
    immediate_discard(read_size);
    n_left -= read_size;
 
    while (n_left != 0 && _status == status_normal) {
      do {
        LANE_REGISTER_DEBUG_READ_WAIT;
        _reading_possible_condition.wait(lock);
      } while (free_read_space() == 0 && _status == status_normal);
 
      free_space = free_read_space();
      read_size = free_space > n_left ? n_left : free_space;
      immediate_discard(read_size);
      n_left -= read_size;
    }
    return n - n_left;
  }
 
  void write_end() {
    std::lock_guard<std::mutex> lock(_mutex);
    LANE_REGISTER_DEBUG_INFO;
    _status = status_end;
    _writing_possible_condition.notify_all();
    _reading_possible_condition.notify_all();
  }
 
  size_t capacity() const noexcept { return _capacity; }
 
  size_t size() const {
    std::lock_guard<std::mutex> lock(_mutex);
    return _capacity - _free_write_space;
  }
 
  bool empty() const {
    std::lock_guard<std::mutex> lock(_mutex);
    return _capacity == _free_write_space;
  }

  bool is_end() const {
    std::lock_guard<std::mutex> lock(_mutex);
    return _status == status_end;
  }

  bool is_end_and_empty() const {
    std::lock_guard<std::mutex> lock(_mutex);
    return _status == status_end && _capacity == _free_write_space;
  }

  void resize(size_t new_capacity) {
    Tp* new_buffer = new Tp[new_capacity];
    delete[] _buffer;
    _buffer = new_buffer;
    _capacity = new_capacity;
    _write_position = 0;
    _free_write_space = new_capacity;
    _status = status_normal;
  }

  void wait_for_empty() {
    std::unique_lock<std::mutex> lock(_mutex);
    while (_capacity != _free_write_space) {
      _writing_possible_condition.wait(lock);
    }
  }
 
#ifdef LANE_DEBUG_MODE
  void setDebugName(const std::string& nameStr) { _debugName = nameStr; }
#endif
 private:
  Tp* _buffer;
 
  size_t _capacity;
 
  size_t _write_position;
 
  size_t _free_write_space;
 
  enum { status_normal, status_end } _status;
 
  mutable std::mutex _mutex;
 
  std::condition_variable _writing_possible_condition, _reading_possible_condition;
 
  size_t read_position() const noexcept {
    return (_write_position + _free_write_space) % _capacity;
  }
 
  size_t free_read_space() const noexcept { return _capacity - _free_write_space; }
 
  // This is a template to allow const and non-const (to be able to move)
  template <typename T>
  void write_generic(T* elements, size_t n) {
    std::unique_lock<std::mutex> lock(_mutex);
    LANE_REGISTER_DEBUG_INFO;
 
    if (_status == status_normal) {
      size_t write_size = _free_write_space > n ? n : _free_write_space;
      immediate_write(elements, write_size);
      n -= write_size;
 
      while (n != 0 && _status == status_normal) {
        elements += write_size;
 
        do {
          LANE_REGISTER_DEBUG_WRITE_WAIT;
          _writing_possible_condition.wait(lock);
        } while (_free_write_space == 0 && _status == status_normal);
 
        write_size = _free_write_space > n ? n : _free_write_space;
        immediate_write(elements, write_size);
        n -= write_size;
      }
    }
  }
 
  // This is a template to allow const and non-const (to be able to move)
  template <typename T>
  void immediate_write(T* elements, size_t n) noexcept {
    // Split the writing in two ranges if needed. The first range fits in
    // [_write_position, _capacity), the second range in [0, end). By doing
    // so, we only have to calculate the modulo in the write position once.
    if (n > 0) {
      size_t nPart;
      if (_write_position + n > _capacity) {
        nPart = _capacity - _write_position;
      } else {
        nPart = n;
      }
      for (size_t i = 0; i < nPart; ++i, ++_write_position) {
        _buffer[_write_position] = std::move(elements[i]);
      }
 
      _write_position = _write_position % _capacity;
 
      for (size_t i = nPart; i < n; ++i, ++_write_position) {
        _buffer[_write_position] = std::move(elements[i]);
      }
 
      _free_write_space -= n;
 
      // Now that there is less free write space, there is more free read
      // space and thus readers may continue.
      _reading_possible_condition.notify_all();
    }
  }
 
  void immediate_read(value_type* elements, size_t n) noexcept {
    // As with write, split in two ranges if needed. The first range fits in
    // [read_position(), _capacity), the second range in [0, end).
    if (n > 0) {
      size_t nPart;
      size_t position = read_position();
      if (position + n > _capacity) {
        nPart = _capacity - position;
      } else {
        nPart = n;
      }
      for (size_t i = 0; i < nPart; ++i, ++position) {
        elements[i] = std::move(_buffer[position]);
      }
 
      position = position % _capacity;
 
      for (size_t i = nPart; i < n; ++i, ++position) {
        elements[i] = std::move(_buffer[position]);
      }
 
      _free_write_space += n;
 
      // Now that there is more free write space, writers can possibly continue.
      _writing_possible_condition.notify_all();
    }
  }
 
  void immediate_discard(size_t n) noexcept {
    if (n > 0) {
      _free_write_space += n;
 
      // Now that there is more free write space, writers can possibly continue.
      _writing_possible_condition.notify_all();
    }
  }
 
#ifdef LANE_DEBUG_MODE
  void registerDebugInfo() noexcept {
    _debugSummedSize += _capacity - _free_write_space;
    _debugMeasureCount++;
  }
  void registerDebugReadWait() noexcept { ++_debugReadWaitCount; }
  void registerDebugWriteWait() noexcept { ++_debugWriteWaitCount; }
  void reportDebugInfo() {
    if (!_debugName.empty()) {
      std::stringstream str;
      str << "*** Debug report for the following Lane: ***\n"
          << "\"" << _debugName << "\"\n"
          << "Capacity: " << _capacity << '\n'
          << "Total read/write ops: " << _debugMeasureCount << '\n'
          << "Average size of buffer, measured per read/write op.: "
          << round(double(_debugSummedSize) * 100.0 / _debugMeasureCount) / 100.0 << '\n'
          << "Number of wait events during reading: " << _debugReadWaitCount << '\n'
          << "Number of wait events during writing: " << _debugWriteWaitCount << '\n';
      std::cout << str.str();
    }
  }
  std::string _debugName;
  size_t _debugSummedSize = 0, _debugMeasureCount = 0, _debugReadWaitCount = 0,
         _debugWriteWaitCount = 0;
#endif
};
 
template <typename Tp>
void swap(aocommon::Lane<Tp>& first, aocommon::Lane<Tp>& second) noexcept {
  first.swap(second);
}
 
}  // namespace casacore::aocommon
 
#endif  // AO_LANE11_H