Buffer.h

   1 #ifndef BUFFER_H_
   2 #define BUFFER_H_
   3
   4 #include<cstdio>
   5 #include<fstream>
   6 #include<pthread.h>
   7
   8 #include "my_assert.h"
   9
  10 typedef unsigned long long bufsize_type;
  11 const int FLOATSIZE = sizeof(float);
  12
  13 class Buffer {
  14 public:
  15         Buffer(int nMB, int nSamples, int cvlen, const char* tmpF) {
  16                 cpos = 0;
  17                 size = bufsize_type(nMB) * 1024 * 1024 / FLOATSIZE / cvlen;
  18                 if (size > (bufsize_type)nSamples) size = nSamples;
  19                 general_assert(size > 0, "Memory allocated for credibility intervals is not enough!");
  20                 size *= cvlen;
  21
  22                 buffer = new float[size];
  23                 ftmpOut.open(tmpF, std::ios::binary);
  24                 pthread_mutex_init(&lock, NULL);
  25
  26                 fr = to = 0;
  27                 this->nSamples = nSamples;
  28                 this->cvlen = cvlen;
  29         }
  30
  31         ~Buffer() {
  32                 if (fr < to) flushToTempFile();
  33
  34                 delete[] buffer;
  35                 pthread_mutex_destroy(&lock);
  36                 ftmpOut.close();
  37         }
  38
  39         void write(int n, float **vecs) {
  40                 pthread_assert(pthread_mutex_lock(&lock), "pthread_mutex_lock", "Error occurred while acquiring the lock!");
  41                 for (int i = 0; i < n; i++) {
  42                         if (size - cpos < bufsize_type(cvlen)) flushToTempFile();
  43                         memcpy(buffer + cpos, vecs[i], FLOATSIZE * cvlen);
  44                         cpos += cvlen;
  45                         ++to;
  46                 }
  47                 pthread_assert(pthread_mutex_unlock(&lock), "pthread_mutex_unlock", "Error occurred while releasing the lock!");
  48         }
  49
  50 private:
  51         bufsize_type size, cpos; // cpos : current position
  52
  53         float *buffer;
  54         std::ofstream ftmpOut;
  55         pthread_mutex_t lock;
  56
  57         int fr, to; // each flush, sample fr .. to - 1
  58         int nSamples, cvlen;
  59
  60         void flushToTempFile() {
  61                 std::streampos gap1 = std::streampos(fr) * FLOATSIZE;
  62                 std::streampos gap2 = std::streampos(nSamples - to) * FLOATSIZE;
  63                 float *p = NULL;
  64
  65                 ftmpOut.seekp(0, std::ios::beg);
  66                 for (int i = 0; i < cvlen; i++) {
  67                         p = buffer + i;
  68                         ftmpOut.seekp(gap1, std::ios::cur);
  69                         for (int j = fr; j < to; j++) {
  70                                 ftmpOut.write((char*)p, FLOATSIZE);
  71                                 p += cvlen;
  72                         }
  73                         ftmpOut.seekp(gap2, std::ios::cur);
  74                 }
  75
  76                 cpos = 0;
  77                 fr = to;
  78         }
  79 };
  80
  81 #endif /* BUFFER_H_ */