Loading...
Searching...
No Matches
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 aocommon
#endif // AO_LANE11_H
LANE_REGISTER_DEBUG_WRITE_WAIT
#define LANE_REGISTER_DEBUG_WRITE_WAIT
Definition Lane.h:40
LANE_REGISTER_DEBUG_INFO
#define LANE_REGISTER_DEBUG_INFO
Definition Lane.h:39
LANE_REGISTER_DEBUG_READ_WAIT
#define LANE_REGISTER_DEBUG_READ_WAIT
Definition Lane.h:41
LANE_REPORT_DEBUG_INFO
#define LANE_REPORT_DEBUG_INFO
Definition Lane.h:42
casacore::aocommon::Lane
The Lane is an efficient cyclic buffer that is synchronized.
Definition Lane.h:100
casacore::aocommon::Lane::_free_write_space
size_t _free_write_space
Definition Lane.h:441
casacore::aocommon::Lane::_mutex
std::mutex _mutex
Definition Lane.h:445
casacore::aocommon::Lane::_writing_possible_condition
std::condition_variable _writing_possible_condition
Definition Lane.h:447
casacore::aocommon::Lane::immediate_discard
void immediate_discard(size_t n) noexcept
Definition Lane.h:543
casacore::aocommon::Lane::emplace
void emplace(Args &&... args)
Write a single element by constructing it.
Definition Lane.h:236
casacore::aocommon::Lane::Lane
Lane(const Lane< Tp > &source)=delete
casacore::aocommon::Lane::is_end
bool is_end() const
True when write_end() was called.
Definition Lane.h:388
casacore::aocommon::Lane::move_write
void move_write(value_type *elements, size_t n)
Definition Lane.h:285
casacore::aocommon::Lane::~Lane
~Lane()
Destructor.
Definition Lane.h:154
casacore::aocommon::Lane::clear
void clear() noexcept
Clear the contents and reset the state of the Lane.
Definition Lane.h:191
casacore::aocommon::Lane::swap
void swap(Lane< Tp > &other) noexcept
Swap the contents of this Lane with another.
Definition Lane.h:176
casacore::aocommon::Lane::operator=
Lane< Tp > & operator=(Lane< Tp > &&source) noexcept
Move assignment.
Definition Lane.h:167
casacore::aocommon::Lane::write_end
void write_end()
Definition Lane.h:364
casacore::aocommon::Lane::wait_for_empty
void wait_for_empty()
Wait until this Lane is empty.
Definition Lane.h:418
casacore::aocommon::Lane::operator=
Lane< Tp > & operator=(const Lane< Tp > &source)=delete
casacore::aocommon::Lane::resize
void resize(size_t new_capacity)
Change the capacity of the Lane.
Definition Lane.h:405
casacore::aocommon::Lane::write
void write(value_type &&element)
Write a single element by moving it in.
Definition Lane.h:263
casacore::aocommon::Lane::immediate_write
void immediate_write(T *elements, size_t n) noexcept
This is a template to allow const and non-const (to be able to move).
Definition Lane.h:486
casacore::aocommon::Lane::Lane
Lane(Lane< Tp > &&source) noexcept
Move construct a Lane.
Definition Lane.h:142
casacore::aocommon::Lane::capacity
size_t capacity() const noexcept
Definition Lane.h:372
casacore::aocommon::Lane::value_type
Tp value_type
Type of elements stored in the Lane.
Definition Lane.h:106
casacore::aocommon::Lane::size
size_t size() const
Definition Lane.h:374
casacore::aocommon::Lane::write
void write(const value_type *elements, size_t n)
Definition Lane.h:281
casacore::aocommon::Lane::_capacity
size_t _capacity
Definition Lane.h:437
casacore::aocommon::Lane::_status
enum casacore::aocommon::Lane::@165250377254016164334045006074227223211343376364 _status
casacore::aocommon::Lane::discard
size_t discard(size_t n)
This method does the same thing as read(buffer, n) but discards the data.
Definition Lane.h:339
casacore::aocommon::Lane::size_type
std::size_t size_type
Integer type used to store size types.
Definition Lane.h:103
casacore::aocommon::Lane::read_position
size_t read_position() const noexcept
Definition Lane.h:450
casacore::aocommon::Lane::write_generic
void write_generic(T *elements, size_t n)
This is a template to allow const and non-const (to be able to move).
Definition Lane.h:460
casacore::aocommon::Lane::empty
bool empty() const
Definition Lane.h:379
casacore::aocommon::Lane::write
void write(const value_type &element)
Write a single element.
Definition Lane.h:206
casacore::aocommon::Lane::read
size_t read(value_type *destinations, size_t n)
Definition Lane.h:307
casacore::aocommon::Lane::_buffer
Tp * _buffer
Definition Lane.h:435
casacore::aocommon::Lane::_reading_possible_condition
std::condition_variable _reading_possible_condition
Definition Lane.h:448
casacore::aocommon::Lane::Lane
Lane(size_t capacity)
Construct a Lane with the given capacity.
Definition Lane.h:128
casacore::aocommon::Lane::_write_position
size_t _write_position
Definition Lane.h:439
casacore::aocommon::Lane::free_read_space
size_t free_read_space() const noexcept
Definition Lane.h:454
casacore::aocommon::Lane::is_end_and_empty
bool is_end_and_empty() const
True when write_end() and the lane does not contain items.
Definition Lane.h:396
casacore::aocommon::Lane::status_normal
@ status_normal
Definition Lane.h:443
casacore::aocommon::Lane::status_end
@ status_end
Definition Lane.h:443
casacore::aocommon::Lane::immediate_read
void immediate_read(value_type *elements, size_t n) noexcept
Definition Lane.h:515
casacore::aocommon::Lane::read
bool read(value_type &destination)
Definition Lane.h:289
casacore::aocommon::Lane::Lane
Lane() noexcept
Construct a Lane with zero elements.
Definition Lane.h:116
first
struct Node * first
Definition malloc.h:328
casacore::aocommon
Definition Lane.h:27
casacore::aocommon::swap
void swap(aocommon::Lane< Tp > &first, aocommon::Lane< Tp > &second) noexcept
Definition Lane.h:584
casacore::round
LatticeExprNode round(const LatticeExprNode &expr)