WaveCodecIO.h

Go to the documentation of this file.

//
// $Id$
//
#ifndef _WaveCodeIO_h_
#define _WaveCodeIO_h_
 
#include <vapor/VDFIOBase.h>
#include <vapor/SignificanceMap.h>
#include <vapor/Compressor.h>
#include <vapor/EasyThreads.h>
#include <vapor/NCBuf.h>
 
#ifdef PARALLEL
    #include <mpi.h>
#endif
 
namespace VAPoR {
 
//
//
class VDF_API WaveCodecIO : public VDFIOBase, protected Wasp::EasyThreads {
public:
    //
    WaveCodecIO(const MetadataVDC &metadata, int nthreads = 0);
 
    //
    WaveCodecIO(const string &metafile, int nthreads = 0);
 
#ifdef VAPOR3_0_0_ALPHA
    WaveCodecIO(const size_t dim[3], const size_t bs[3], int numTransforms, const vector<size_t> cratios, const string &wname, const string &filebase);
 
    WaveCodecIO(const vector<string> &files);
#endif
 
    virtual ~WaveCodecIO();
 
    virtual int OpenVariableRead(size_t timestep, const char *varname, int reflevel = 0, int lod = 0);
 
    virtual int OpenVariableWrite(size_t timestep, const char *varname, int reflevel = -1 /*ignored*/, int lod = -1);
 
    //
    virtual int CloseVariable();
 
    //
    virtual int BlockReadRegion(const size_t bmin[3], const size_t bmax[3], float *region, bool unblock = true);
 
    //
    virtual int ReadRegion(const size_t min[3], const size_t max[3], float *region);
 
    virtual int ReadRegion(float *region);
 
    //
    virtual int ReadSlice(float *slice);
 
    //
    virtual int BlockWriteRegion(const float *region, const size_t bmin[3], const size_t bmax[3], bool block = true);
 
    virtual int WriteRegion(const float *region, const size_t min[3], const size_t max[3]);
 
    virtual int WriteRegion(const float *region);
 
    virtual int WriteSlice(const float *slice);
 
    virtual void SetBoundaryPadOnOff(bool pad) { _pad = pad; };
 
    //
    const float *GetDataRange() const { return (_dataRange); }
 
    void GetValidRegion(size_t min[3], size_t max[3], int reflevel) const;
 
    //
    virtual int VariableExists(size_t ts, const char *varname, int reflevel = 0, int lod = 0) const;
 
    static size_t GetMaxCRatio(const size_t bs[3], string wavename, string wmode);
 
    //
    virtual int GetNumTransforms() const;
 
    //
    virtual void GetBlockSize(size_t bs[3], int reflevel) const;
#ifdef PARALLEL
    void SetIOComm(MPI_Comm NewIOComm) { _IO_Comm = NewIOComm; };
#endif
    void         SetCollectiveIO(bool newCollectiveIO) { _collectiveIO = newCollectiveIO; };
    friend void *RunBlockReadRegionThread(void *object);
    friend void *RunBlockWriteRegionThread(void *object);
 
private:
#ifdef PARALLEL
    MPI_Comm _IO_Comm;
#endif
    bool   _collectiveIO;
    double _xformMPI;
    double _methodTimer;
    double _methodThreadTimer;
    double _ioMPI;
    //
    // Threaded read object for parallel inverse transforms
    // (data reconstruction)
    //
    class ReadWriteThreadObj {
    public:
        ReadWriteThreadObj(WaveCodecIO *wc, int id, float *region, const size_t bmin_p[3], const size_t bmax_p[3], const size_t bdim_p[3], const size_t dim_p[3], const size_t bs_p[3], bool reblock,
                           bool pad);
        void         BlockReadRegionThread();
        void         BlockWriteRegionThread();
        const float *GetDataRange() const { return (_dataRange); }
 
    private:
        WaveCodecIO * _wc;
        int           _id;        // thread id
        float *       _region;    // destination buffer for read
        const size_t *_bmin_p;
        const size_t *_bmax_p;    // block coordinates of data
        const size_t *_bdim_p;
        const size_t *_dim_p;
        const size_t *_bs_p;    // dimensions of block
        float         _dataRange[2];
        bool          _reblock;
        bool          _pad;
        int           _FetchBlock(size_t bx, size_t by, size_t bz);
        int           _WriteBlock(size_t bx, size_t by, size_t bz);
    };
 
public:
    int  _nthreads;    // num execution threads
    int  getNumThread() { return _nthreads; }
    void EnableBuffering(size_t count[3], size_t divisor, int rank);
 
private:
    int                             _next_block;
    int                             _threadStatus;
    size_t                          _NC_BUF_SIZE;    // buffering disabled by default
    ReadWriteThreadObj **           _rw_thread_objs;
    vector<vector<SignificanceMap>> _sigmapsThread;    // one set for each thread
    vector<size_t>                  _sigmapsizes;      // size of each encoded sig map
    Compressor *                    _compressor3D;     // 3D compressor
    Compressor *                    _compressor2DXY;
    Compressor *                    _compressor2DXZ;
    Compressor *                    _compressor2DYZ;
    Compressor *                    _compressor;    // compressor for currently opened variable
    vector<Compressor *>            _compressorThread3D;
    vector<Compressor *>            _compressorThread2DXY;
    vector<Compressor *>            _compressorThread2DXZ;
    vector<Compressor *>            _compressorThread2DYZ;
    vector<Compressor *>            _compressorThread;    // current compressor threads
    vector<NCBuf *>                 _ncbufs;
 
    VarType_T               _vtype;             // Type (2d, or 3d) of currently opened variable
    VarType_T               _compressorType;    // Type (2d, or 3d) of current _compressor
    int                     _lod;               // compression level of currently opened file
    int                     _reflevel;          // current refinement level
    size_t                  _validRegMin[3];    // min region bounds of current file
    size_t                  _validRegMax[3];    // max region bounds of current file
    bool                    _writeMode;         // true if opened for writes
    bool                    _isOpen;            // true if a file is opened
    size_t                  _timeStep;          // currently opened time step
    string                  _varName;           // Currently opened variable
    vector<string>          _ncpaths;
    vector<int>             _ncids;
    vector<int>             _nc_sig_vars;    // ncdf ids for wave and sig vars
    vector<int>             _nc_wave_vars;
    float *                 _cvector;        // storage for wavelet coefficients
    size_t                  _cvectorsize;    // amount of space allocated to _cvector
    vector<float *>         _cvectorThread;
    unsigned char *         _svector;        // storage for encoded signficance map
    size_t                  _svectorsize;    // amount of space allocated to _svector
    vector<unsigned char *> _svectorThread;
    float *                 _block;    // storage for a block
    vector<float *>         _blockThread;
    float *                 _blockReg;    // more storage
    float                   _dataRange[2];
    vector<size_t>          _ncoeffs;      // num wave coeff. at each compression level
    vector<size_t>          _cratios3D;    // 3D compression ratios
    vector<size_t>          _cratios2D;    // 2D compression ratios
    vector<size_t>          _cratios;      // compression ratios for currently opened file
 
    float *_sliceBuffer;
    size_t _sliceBufferSize;    // size of slice buffer in elements
    int    _sliceCount;         // num slices written
 
    bool _pad;    // Padding enabled?
 
    int _OpenVarWrite(const string &basename);
    int _OpenVarRead(const string &basename);
    int _WaveCodecIO(int nthreads);
    int _SetupCompressor();
 
    void _UnpackCoord(VarType_T vtype, const size_t src[3], size_t dst[3], size_t fill) const;
 
    void _PackCoord(VarType_T vtype, const size_t src[3], size_t dst[3], size_t fill) const;
 
    void _FillPackedCoord(VarType_T vtype, const size_t src[3], size_t dst[3], size_t fill) const;
};
};    // namespace VAPoR
 
#endif    // _WaveCodeIO_h_