Inheritance diagram for HsFileSchema:

Collaboration diagram for HsFileSchema:

[legend]

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 int 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

int	fRecordSize = 0

int	fDataOffset = 0

int	fLastSize = 0

int	fWriterFd = -1

int	fRecordBufferSize = 0

char *	fRecordBuffer = NULL

Public Attributes inherited from HsSchema
std::string	fEventName

time_t	fTimeFrom = 0

time_t	fTimeTo = 0

std::vector< HsSchemaEntry >	fVariables

std::vector< int >	fOffsets

int	fNumBytes = 0

int	fCountWriteUndersize = 0

int	fCountWriteOversize = 0

int	fWriteMaxSize = 0

int	fWriteMinSize = 0

bool	fDisabled = true

Detailed Description

Definition at line 733 of file history_schema.cxx.

Constructor & Destructor Documentation

◆ HsFileSchema()

HsFileSchema::HsFileSchema ( )

inline

Definition at line 747 of file history_schema.cxx.

   {
      // empty
   }

◆ ~HsFileSchema()

HsFileSchema::~HsFileSchema ( )

inline

Definition at line 752 of file history_schema.cxx.

   {
      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 2550 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 768 of file history_schema.cxx.

768{ return HS_SUCCESS; };

◆ print()

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

virtual

Reimplemented from HsSchema.

Definition at line 809 of file history_schema.cxx.

{
   unsigned nv = this->fVariables.size();
   printf("event [%s], file_name [%s], time %s..%s, %d variables, %d bytes, dat_offset %d, record_size %d\n", this->fEventName.c_str(), this->fFileName.c_str(), TimeToString(this->fTimeFrom).c_str(), TimeToString(this->fTimeTo).c_str(), nv, fNumBytes, fDataOffset, fRecordSize);
   if (print_tags) {
      for (unsigned j=0; j<nv; j++)
         printf("  %d: 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 2801 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;
   }
 
   off_t file_size = ::lseek(fd, 0, SEEK_END);
 
   if (file_size == (off_t)-1) {
      cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', fseek(SEEK_END) errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
      ::close(fd);
      return HS_FILE_ERROR;
   }
 
   int nrec = (file_size - s->fDataOffset)/s->fRecordSize;
   if (nrec < 0)
      nrec = 0;
 
   if (nrec < 1) {
      ::close(fd);
      return HS_SUCCESS;
   }
 
   int iunused = 0;
   time_t tunused = 0;
   int irec = 0;
   time_t trec   = 0;
   time_t tstart = 0;
   time_t tend   = 0;
 
   int status = FindTime(s->fFileName.c_str(), fd, s->fDataOffset, s->fRecordSize, nrec, start_time, &iunused, &tunused, &irec, &trec, &tstart, &tend, 0*debug);
 
   if (status != HS_SUCCESS) {
      ::close(fd);
      return HS_FILE_ERROR;
   }
 
   if (debug) {
      printf("FindTime %d, nrec %d, (%d, %s) (%d, %s), tstart %s, tend %s, want %s\n", status, nrec, iunused, TimeToString(tunused).c_str(), 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;
 
   off_t fpos = s->fDataOffset + irec*s->fRecordSize;
   
   off_t xpos = ::lseek(fd, fpos, SEEK_SET);
   if (xpos == (off_t)-1) {
      cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', lseek(%zu) errno %d (%s)", s->fFileName.c_str(), (size_t)fpos, errno, strerror(errno));
      ::close(fd);
      return HS_FILE_ERROR;
   }
   
   char* buf = new char[s->fRecordSize];
 
   int prec = irec;
 
   while (1) {
      status = ::read(fd, buf, s->fRecordSize);
      if (status == 0) // EOF
         break;
      if (status == -1) {
         cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', read() errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
         break;
      }
      if (status != s->fRecordSize) {
         cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', short read() returned %d instead of %d bytes", s->fFileName.c_str(), status, 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 %d timestamp %s is before start time %s", s->fFileName.c_str(), 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 %d timestamp %s is after last timestamp %s", s->fFileName.c_str(), 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 %d timestamp %s is before timestamp %s of variable %d", s->fFileName.c_str(), 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 2721 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;
   }
 
   int file_size = ::lseek(fd, 0, SEEK_END);
 
   int nrec = (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) {
      int irec = 0;
      time_t trec = 0;
      int 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 766 of file history_schema.cxx.

766{ /* empty */ };

◆ write_event()

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

virtual

Implements HsSchema.

Definition at line 2454 of file history_schema.cxx.

{
   HsFileSchema* s = this;
 
   assert(s->fVariables.size() == s->fOffsets.size());
 
   int status;
 
   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;
      }
 
      int file_size = lseek(s->fWriterFd, 0, SEEK_END);
 
      int nrec = (file_size - s->fDataOffset)/s->fRecordSize;
      if (nrec < 0)
         nrec = 0;
      int data_end = s->fDataOffset + nrec*s->fRecordSize;
 
      //printf("file_size %d, nrec %d, data_end %d\n", file_size, nrec, data_end);
 
      if (data_end != file_size) {
         if (nrec > 0)
            cm_msg(MERROR, "FileHistory::write_event", "File \'%s\' may be truncated, data offset %d, record size %d, file size: %d, should be %d, truncating the file", s->fFileName.c_str(), s->fDataOffset, s->fRecordSize, file_size, data_end);
 
         status = lseek(s->fWriterFd, data_end, SEEK_SET);
         if (status < 0) {
            cm_msg(MERROR, "FileHistory::write_event", "Cannot seek \'%s\' to offset %d, lseek() errno %d (%s)", s->fFileName.c_str(), data_end, errno, strerror(errno));
            return HS_FILE_ERROR;
         }
         status = ftruncate(s->fWriterFd, data_end);
         if (status < 0) {
            cm_msg(MERROR, "FileHistory::write_event", "Cannot truncate \'%s\' to size %d, ftruncate() errno %d (%s)", s->fFileName.c_str(), data_end, errno, strerror(errno));
            return HS_FILE_ERROR;
         }
      }
   }
 
   int 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 %d bytes, got %d bytes, previously %d 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;
   }
 
   int 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);
 
   status = write(s->fWriterFd, s->fRecordBuffer, size);
   if (status != size) {
      cm_msg(MERROR, "FileHistory::write_event", "Cannot write to \'%s\', write(%d) returned %d, errno %d (%s)", s->fFileName.c_str(), size, status, errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
#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

int HsFileSchema::fDataOffset = 0

Definition at line 739 of file history_schema.cxx.

◆ fFileName

std::string HsFileSchema::fFileName

Definition at line 737 of file history_schema.cxx.

◆ fLastSize

int HsFileSchema::fLastSize = 0

Definition at line 740 of file history_schema.cxx.

◆ fRecordBuffer

char* HsFileSchema::fRecordBuffer = NULL

Definition at line 743 of file history_schema.cxx.

◆ fRecordBufferSize

int HsFileSchema::fRecordBufferSize = 0

Definition at line 742 of file history_schema.cxx.

◆ fRecordSize

int HsFileSchema::fRecordSize = 0

Definition at line 738 of file history_schema.cxx.

◆ fWriterFd

int HsFileSchema::fWriterFd = -1

Definition at line 741 of file history_schema.cxx.

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

src/history_schema.cxx

Public Member Functions

Public Attributes

Detailed Description

Constructor & Destructor Documentation

◆ HsFileSchema()

◆ ~HsFileSchema()

Member Function Documentation

◆ close()

◆ flush_buffers()

◆ print()

◆ read_data()

◆ read_last_written()

◆ remove_inactive_columns()

◆ write_event()

Member Data Documentation

◆ fDataOffset

◆ fFileName

◆ fLastSize

◆ fRecordBuffer

◆ fRecordBufferSize

◆ fRecordSize

◆ fWriterFd