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 VetsUtil::EasyThreads {
public:

 //
 WaveCodecIO(const MetadataVDC &metadata, int nthreads = 0);

 //
 WaveCodecIO(const string &metafile, int nthreads = 0);

#ifdef  DEAD
 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;


};
};

#endif  // _WaveCodeIO_h_