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