HsFileSchema Class Reference

Inheritance diagram for HsFileSchema:

Collaboration diagram for HsFileSchema:

Public Member Functions
	HsFileSchema ()
	~HsFileSchema ()
void	remove_inactive_columns ()
void	print (bool print_tags=true) const
int	flush_buffers ()
int	close ()
int	write_event (const time_t t, const char *data, const size_t data_size)
int	read_last_written (const time_t timestamp, const int debug, time_t *last_written)
int	read_data (const time_t start_time, const time_t end_time, const int num_var, const std::vector< int > &var_schema_index, const int var_index[], const int debug, std::vector< time_t > &last_time, MidasHistoryBufferInterface *buffer[])
Public Member Functions inherited from HsSchema
	HsSchema ()
virtual	~HsSchema ()
virtual int	match_event_var (const char *event_name, const char *var_name, const int var_index)

Public Attributes
std::string	fFileName
size_t	fRecordSize = 0
off64_t	fDataOffset = 0
size_t	fLastSize = 0
int	fWriterFd = -1
size_t	fRecordBufferSize = 0
char *	fRecordBuffer = NULL
off64_t	fFileSizeInitial = 0
off64_t	fFileSize = 0
Public Attributes inherited from HsSchema
std::string	fEventName
time_t	fTimeFrom = 0
time_t	fTimeTo = 0
std::vector< HsSchemaEntry >	fVariables
std::vector< int >	fOffsets
size_t	fNumBytes = 0
int	fCountWriteUndersize = 0
int	fCountWriteOversize = 0
size_t	fWriteMaxSize = 0
size_t	fWriteMinSize = 0
bool	fDisabled = true

Detailed Description

Definition at line 757 of file history_schema.cxx.

Constructor & Destructor Documentation

HsFileSchema()

HsFileSchema::HsFileSchema ( )

inline

Definition at line 775 of file history_schema.cxx.

   {
      // empty
   }

~HsFileSchema()

HsFileSchema::~HsFileSchema ( )

inline

Definition at line 780 of file history_schema.cxx.

   {
      //printf("HsFileSchema::dtor %p!\n", this);
      close();
      fRecordSize = 0;
      fDataOffset = 0;
      fLastSize = 0;
      fWriterFd = -1;
      if (fRecordBuffer) {
         free(fRecordBuffer);
         fRecordBuffer = NULL;
      }
      fRecordBufferSize = 0;
   }

Here is the call graph for this function:

Member Function Documentation

close()

int HsFileSchema::close ( )

virtual

Implements HsSchema.

Definition at line 2606 of file history_schema.cxx.

{
   if (fWriterFd >= 0) {
      ::close(fWriterFd);
      fWriterFd = -1;
   }
   return HS_SUCCESS;
}

Here is the call graph for this function:

Here is the caller graph for this function:

flush_buffers()

int HsFileSchema::flush_buffers ( )

inlinevirtual

Implements HsSchema.

Definition at line 797 of file history_schema.cxx.

{ return HS_SUCCESS; };

print()

void HsFileSchema::print ( bool  print_tags = true ) const

virtual

Reimplemented from HsSchema.

Definition at line 838 of file history_schema.cxx.

{
   size_t nv = this->fVariables.size();
   printf("event [%s], file_name [%s], time %s..%s, %zu variables, %zu bytes, dat_offset %jd, record_size %zu\n", this->fEventName.c_str(), this->fFileName.c_str(), TimeToString(this->fTimeFrom).c_str(), TimeToString(this->fTimeTo).c_str(), nv, fNumBytes, (intmax_t)fDataOffset, fRecordSize);
   if (print_tags) {
      for (size_t j=0; j<nv; j++)
         printf("  %zu: name [%s], type [%s] tid %d, n_data %d, n_bytes %d, offset %d\n", j, this->fVariables[j].name.c_str(), rpc_tid_name(this->fVariables[j].type), this->fVariables[j].type, this->fVariables[j].n_data, this->fVariables[j].n_bytes, this->fOffsets[j]);
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

read_data()

int HsFileSchema::read_data ( const time_t  start_time, const time_t  end_time, const int  num_var, const std::vector< int > &  var_schema_index, const int  var_index[], const int  debug, std::vector< time_t > &  last_time, MidasHistoryBufferInterface *  buffer[]  )

virtual

Implements HsSchema.

Definition at line 2877 of file history_schema.cxx.

{
   HsFileSchema* s = this;
 
   if (debug)
      printf("FileHistory::read_data: file %s, schema time %s..%s, read time %s..%s, %d vars\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(start_time).c_str(), TimeToString(end_time).c_str(), num_var);
 
   //if (1) {
   //   printf("Last time: ");
   //   for (int i=0; i<num_var; i++) {
   //      printf(" %s", TimeToString(last_time[i]).c_str());
   //   }
   //   printf("\n");
   //}
 
   if (debug) {
      printf("FileHistory::read_data: file %s map", s->fFileName.c_str());
      for (size_t i=0; i<var_schema_index.size(); i++) {
         printf(" %2d", var_schema_index[i]);
      }
      printf("\n");
   }
 
   int fd = ::open(s->fFileName.c_str(), O_RDONLY);
   if (fd < 0) {
      cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', open() errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   off64_t file_size = ::lseek64(fd, 0, SEEK_END);
 
   if (file_size < 0) {
      cm_msg(MERROR, "FileHistory::read_data", "Cannot read file size of \'%s\', lseek64(SEEK_END) errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
      ::close(fd);
      return HS_FILE_ERROR;
   }
 
   if (file_size < s->fDataOffset) {
      // empty file
      ::close(fd);
      return HS_SUCCESS;
   }
 
   off64_t nrec = (file_size - s->fDataOffset)/s->fRecordSize;
 
   //printf("read_data: nrec %jd, file_size %jd, offset %jd, record size %zu\n", (intmax_t)nrec, (intmax_t)file_size, s->fDataOffset, s->fRecordSize);
 
   if (nrec < 0)
      nrec = 0;
 
   if (nrec < 1) {
      ::close(fd);
      return HS_SUCCESS;
   }
 
   off64_t iunused = 0;
   time_t tunused = 0;
   off64_t irec = 0;
   time_t trec   = 0;
   time_t tstart = 0;
   time_t tend   = 0;
 
   int istatus = FindTime(s->fFileName.c_str(), fd, s->fDataOffset, s->fRecordSize, nrec, start_time, &iunused, &tunused, &irec, &trec, &tstart, &tend, 0*debug);
 
   if (istatus != HS_SUCCESS) {
      ::close(fd);
      return HS_FILE_ERROR;
   }
 
   if (debug) {
      printf("FindTime %d, nrec %jd, (%jd, %s) (%jd, %s), tstart %s, tend %s, want %s\n", istatus, (intmax_t)nrec, (intmax_t)iunused, TimeToString(tunused).c_str(), (intmax_t)irec, TimeToString(trec).c_str(), TimeToString(tstart).c_str(), TimeToString(tend).c_str(), TimeToString(start_time).c_str());
   }
 
   if (irec < 0 || irec >= nrec) {
      // all data in this file is older than start_time
 
      ::close(fd);
 
      if (debug)
         printf("FileHistory::read: file %s, schema time %s..%s, read time %s..%s, file time %s..%s, data in this file is too old\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(start_time).c_str(), TimeToString(end_time).c_str(), TimeToString(tstart).c_str(), TimeToString(tend).c_str());
 
      return HS_SUCCESS;
   }
 
   if (tstart < s->fTimeFrom) {
      // data starts before time declared in schema
 
      ::close(fd);
 
      cm_msg(MERROR, "FileHistory::read_data", "Bad history file \'%s\': timestamp of first data %s is before schema start time %s", s->fFileName.c_str(), TimeToString(tstart).c_str(), TimeToString(s->fTimeFrom).c_str());
 
      return HS_FILE_ERROR;
   }
 
   if (tend && s->fTimeTo && tend > s->fTimeTo) {
      // data ends after time declared in schema (overlaps with next file)
 
      ::close(fd);
 
      cm_msg(MERROR, "FileHistory::read_data", "Bad history file \'%s\': timestamp of last data %s is after schema end time %s", s->fFileName.c_str(), TimeToString(tend).c_str(), TimeToString(s->fTimeTo).c_str());
 
      return HS_FILE_ERROR;
   }
 
   for (int i=0; i<num_var; i++) {
      int si = var_schema_index[i];
      if (si < 0)
         continue;
         
      if (trec < last_time[i]) { // protect against duplicate and non-monotonous data
         ::close(fd);
 
         cm_msg(MERROR, "FileHistory::read_data", "Internal history error at file \'%s\': variable %d data timestamp %s is before last timestamp %s", s->fFileName.c_str(), i, TimeToString(trec).c_str(), TimeToString(last_time[i]).c_str());
         
         return HS_FILE_ERROR;
      }
   }
 
   int count = 0;
 
   off64_t fpos = s->fDataOffset + irec*s->fRecordSize;
   
   off64_t xpos = ::lseek64(fd, fpos, SEEK_SET);
 
   //printf("read_data: lseek64() returned %jd, irec %jd, offset %jd, record size %zu, fpos %jd\n", xpos, irec, s->fDataOffset, s->fRecordSize, fpos);
 
   if (xpos < 0) {
      cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', lseek64(%jd) errno %d (%s)", s->fFileName.c_str(), (intmax_t)fpos, errno, strerror(errno));
      ::close(fd);
      return HS_FILE_ERROR;
   }
 
   char* buf = new char[s->fRecordSize];
 
   off64_t prec = irec;
 
   while (1) {
      ssize_t rd = ::read(fd, buf, s->fRecordSize);
 
      if (rd < 0) {
         cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', read() errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
         break;
      }
 
      if (rd == 0) {
         // EOF
         break;
      }
 
      if ((size_t)rd != s->fRecordSize) {
         cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', short read() returned %zd instead of %zu bytes", s->fFileName.c_str(), rd, s->fRecordSize);
         break;
      }
 
      prec++;
 
      bool past_end_of_last_file = (s->fTimeTo == 0) && (prec > nrec);
 
      time_t t = *(DWORD*)buf;
      
      if (debug > 1)
         printf("FileHistory::read: file %s, schema time %s..%s, read time %s..%s, row time %s\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(start_time).c_str(), TimeToString(end_time).c_str(), TimeToString(t).c_str());
      
      if (t < trec) {
         delete[] buf;
         ::close(fd);
         cm_msg(MERROR, "FileHistory::read_data", "Bad history file \'%s\': record %jd timestamp %s is before start time %s", s->fFileName.c_str(), (intmax_t)(irec + count), TimeToString(t).c_str(), TimeToString(trec).c_str());
         return HS_FILE_ERROR;
      }
      
      if (tend && (t > tend) && !past_end_of_last_file) {
         delete[] buf;
         ::close(fd);
         cm_msg(MERROR, "FileHistory::read_data", "Bad history file \'%s\': record %jd timestamp %s is after last timestamp %s", s->fFileName.c_str(), (intmax_t)(irec + count), TimeToString(t).c_str(), TimeToString(tend).c_str());
         return HS_FILE_ERROR;
      }
 
      if (t > end_time)
         break;
      
      char* data = buf + 4;
      
      for (int i=0; i<num_var; i++) {
         int si = var_schema_index[i];
         if (si < 0)
            continue;
         
         if (t < last_time[i]) { // protect against duplicate and non-monotonous data
            delete[] buf;
            ::close(fd);
 
            cm_msg(MERROR, "FileHistory::read_data", "Bad history file \'%s\': record %jd timestamp %s is before timestamp %s of variable %d", s->fFileName.c_str(), (intmax_t)(irec + count), TimeToString(t).c_str(), TimeToString(last_time[i]).c_str(), i);
 
            return HS_FILE_ERROR;
         }
         
         double v = 0;
         void* ptr = data + s->fOffsets[si];
         
         int ii = var_index[i];
         assert(ii >= 0);
         assert(ii < s->fVariables[si].n_data);
         
         switch (s->fVariables[si].type) {
         default:
            // unknown data type
            v = 0;
            break;
         case TID_BYTE:
            v = ((unsigned char*)ptr)[ii];
            break;
         case TID_SBYTE:
            v = ((signed char *)ptr)[ii];
            break;
         case TID_CHAR:
            v = ((char*)ptr)[ii];
            break;
         case TID_WORD:
            v = ((unsigned short *)ptr)[ii];
            break;
         case TID_SHORT:
            v = ((signed short *)ptr)[ii];
            break;
         case TID_DWORD:
            v = ((unsigned int *)ptr)[ii];
            break;
         case TID_INT:
            v = ((int *)ptr)[ii];
            break;
         case TID_BOOL:
            v = ((unsigned int *)ptr)[ii];
            break;
         case TID_FLOAT:
            v = ((float*)ptr)[ii];
            break;
         case TID_DOUBLE:
            v = ((double*)ptr)[ii];
            break;
         }
         
         buffer[i]->Add(t, v);
         last_time[i] = t;
      }
      count++;
   }
 
   delete[] buf;
 
   ::close(fd);
 
   if (debug) {
      printf("FileHistory::read_data: file %s map", s->fFileName.c_str());
      for (size_t i=0; i<var_schema_index.size(); i++) {
         printf(" %2d", var_schema_index[i]);
      }
      printf(" read %d rows\n", count);
   }
 
   if (debug)
      printf("FileHistory::read: file %s, schema time %s..%s, read time %s..%s, %d vars, read %d rows\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(start_time).c_str(), TimeToString(end_time).c_str(), num_var, count);
 
   return HS_SUCCESS;
}

Here is the call graph for this function:

read_last_written()

int HsFileSchema::read_last_written ( const time_t  timestamp, const int  debug, time_t *  last_written  )

virtual

Implements HsSchema.

Definition at line 2781 of file history_schema.cxx.

{
   int status;
   HsFileSchema* s = this;
 
   if (debug)
      printf("FileHistory::read_last_written: file %s, schema time %s..%s, timestamp %s\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(timestamp).c_str());
 
   int fd = open(s->fFileName.c_str(), O_RDONLY);
   if (fd < 0) {
      cm_msg(MERROR, "FileHistory::read_last_written", "Cannot read \'%s\', open() errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   off64_t file_size = ::lseek64(fd, 0, SEEK_END);
 
   if (file_size < 0) {
      cm_msg(MERROR, "FileHistory::read_last_written", "Cannot read file size of \'%s\', lseek64(SEEK_END) errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   if (file_size < s->fDataOffset) {
      // empty file
      ::close(fd);
      if (last_written)
         *last_written = 0;
      return HS_SUCCESS;
   }
 
   off64_t nrec = (file_size - s->fDataOffset)/s->fRecordSize;
 
   //printf("read_last_written: nrec %jd, file_size %jd, offset %jd, record size %zu\n", (intmax_t)nrec, (intmax_t)file_size, s->fDataOffset, s->fRecordSize);
 
   if (nrec < 0)
      nrec = 0;
 
   if (nrec < 1) {
      ::close(fd);
      if (last_written)
         *last_written = 0;
      return HS_SUCCESS;
   }
 
   time_t lw = 0;
 
   // read last record to check if desired time is inside or outside of the file
 
   if (1) {
      char* buf = new char[s->fRecordSize];
 
      status = ReadRecord(s->fFileName.c_str(), fd, s->fDataOffset, s->fRecordSize, nrec - 1, buf);
      if (status != HS_SUCCESS) {
         delete[] buf;
         ::close(fd);
         return HS_FILE_ERROR;
      }
 
      lw = *(DWORD*)buf;
 
      delete[] buf;
   }
 
   if (lw >= timestamp) {
      off64_t irec = 0;
      time_t trec = 0;
      off64_t iunused = 0;
      time_t tunused = 0;
      time_t tstart = 0; // not used
      time_t tend   = 0; // not used
 
      status = FindTime(s->fFileName.c_str(), fd, s->fDataOffset, s->fRecordSize, nrec, timestamp, &irec, &trec, &iunused, &tunused, &tstart, &tend, 0*debug);
      if (status != HS_SUCCESS) {
         ::close(fd);
         return HS_FILE_ERROR;
      }
 
      assert(trec < timestamp);
 
      lw = trec;
   }
 
   if (last_written)
      *last_written = lw;
 
   if (debug)
      printf("FileHistory::read_last_written: file %s, schema time %s..%s, timestamp %s, last_written %s\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(timestamp).c_str(), TimeToString(lw).c_str());
 
   assert(lw < timestamp);
 
   ::close(fd);
 
   return HS_SUCCESS;
}

Here is the call graph for this function:

remove_inactive_columns()

void HsFileSchema::remove_inactive_columns ( )

inlinevirtual

Implements HsSchema.

Definition at line 795 of file history_schema.cxx.

{ /* empty */ };

write_event()

int HsFileSchema::write_event ( const time_t  t, const char *  data, const size_t  data_size  )

virtual

Implements HsSchema.

Definition at line 2483 of file history_schema.cxx.

{
   HsFileSchema* s = this;
 
   assert(s->fVariables.size() == s->fOffsets.size());
 
   if (s->fWriterFd < 0) {
      s->fWriterFd = open(s->fFileName.c_str(), O_RDWR);
      if (s->fWriterFd < 0) {
         cm_msg(MERROR, "FileHistory::write_event", "Cannot write to \'%s\', open() errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
         return HS_FILE_ERROR;
      }
 
      //static_assert((sizeof(int)==3),"here!");
      //static_assert((sizeof(off64_t)==7),"here!");
      //static_assert((sizeof(size_t)==3),"here!");
      //static_assert((sizeof(intmax_t)==3),"here!");
 
      //off64_t zzz = 1;
      //off64_t xxx = zzz<<60;
      //int yyy = xxx;
      //printf("%d", yyy);
 
      off64_t file_size = ::lseek64(s->fWriterFd, 0, SEEK_END);
 
      if (file_size < 0) {
         cm_msg(MERROR, "FileHistory::write_event", "Cannot read file size of \'%s\', lseek64(SEEK_END) errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
         return HS_FILE_ERROR;
      }
 
      off64_t nrec = 0;
 
      if (file_size >= s->fDataOffset) {
         nrec = (file_size - s->fDataOffset)/s->fRecordSize;
      }
 
      if (nrec < 0)
         nrec = 0;
 
      off64_t data_end = s->fDataOffset + nrec*s->fRecordSize;
      
      //printf("write_event: file_size %jd, nrec %jd, data_end %jd\n", (intmax_t)file_size, (intmax_t)nrec, (intmax_t)data_end);
 
      if (data_end != file_size) {
         if (nrec > 0)
            cm_msg(MERROR, "FileHistory::write_event", "File \'%s\' may be truncated, data offset %jd, record size %zu, file size: %jd, should be %jd, making it so", s->fFileName.c_str(), (intmax_t)s->fDataOffset, s->fRecordSize, (intmax_t)file_size, (intmax_t)data_end);
 
         off64_t status64 = ::lseek64(s->fWriterFd, data_end, SEEK_SET);
 
         if (status64 < 0) {
            cm_msg(MERROR, "FileHistory::write_event", "Cannot seek \'%s\' to offset %jd, lseek64() errno %d (%s)", s->fFileName.c_str(), (intmax_t)data_end, errno, strerror(errno));
            return HS_FILE_ERROR;
         }
 
         int status = ::ftruncate64(s->fWriterFd, data_end);
 
         if (status < 0) {
            cm_msg(MERROR, "FileHistory::write_event", "Cannot truncate \'%s\' to size %jd, ftruncate64() errno %d (%s)", s->fFileName.c_str(), (intmax_t)data_end, errno, strerror(errno));
            return HS_FILE_ERROR;
         }
      }
 
      s->fFileSizeInitial = data_end;
      s->fFileSize = data_end;
   }
 
   size_t expected_size = s->fRecordSize - 4;
 
   // sanity check: record_size and n_bytes are computed from the byte counts in the file header
   assert(expected_size == s->fNumBytes);
 
   if (s->fLastSize == 0)
      s->fLastSize = expected_size;
 
   if (data_size != s->fLastSize) {
      cm_msg(MERROR, "FileHistory::write_event", "Event \'%s\' data size mismatch, expected %zu bytes, got %zu bytes, previously %zu bytes", s->fEventName.c_str(), expected_size, data_size, s->fLastSize);
      //printf("schema:\n");
      //s->print();
 
      if (data_size < expected_size)
         return HS_FILE_ERROR;
 
      // truncate for now
      // data_size = expected_size;
      s->fLastSize = data_size;
   }
 
   size_t size = 4 + expected_size;
 
   if (size != s->fRecordBufferSize) {
      s->fRecordBuffer = (char*)realloc(s->fRecordBuffer, size);
      assert(s->fRecordBuffer != NULL);
      s->fRecordBufferSize = size;
   }
 
   memcpy(s->fRecordBuffer, &t, 4);
   memcpy(s->fRecordBuffer+4, data, expected_size);
 
   ssize_t wr = write(s->fWriterFd, s->fRecordBuffer, size);
   if ((size_t)wr != size) {
      cm_msg(MERROR, "FileHistory::write_event", "Cannot write to \'%s\', write(%zu) returned %zd, errno %d (%s)", s->fFileName.c_str(), size, wr, errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   s->fFileSize += size;
 
#if 0
   status = write(s->fWriterFd, &t, 4);
   if (status != 4) {
      cm_msg(MERROR, "FileHistory::write_event", "Cannot write to \'%s\', write(timestamp) errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   status = write(s->fWriterFd, data, expected_size);
   if (status != expected_size) {
      cm_msg(MERROR, "FileHistory::write_event", "Cannot write to \'%s\', write(%d) errno %d (%s)", s->fFileName.c_str(), data_size, errno, strerror(errno));
      return HS_FILE_ERROR;
   }
#endif
 
   return HS_SUCCESS;
}

Here is the call graph for this function:

Member Data Documentation

fDataOffset

off64_t HsFileSchema::fDataOffset = 0

Definition at line 763 of file history_schema.cxx.

fFileName

std::string HsFileSchema::fFileName

Definition at line 761 of file history_schema.cxx.

fFileSize

off64_t HsFileSchema::fFileSize = 0

Definition at line 771 of file history_schema.cxx.

fFileSizeInitial

off64_t HsFileSchema::fFileSizeInitial = 0

Definition at line 770 of file history_schema.cxx.

fLastSize

size_t HsFileSchema::fLastSize = 0

Definition at line 764 of file history_schema.cxx.

fRecordBuffer

char* HsFileSchema::fRecordBuffer = NULL

Definition at line 767 of file history_schema.cxx.

fRecordBufferSize

size_t HsFileSchema::fRecordBufferSize = 0

Definition at line 766 of file history_schema.cxx.

fRecordSize

size_t HsFileSchema::fRecordSize = 0

Definition at line 762 of file history_schema.cxx.

fWriterFd

int HsFileSchema::fWriterFd = -1

Definition at line 765 of file history_schema.cxx.

---

The documentation for this class was generated from the following file:

• src/history_schema.cxx