source: Evidence/DColl/DColl.cc@ 132

/********************************************************************\

  Central data collector of the Evidence Control System
  
  - DColl subscribes to all services given by the configuration
    server and writes these to the data file at every update.
  - For each service, it creates a new service with '.hist' appended that
    contains a history of events kept within a ring buffer. Each entry will
        be the result of a conversion to double of the text written to the data file.
  - The command 'DColl/Log' writes the associated string to the log 
    file specified in the configuration.
  
  Oliver Grimm, December 2009

\********************************************************************/

#define SERVER_NAME "DColl"

#include "../Evidence.h"

#define NO_LINK "__&DIM&NOLINK&__" // for checking if DIMserver is alive

//
// Class declaration
//
class DataHandler:      public DimClient, public DimCommand,
                                        public DimBrowser, public EvidenceServer {
  private:
    pthread_mutex_t DataMutex;
    pthread_mutex_t LogMutex;

    unsigned int NumItems;
    FILE *DataFile;
    FILE *LogFile;
    char *Filename;

    DimStampedInfo **DataItem;
    DimService **HistService;       
    int HistSize;
    struct EvidenceHistoryItem **HistBuffer;
        int *HistPointer;
        
    void infoHandler();
    void commandHandler();
    bool ReallocMem(void *&, int);
            
  public:
    DataHandler();
    ~DataHandler();
}; 

//
// Constructor
//
DataHandler::DataHandler(): DimCommand((char *) "DColl/Log", (char *) "C"), EvidenceServer(SERVER_NAME) {

  // Initialization to prevent freeing unallocated memory 
  DataFile = NULL;
  Filename = NULL;

  DataItem = NULL;
  HistService = NULL;
  HistBuffer = NULL;
  HistPointer = NULL;
  
  // Request configuration data
  char *Items = GetConfig(SERVER_NAME " items");
  char *DataDir = GetConfig(SERVER_NAME " datadir");
  char *Logname = GetConfig(SERVER_NAME " logfile");
  HistSize = atoi(GetConfig(SERVER_NAME " histsize"));
  if (HistSize < 1) HistSize = 1; // Minimum one items
   
  // Create mutex for thread synchronization 
  if ((errno=pthread_mutex_init(&DataMutex, NULL)) != 0) {
    Msg(FATAL, "pthread_mutex_init() failed for data file (%s)", strerror(errno));
  }
  if ((errno=pthread_mutex_init(&LogMutex, NULL)) != 0) {
    Msg(FATAL, "pthread_mutex_init() failed for log file (%s)", strerror(errno));
  }

  // Interpret DIM services to observe and prepare history buffer
  char *Buf, *ServiceName, *Format;
  int NumServices;
  NumItems = 0;
  char *NextToken = strtok(Items, " \t");
  while (NextToken != NULL) {
    NumServices = getServices(NextToken);
    for (int j=0; j<NumServices; j++)  {
          if (getNextService(ServiceName, Format) == DimSERVICE) {

            // Check if already subsubed to this service
                for (int i=0; i<NumItems; i++) {
                  if(strcmp(ServiceName, DataItem[i]->getName()) == 0) continue;
                }
                
            // Increase capacity of arrays
            if (!ReallocMem((void *&) DataItem, NumItems+1)) continue;
            if (!ReallocMem((void *&) HistService, NumItems+1)) continue;
            if (!ReallocMem((void *&) HistBuffer, NumItems+1)) continue;
            if (!ReallocMem((void *&) HistPointer, NumItems+1)) continue;

                // Subscribe to service
        DataItem[NumItems] = new DimStampedInfo(ServiceName, (char *) NO_LINK, this);

        // Create history service
        HistBuffer[NumItems] = new struct EvidenceHistoryItem [HistSize];
                memset(HistBuffer[NumItems], 0, HistSize*sizeof(EvidenceHistoryItem));
                HistPointer[NumItems] = 0;

        if (asprintf(&Buf, "%s.hist", ServiceName) == -1) {
          Msg(ERROR, "Could not create Hist service for %s because asprintf() failed", ServiceName);
        }
        else {
                  HistService[NumItems] = new DimService (Buf, (char *) "C",
                                                                HistBuffer[NumItems], HistSize*sizeof(EvidenceHistoryItem));
          free(Buf);
                }
                NumItems++;
          }
        }
    NextToken = strtok(NULL, " \t");
  }

  // Open data file
  time_t rawtime = time(NULL);
  struct tm * timeinfo = gmtime(&rawtime);
  if(timeinfo->tm_hour >= 13) rawtime += 12*60*60;
  timeinfo = gmtime(&rawtime);

  if (asprintf(&Filename, "%s/%d%02d%02d.slow", DataDir, timeinfo->tm_year+1900, timeinfo->tm_mon+1, timeinfo->tm_mday) == -1) {
    Filename = NULL;
    Msg(FATAL, "Could not create filename with asprintf()(%s)", strerror(errno));
  }
  if ((DataFile = fopen(Filename, "a")) == NULL) {
    Msg(FATAL, "Could not open data file '%s' (%s)", Filename, strerror(errno));
  }
  
  // Open log file
  if ((LogFile = fopen(Logname, "a")) == NULL) {
    Msg(FATAL, "Could not open log file '%s' (%s)", Logname, strerror(errno));
  }

}

//
// Destructor: Close files and free memory
//
DataHandler::~DataHandler() {

  // Close files
  if (LogFile != NULL && fclose(LogFile) != 0) {
        Msg(ERROR, "Could not close log file (%s)", strerror(errno));
  }
  if (DataFile != NULL && fclose(DataFile) != 0) {
        Msg(ERROR, "Error: Could not close data file (%s)", strerror(errno));
  }     

  // Free all memory
  for (int i=0; i<NumItems; i++) {
    delete[] HistBuffer[i];
    delete DataItem[i];
  }

  free(Filename);
  free(HistService);
  free(HistPointer);
  free(HistBuffer);
  free(DataItem);
  
  // Destroy mutex
  pthread_mutex_destroy (&LogMutex);
  pthread_mutex_destroy (&DataMutex);
}

//
// Implementation of data handling
//
// More than one infoHandler() might run in parallel, therefore
// the mutex mechanism is used to serialize writing to the file
void DataHandler::infoHandler() {
  
  DimInfo *Info = getInfo();

  // Check if service actually available, if it contains data and if data file is open
  if (Info->getSize()==strlen(NO_LINK)+1 && strcmp(Info->getString(), NO_LINK)==0) return;
  if (Info->getSize() == 0 || DataFile == NULL) return;

  // Identify index of service
  int Service;  
  for (Service=0; Service<NumItems; Service++) if (Info == DataItem[Service]) break;
  if (Service == NumItems) return;  // Not found: should never happen

  pthread_mutex_lock(&DataMutex);
  
  // Write data header
  time_t RawTime = Info->getTimestamp();
  struct tm *TM = localtime(&RawTime);

  fprintf(DataFile, "%s %d %d %d %d %d %d %d %lu ", Info->getName(), TM->tm_year+1900, TM->tm_mon+1, TM->tm_mday, TM->tm_hour, TM->tm_min, TM->tm_sec, Info->getTimestampMillisecs(), Info->getTimestamp());

  // Translate info data into ASCII and write to file and to history buffer
  char *Text = EvidenceServer::ToString(Info);
  if (Text != NULL) {
    fprintf(DataFile, "%s\n", Text);
        
        HistBuffer[Service][HistPointer[Service]].Seconds = Info->getTimestamp();
    HistBuffer[Service][HistPointer[Service]].Value = atof(Text);
    HistService[Service]->updateService();
        HistPointer[Service]++;
        if (HistPointer[Service] >= HistSize) HistPointer[Service] = 0;

        free(Text);
  }
  else fprintf(DataFile, "Cannot interpret format identifier\n");
 
  fflush(DataFile);
  if(ferror(DataFile)) Msg(ERROR, "Error writing to data file (%s)", strerror(errno));

  pthread_mutex_unlock(&DataMutex);
}

//
// Implementation of log writing (the only command is 'DColl/Log')
//
void DataHandler::commandHandler() {

  if (LogFile == NULL) return;  // Handler might be called before file is opened
 
  pthread_mutex_lock(&LogMutex);

  time_t RawTime = time(NULL);
  struct tm *TM = localtime(&RawTime);

  fprintf(LogFile, "%d/%d/%d %d:%d:%d: %s\n",
                TM->tm_mday, TM->tm_mon+1, TM->tm_year+1900,
                TM->tm_hour, TM->tm_min, TM->tm_sec, getString());

  fflush(LogFile);
  if(ferror(LogFile)) Msg(ERROR, "Error writing to log file (%s)", strerror(errno));
  pthread_mutex_unlock(&LogMutex);
}


//
// Implementation of memory re-allocation for pointer arrays
//
bool DataHandler::ReallocMem(void *&Memory, int Size) {

  void *NewMem = realloc(Memory, Size*sizeof(void *));

  if (NewMem != NULL) {
    Memory = NewMem;
        return true;
  }
  else {
        Msg(ERROR, "Could not allocate memory ()", strerror(errno));
        return false;
  }
}

//          
// Main program
//
int main() {
        
  // Static ensures calling of destructor by exit()
  static DataHandler Data;
  
  pause();        // Sleep until signal caught
}