Loading...
Searching...
No Matches
threadeddyscocolumn.h
Go to the documentation of this file.
#ifndef DYSCO_THREADED_DYSCO_COLUMN_H
#define DYSCO_THREADED_DYSCO_COLUMN_H
#include <casacore/tables/DataMan/DataManError.h>
#include <casacore/casa/Arrays/IPosition.h>
#include <casacore/tables/Tables/ScalarColumn.h>
#include <condition_variable>
#include <cstdint>
#include <map>
#include <memory>
#include <mutex>
#include <random>
#include "dyscostmancol.h"
#include "serializable.h"
#include "stochasticencoder.h"
#include "threadgroup.h"
#include "timeblockbuffer.h"
namespace dyscostman {
class DyscoStMan;
template <typename DataType>
class ThreadedDyscoColumn : public DyscoStManColumn {
public:
typedef DataType data_t;
ThreadedDyscoColumn(DyscoStMan *parent, int dtype);
ThreadedDyscoColumn(const ThreadedDyscoColumn &source) = delete;
void operator=(const ThreadedDyscoColumn &source) = delete;
virtual ~ThreadedDyscoColumn();
virtual void setShapeColumn(const casacore::IPosition &shape) override;
virtual casacore::IPosition shape(casacore::rownr_t /*rownr*/) override {
return _shape;
}
virtual void getArrayV(
casacore::rownr_t rowNr,
casacore::ArrayBase &dataPtr) override {
return DyscoStManColumn::getArrayV(rowNr, dataPtr);
}
virtual void putArrayV(
casacore::rownr_t rowNr,
const casacore::ArrayBase &dataPtr) override {
return DyscoStManColumn::putArrayV(rowNr, dataPtr);
}
virtual void Prepare(DyscoDistribution distribution,
Normalization normalization, double studentsTNu,
double distributionTruncation) override;
virtual void InitializeAfterNRowsPerBlockIsKnown() override;
void SetBitsPerSymbol(unsigned bitsPerSymbol) {
_bitsPerSymbol = bitsPerSymbol;
}
virtual size_t CalculateBlockSize(size_t nRowsInBlock,
size_t nAntennae) const final override;
virtual size_t ExtraHeaderSize() const override { return Header::Size(); }
virtual void SerializeExtraHeader(std::ostream &stream) const final override;
virtual void UnserializeExtraHeader(std::istream &stream) final override;
protected:
class ThreadDataBase {
public:
virtual ~ThreadDataBase(){};
};
typedef typename TimeBlockBuffer<data_t>::symbol_t symbol_t;
virtual void initializeDecode(TimeBlockBuffer<data_t> *buffer,
const float *metaBuffer, size_t nRow,
size_t nAntennae) = 0;
virtual void decode(TimeBlockBuffer<data_t> *buffer, const symbol_t *data,
size_t blockRow, size_t a1, size_t a2) = 0;
virtual std::unique_ptr<ThreadDataBase> initializeEncodeThread() = 0;
virtual void encode(ThreadDataBase *threadData,
TimeBlockBuffer<data_t> *buffer, float *metaBuffer,
symbol_t *symbolBuffer, size_t nAntennae) = 0;
virtual size_t metaDataFloatCount(size_t nRow, size_t nPolarizations,
size_t nChannels,
size_t nAntennae) const = 0;
virtual size_t symbolCount(size_t nRowsInBlock, size_t nPolarizations,
size_t nChannels) const = 0;
virtual void shutdown() override final;
virtual size_t defaultThreadCount() const;
size_t getBitsPerSymbol() const { return _bitsPerSymbol; }
const casacore::IPosition &shape() const { return _shape; }
private:
struct CacheItem {
CacheItem(std::unique_ptr<TimeBlockBuffer<data_t>> &&encoder_)
: encoder(std::move(encoder_)), isBeingWritten(false) {}
std::unique_ptr<TimeBlockBuffer<data_t>> encoder;
bool isBeingWritten;
};
struct EncodingThreadFunctor {
void operator()();
ThreadedDyscoColumn *parent;
};
struct Header : public Serializable {
uint32_t blockSize;
uint32_t antennaCount;
static uint32_t Size() { return 8; }
virtual void Serialize(std::ostream &stream) const override {
SerializeToUInt32(stream, blockSize);
SerializeToUInt32(stream, antennaCount);
}
virtual void Unserialize(std::istream &stream) override {
blockSize = UnserializeUInt32(stream);
antennaCount = UnserializeUInt32(stream);
}
};
typedef std::map<size_t, CacheItem *> cache_t;
void getValues(casacore::rownr_t rowNr, casacore::Array<data_t> *dataPtr);
void putValues(casacore::rownr_t rowNr, const casacore::Array<data_t> *dataPtr);
void stopThreads();
void encodeAndWrite(size_t blockIndex, const CacheItem &item,
unsigned char *packedSymbolBuffer,
unsigned int *unpackedSymbolBuffer,
ThreadDataBase *threadUserData);
bool isWriteItemAvailable(typename cache_t::iterator &i);
void loadBlock(size_t blockIndex);
void storeBlock();
size_t maxCacheSize() const {
return ThreadedDyscoColumn::defaultThreadCount() * 12 / 10 + 1;
}
unsigned _bitsPerSymbol;
casacore::IPosition _shape;
std::unique_ptr<casacore::ScalarColumn<int>> _ant1Col, _ant2Col, _fieldCol,
_dataDescIdCol;
std::unique_ptr<casacore::ScalarColumn<double>> _timeCol;
double _lastWrittenTime;
int _lastWrittenField, _lastWrittenDataDescId;
aocommon::UVector<unsigned char> _packedBlockReadBuffer;
aocommon::UVector<unsigned int> _unpackedSymbolReadBuffer;
cache_t _cache;
bool _stopThreads;
std::mutex _mutex;
threadgroup _threadGroup;
std::condition_variable _cacheChangedCondition;
size_t _currentBlock;
bool _isCurrentBlockChanged;
size_t _blockSize;
size_t _antennaCount;
std::unique_ptr<TimeBlockBuffer<data_t>> _timeBlockBuffer;
};
template <>
inline void ThreadedDyscoColumn<std::complex<float>>::getArrayV(
casacore::rownr_t rowNr, casacore::ArrayBase &dataPtr) {
getValues(rowNr, static_cast<casacore::Array<std::complex<float>>*>(&dataPtr));
}
template <>
inline void ThreadedDyscoColumn<std::complex<float>>::putArrayV(
casacore::rownr_t rowNr, const casacore::ArrayBase &dataPtr) {
putValues(rowNr, static_cast<const casacore::Array<std::complex<float>>*>(&dataPtr));
}
template <>
inline void ThreadedDyscoColumn<float>::getArrayV(
casacore::rownr_t rowNr, casacore::ArrayBase &dataPtr) {
getValues(rowNr, static_cast<casacore::Array<float>*>(&dataPtr));
}
template <>
inline void ThreadedDyscoColumn<float>::putArrayV(
casacore::rownr_t rowNr, const casacore::ArrayBase &dataPtr) {
putValues(rowNr, static_cast<const casacore::Array<float>*>(&dataPtr));
}
extern template class ThreadedDyscoColumn<std::complex<float>>;
extern template class ThreadedDyscoColumn<float>;
} // namespace dyscostman
#endif
Serializable
Definition serializable.h:9
Serializable::UnserializeUInt32
static uint32_t UnserializeUInt32(std::istream &stream)
Definition serializable.h:76
Serializable::SerializeToUInt32
static void SerializeToUInt32(std::ostream &stream, T value)
Definition serializable.h:22
TimeBlockBuffer
Definition timeblockbuffer.h:10
TimeBlockBuffer::symbol_t
unsigned symbol_t
Definition timeblockbuffer.h:12
aocommon::UVector
A container similar to std::vector, but one that allows construction without initializing its element...
Definition uvector.h:77
casacore::ArrayBase
Non-templated base class for templated Array class.
Definition ArrayBase.h:71
casacore::Array
Definition Array.h:155
casacore::DataManagerColumn::putArrayV
virtual void putArrayV(rownr_t rownr, const ArrayBase &data)
Put the array value into the given row.
casacore::DataManagerColumn::getArrayV
virtual void getArrayV(rownr_t rownr, ArrayBase &dataPtr)
Get the array value in the given row.
casacore::IPosition
Definition IPosition.h:117
casacore::StManColumnBase::dtype
DataType dtype() const
Definition StManColumnBase.h:90
dyscostman::DyscoStManColumn::DyscoStManColumn
DyscoStManColumn(DyscoStMan *parent, int dtype)
Constructor, to be overloaded by subclass.
Definition dyscostmancol.h:29
dyscostman::DyscoStManColumn::nAntennae
size_t nAntennae() const
Definition dyscostmancol.h:150
dyscostman::DyscoStManColumn::nRowsInBlock
size_t nRowsInBlock() const
Definition dyscostmancol.h:146
dyscostman::DyscoStMan
The main class for the Dysco storage manager.
Definition dyscostman.h:46
dyscostman::ThreadedDyscoColumn::ThreadDataBase
Definition threadeddyscocolumn.h:110
dyscostman::ThreadedDyscoColumn::ThreadDataBase::~ThreadDataBase
virtual ~ThreadDataBase()
Definition threadeddyscocolumn.h:112
dyscostman::ThreadedDyscoColumn
A column for storing compressed values in a threaded way, tailored for the data and weight columns th...
Definition threadeddyscocolumn.h:32
dyscostman::ThreadedDyscoColumn::_lastWrittenTime
double _lastWrittenTime
Definition threadeddyscocolumn.h:197
dyscostman::ThreadedDyscoColumn::InitializeAfterNRowsPerBlockIsKnown
virtual void InitializeAfterNRowsPerBlockIsKnown() override
Prepare this column for reading/writing.
dyscostman::ThreadedDyscoColumn::symbol_t
TimeBlockBuffer< data_t >::symbol_t symbol_t
Definition threadeddyscocolumn.h:115
dyscostman::ThreadedDyscoColumn::_timeBlockBuffer
std::unique_ptr< TimeBlockBuffer< data_t > > _timeBlockBuffer
Definition threadeddyscocolumn.h:211
dyscostman::ThreadedDyscoColumn::_packedBlockReadBuffer
aocommon::UVector< unsigned char > _packedBlockReadBuffer
Definition threadeddyscocolumn.h:199
dyscostman::ThreadedDyscoColumn::cache_t
std::map< size_t, CacheItem * > cache_t
Definition threadeddyscocolumn.h:175
dyscostman::ThreadedDyscoColumn::~ThreadedDyscoColumn
virtual ~ThreadedDyscoColumn()
Destructor.
dyscostman::ThreadedDyscoColumn::stopThreads
void stopThreads()
dyscostman::ThreadedDyscoColumn::_cacheChangedCondition
std::condition_variable _cacheChangedCondition
Definition threadeddyscocolumn.h:205
dyscostman::ThreadedDyscoColumn::encode
virtual void encode(ThreadDataBase *threadData, TimeBlockBuffer< data_t > *buffer, float *metaBuffer, symbol_t *symbolBuffer, size_t nAntennae)=0
dyscostman::ThreadedDyscoColumn::setShapeColumn
virtual void setShapeColumn(const casacore::IPosition &shape) override
Set the dimensions of values in this column.
dyscostman::ThreadedDyscoColumn::isWriteItemAvailable
bool isWriteItemAvailable(typename cache_t::iterator &i)
dyscostman::ThreadedDyscoColumn::_blockSize
size_t _blockSize
Definition threadeddyscocolumn.h:208
dyscostman::ThreadedDyscoColumn::getValues
void getValues(casacore::rownr_t rowNr, casacore::Array< data_t > *dataPtr)
dyscostman::ThreadedDyscoColumn::_fieldCol
std::unique_ptr< casacore::ScalarColumn< int > > _fieldCol
Definition threadeddyscocolumn.h:194
dyscostman::ThreadedDyscoColumn::_stopThreads
bool _stopThreads
Definition threadeddyscocolumn.h:202
dyscostman::ThreadedDyscoColumn< std::complex< float > >::putArrayV
void putArrayV(casacore::rownr_t rowNr, const casacore::ArrayBase &dataPtr)
Definition threadeddyscocolumn.h:220
dyscostman::ThreadedDyscoColumn::shape
const casacore::IPosition & shape() const
Definition threadeddyscocolumn.h:143
dyscostman::ThreadedDyscoColumn::_dataDescIdCol
std::unique_ptr< casacore::ScalarColumn< int > > _dataDescIdCol
Definition threadeddyscocolumn.h:195
dyscostman::ThreadedDyscoColumn::initializeEncodeThread
virtual std::unique_ptr< ThreadDataBase > initializeEncodeThread()=0
dyscostman::ThreadedDyscoColumn::ThreadedDyscoColumn
ThreadedDyscoColumn(DyscoStMan *parent, int dtype)
Create a new column.
dyscostman::ThreadedDyscoColumn::_currentBlock
size_t _currentBlock
Definition threadeddyscocolumn.h:206
dyscostman::ThreadedDyscoColumn::Prepare
virtual void Prepare(DyscoDistribution distribution, Normalization normalization, double studentsTNu, double distributionTruncation) override
dyscostman::ThreadedDyscoColumn::CalculateBlockSize
virtual size_t CalculateBlockSize(size_t nRowsInBlock, size_t nAntennae) const final override
dyscostman::ThreadedDyscoColumn::data_t
DataType data_t
Definition threadeddyscocolumn.h:34
dyscostman::ThreadedDyscoColumn::symbolCount
virtual size_t symbolCount(size_t nRowsInBlock, size_t nPolarizations, size_t nChannels) const =0
dyscostman::ThreadedDyscoColumn::ThreadedDyscoColumn
ThreadedDyscoColumn(const ThreadedDyscoColumn &source)=delete
dyscostman::ThreadedDyscoColumn::SetBitsPerSymbol
void SetBitsPerSymbol(unsigned bitsPerSymbol)
Set the bits per symbol.
Definition threadeddyscocolumn.h:96
dyscostman::ThreadedDyscoColumn::storeBlock
void storeBlock()
dyscostman::ThreadedDyscoColumn::decode
virtual void decode(TimeBlockBuffer< data_t > *buffer, const symbol_t *data, size_t blockRow, size_t a1, size_t a2)=0
dyscostman::ThreadedDyscoColumn::UnserializeExtraHeader
virtual void UnserializeExtraHeader(std::istream &stream) final override
dyscostman::ThreadedDyscoColumn::_threadGroup
threadgroup _threadGroup
Definition threadeddyscocolumn.h:204
dyscostman::ThreadedDyscoColumn::getBitsPerSymbol
size_t getBitsPerSymbol() const
Definition threadeddyscocolumn.h:141
dyscostman::ThreadedDyscoColumn::operator=
void operator=(const ThreadedDyscoColumn &source)=delete
dyscostman::ThreadedDyscoColumn::putValues
void putValues(casacore::rownr_t rowNr, const casacore::Array< data_t > *dataPtr)
dyscostman::ThreadedDyscoColumn::initializeDecode
virtual void initializeDecode(TimeBlockBuffer< data_t > *buffer, const float *metaBuffer, size_t nRow, size_t nAntennae)=0
dyscostman::ThreadedDyscoColumn::_cache
cache_t _cache
Definition threadeddyscocolumn.h:201
dyscostman::ThreadedDyscoColumn::_ant2Col
std::unique_ptr< casacore::ScalarColumn< int > > _ant2Col
Definition threadeddyscocolumn.h:194
dyscostman::ThreadedDyscoColumn::_unpackedSymbolReadBuffer
aocommon::UVector< unsigned int > _unpackedSymbolReadBuffer
Definition threadeddyscocolumn.h:200
dyscostman::ThreadedDyscoColumn::defaultThreadCount
virtual size_t defaultThreadCount() const
dyscostman::ThreadedDyscoColumn::_ant1Col
std::unique_ptr< casacore::ScalarColumn< int > > _ant1Col
Definition threadeddyscocolumn.h:194
dyscostman::ThreadedDyscoColumn::_lastWrittenField
int _lastWrittenField
Definition threadeddyscocolumn.h:198
dyscostman::ThreadedDyscoColumn::shape
virtual casacore::IPosition shape(casacore::rownr_t) override
Get the dimensions of the values in a particular row.
Definition threadeddyscocolumn.h:54
dyscostman::ThreadedDyscoColumn::shutdown
virtual void shutdown() override final
To be called before destructing the class.
dyscostman::ThreadedDyscoColumn::ExtraHeaderSize
virtual size_t ExtraHeaderSize() const override
Get number of bytes needed for column header of this column.
Definition threadeddyscocolumn.h:103
dyscostman::ThreadedDyscoColumn::_mutex
std::mutex _mutex
Definition threadeddyscocolumn.h:203
dyscostman::ThreadedDyscoColumn::loadBlock
void loadBlock(size_t blockIndex)
dyscostman::ThreadedDyscoColumn::_lastWrittenDataDescId
int _lastWrittenDataDescId
Definition threadeddyscocolumn.h:198
dyscostman::ThreadedDyscoColumn::maxCacheSize
size_t maxCacheSize() const
Definition threadeddyscocolumn.h:188
dyscostman::ThreadedDyscoColumn::metaDataFloatCount
virtual size_t metaDataFloatCount(size_t nRow, size_t nPolarizations, size_t nChannels, size_t nAntennae) const =0
dyscostman::ThreadedDyscoColumn::putArrayV
virtual void putArrayV(casacore::rownr_t rowNr, const casacore::ArrayBase &dataPtr) override
Write values into a particular row.
Definition threadeddyscocolumn.h:77
dyscostman::ThreadedDyscoColumn::_timeCol
std::unique_ptr< casacore::ScalarColumn< double > > _timeCol
Definition threadeddyscocolumn.h:196
dyscostman::ThreadedDyscoColumn< std::complex< float > >::getArrayV
void getArrayV(casacore::rownr_t rowNr, casacore::ArrayBase &dataPtr)
Definition threadeddyscocolumn.h:215
dyscostman::ThreadedDyscoColumn::_antennaCount
size_t _antennaCount
Definition threadeddyscocolumn.h:209
dyscostman::ThreadedDyscoColumn::_isCurrentBlockChanged
bool _isCurrentBlockChanged
Definition threadeddyscocolumn.h:207
dyscostman::ThreadedDyscoColumn::_bitsPerSymbol
unsigned _bitsPerSymbol
Definition threadeddyscocolumn.h:192
dyscostman::ThreadedDyscoColumn::getArrayV
virtual void getArrayV(casacore::rownr_t rowNr, casacore::ArrayBase &dataPtr) override
Read the values for a particular row.
Definition threadeddyscocolumn.h:64
dyscostman::ThreadedDyscoColumn::encodeAndWrite
void encodeAndWrite(size_t blockIndex, const CacheItem &item, unsigned char *packedSymbolBuffer, unsigned int *unpackedSymbolBuffer, ThreadDataBase *threadUserData)
dyscostman::ThreadedDyscoColumn::_shape
casacore::IPosition _shape
Definition threadeddyscocolumn.h:193
dyscostman::ThreadedDyscoColumn::SerializeExtraHeader
virtual void SerializeExtraHeader(std::ostream &stream) const final override
dyscostman::threadgroup
Group of threads.
Definition threadgroup.h:13
dyscostmancol.h
casacore::rownr_t
uInt64 rownr_t
Define the type of a row number in a table.
Definition aipsxtype.h:44
dyscostman
Definition bytepacker.h:7
dyscostman::Normalization
Normalization
Definition dysconormalization.h:5
dyscostman::DyscoDistribution
DyscoDistribution
Definition dyscodistribution.h:5
std
Define real & complex conjugation for non-complex types and put comparisons into std namespace.
Definition Complex.h:350
serializable.h
stochasticencoder.h
dyscostman::ThreadedDyscoColumn::CacheItem
Definition threadeddyscocolumn.h:146
dyscostman::ThreadedDyscoColumn::CacheItem::isBeingWritten
bool isBeingWritten
Definition threadeddyscocolumn.h:151
dyscostman::ThreadedDyscoColumn::CacheItem::encoder
std::unique_ptr< TimeBlockBuffer< data_t > > encoder
Definition threadeddyscocolumn.h:150
dyscostman::ThreadedDyscoColumn::CacheItem::CacheItem
CacheItem(std::unique_ptr< TimeBlockBuffer< data_t > > &&encoder_)
Definition threadeddyscocolumn.h:147
dyscostman::ThreadedDyscoColumn::EncodingThreadFunctor
Definition threadeddyscocolumn.h:154
dyscostman::ThreadedDyscoColumn::EncodingThreadFunctor::operator()
void operator()()
dyscostman::ThreadedDyscoColumn::EncodingThreadFunctor::parent
ThreadedDyscoColumn * parent
Definition threadeddyscocolumn.h:156
dyscostman::ThreadedDyscoColumn::Header
Definition threadeddyscocolumn.h:158
dyscostman::ThreadedDyscoColumn::Header::Serialize
virtual void Serialize(std::ostream &stream) const override
Definition threadeddyscocolumn.h:164
dyscostman::ThreadedDyscoColumn::Header::Size
static uint32_t Size()
Definition threadeddyscocolumn.h:162
dyscostman::ThreadedDyscoColumn::Header::Unserialize
virtual void Unserialize(std::istream &stream) override
Definition threadeddyscocolumn.h:169
dyscostman::ThreadedDyscoColumn::Header::blockSize
uint32_t blockSize
Definition threadeddyscocolumn.h:159
dyscostman::ThreadedDyscoColumn::Header::antennaCount
uint32_t antennaCount
Definition threadeddyscocolumn.h:160
threadgroup.h
timeblockbuffer.h