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;
  ao::uvector<unsigned char> _packedBlockReadBuffer;
  ao::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