Changeset 159 for Evidence/DColl.cc

Timestamp:

02/04/10 15:26:55 (14 years ago)

Author:

ogrimm

Message:

Change in history handling

File:

• Evidence/DColl.cc (modified) (16 diffs)

Evidence/DColl.cc

@@ -38 +38 @@
 #include <regex.h>
 
+using namespace std;
+
 //
 // Class declaration
@@ -47 +49 @@
           DimStampedInfo *DataItem;
       DimService *HistService;
-          struct EvidenceHistoryItem *HistBuffer;
-          int HistPointer;
+          char *Buffer;
+          int Next;
           double LastValue;
           double MinAbsChange;
@@ -101 +103 @@
   // Request configuration data
   char *Change = GetConfig(SERVER_NAME " minchange");
-
   DataDir = GetConfig(SERVER_NAME " datadir");
-
   char *Logname = GetConfig(SERVER_NAME " logfile");
-
   SizeUpdateDelay = atoi(GetConfig(SERVER_NAME " sizeupdate"));
-
   HistSize = atoi(GetConfig(SERVER_NAME " histsize"));
-  if (HistSize < 1) HistSize = 1; // Minimum one items
-
+  if (HistSize < 3*sizeof(int)) HistSize = 3*sizeof(int);
   HistDir = GetConfig(SERVER_NAME " histdir");
    
@@ -163 +160 @@
   // Subscribe to list of servers at DIS_DNS 
   AddService("DIS_DNS/SERVER_LIST");
-
-  DimClient::sendCommand("DColl/Log", SERVER_NAME" *** Logging started ***");
 }
 
@@ -171 +166 @@
 //
 DataHandler::~DataHandler() {
-
-  DimClient::sendCommand("DColl/Log", SERVER_NAME" *** Logging stopped ***");
 
   // Delete DIM subscriptions and command first so handlers and not called anymore
@@ -179 +172 @@
   }
   delete LogCommand;
-  
+
   // Save history buffers to files (replace '/' by '_')
+  State(INFO, "Writing history buffers to files");
+
   for (int i=0; i<List.size(); i++) {
     string Name = List[i].HistService->getName();
@@ -186 +181 @@
     FILE *File = fopen((HistDir + "/" + Name).c_str(), "wb");
         if (File != NULL) {
-      fwrite(&List[i].HistPointer, sizeof(List[i].HistPointer), 1, File);
-      fwrite(List[i].HistBuffer, sizeof(EvidenceHistoryItem), HistSize, File);
+      fwrite(&List[i].Next, sizeof(List[i].Next), 1, File);
+      fwrite(List[i].Buffer, sizeof(char), HistSize, File);
       fclose(File);
         }
         delete List[i].HistService;
-    delete[] List[i].HistBuffer;
+    delete[] List[i].Buffer;
   }
 
@@ -223 +218 @@
 void DataHandler::infoHandler() {
 
+  static const int WrapMark[] = {0, -1};
+  static const int EndMark[] = {0, 0};
+
   DimInfo *Info = getInfo();
 
   // Check if service available
-  if (Info->getSize()==strlen(NO_LINK)+1 && strcmp(Info->getString(), NO_LINK)==0) return;
-
+  if (!ServiceOK(Info)) return;
+
+  //
+  // ====== Part A: Handle service subscriptions ===
+  //
+  
   // If service is DIS_DNS/SERVER_LIST, subscribe to all SERVICE_LIST services
   if (strcmp(Info->getName(), "DIS_DNS/SERVER_LIST") == 0) {    
@@ -259 +261 @@
   }
 
+  //
+  // ====== Part B: Handle opening data files ===
+  //
+
   // If it is time to open new data file, close the current one 
   if (time(NULL) >= TimeForNextFile) {
@@ -296 +302 @@
         TimeForNextFile = mktime(T);
   }
-   
+
+  //
+  // ====== Part C: Handle writing to data file ===
+  //
+
   // Identify index of service
-  int Service;  
+  int Service;
   for (Service=0; Service<List.size(); Service++) if (Info == List[Service].DataItem) break;
   if (Service == List.size()) return;  // Service not found
 
-  // If negative value for absolute change, ignore this entry
-  if (List[Service].MinAbsChange < 0) return;
+  // If negative value for absolute change, do not write to file
+  if (List[Service].MinAbsChange >= 0) {
 
   // Write data header
@@ -313 +323 @@
   // Translate data into ASCII
   char *Text = EvidenceServer::ToString(Info);
+
   if (Text != NULL) {
         // Replace all control characters by white space
@@ -320 +331 @@
     fprintf(DataFile, "%s\n", Text);
         
-        // Add to history buffer if change large enough
-        if ((fabs(atof(Text)-List[Service].LastValue) > List[Service].MinAbsChange)) {
-          List[Service].HistBuffer[List[Service].HistPointer].Seconds = Info->getTimestamp();
-      List[Service].HistBuffer[List[Service].HistPointer].Value = atof(Text);
-      List[Service].HistService->updateService();
-          List[Service].HistPointer++;
-          if (List[Service].HistPointer >= HistSize) List[Service].HistPointer = 0;
-          List[Service].LastValue = atof(Text);
-        }
         free(Text);
   }
   else fprintf(DataFile, "Cannot interpret format identifier\n");
  
-  fflush(DataFile);
-
   // Terminate if error because otherwise infinite loop might result as
   // next call to this infoHandler() will try to (re-)open file
@@ -343 +343 @@
   }
 
-  // Update datafile size service
+  // Update datafile size service (not every time to avoid infinite loop)
   if (time(NULL) - DataSizeLastUpdate > SizeUpdateDelay) {
+    fflush(DataFile); // not continuously to reduce load
+
         DataSizekB = FileSize(DataFile);
         DataSizeService->updateService();
         DataSizeLastUpdate = time(NULL);
   }
+  
+  } // Check for MinAbsChange
+
+  //
+  // ====== Part D: Handle history service ===
+  //
+  
+  if (Info->getSize() == 0) return;
+  
+  // Check if data should be added to history buffer
+  char *Text = EvidenceServer::ToString(Info);
+  if (Text != NULL && strcmp(Info->getFormat(),"C") != 0
+          && fabs(atof(Text)-List[Service].LastValue) < fabs(List[Service].MinAbsChange)) {
+        free(Text);
+        return;
+  }
+  free(Text);
+
+  // Check if data fits into buffer
+  if (HistSize < Info->getSize() + 5*sizeof(int)) return;
+
+  int Size = Info->getSize() + 4*sizeof(int), Next = List[Service].Next;
+  void *WrapPos = NULL;
+  char *Buffer = List[Service].Buffer;
+  int Oldest = *(int *) Buffer;
+
+  // Check if buffer wrap-around (write wrap mark after Oldest is adjusted)
+  if (Next + Size >= HistSize) {
+    WrapPos = Buffer + Next;
+    Next = 4;
+  }
+
+  // Adapt pointer to oldest entry
+  while ((Oldest < Next + Size) && 
+                 (Oldest + *((int *) (Buffer + Oldest) + 1) + 2*sizeof(int) > Next)) {
+        // Check for wrap-around
+        if (memcmp(Buffer + Oldest, WrapMark, sizeof(WrapMark)) == 0) {
+          Oldest = 4;
+          continue;
+        }
+        // Check if end marker reached, then only one event fits buffer
+        if (memcmp(Buffer + Oldest, EndMark, sizeof(EndMark)) == 0) {
+          Oldest = Next;
+          break;
+        }
+        // Move to next entry
+    Oldest += *((int *) (Buffer + Oldest) + 1) + 2*sizeof(int);
+  }
+  // Update pointer in buffer
+  *(int *) Buffer = Oldest;
+
+  // Write wrap mark if necessary
+  if (WrapPos != NULL) memcpy(WrapPos, WrapMark, sizeof(WrapMark));
+
+  // Copy data into ring buffer
+  *((int *) (Buffer + Next)) = Info->getTimestamp();
+  *((int *) (Buffer + Next + sizeof(int))) = Info->getSize();
+  memcpy(Buffer + Next + 2*sizeof(int), Info->getData(), Info->getSize());
+
+  // Adjust pointer for next entry and write end marker to buffer
+  Next += Info->getSize() + 2*sizeof(int);
+  memcpy(Buffer + Next, EndMark, sizeof(EndMark));
+  
+  List[Service].Next = Next;
 }
 
@@ -403 +469 @@
   }
   
-  // Create new entry in item list
   struct Item New;
 
@@ -413 +478 @@
         }
   }
-                
+
   // Create history service
-  New.HistBuffer = new struct EvidenceHistoryItem [HistSize];
-  memset(New.HistBuffer, 0, HistSize*sizeof(EvidenceHistoryItem));
-  New.HistPointer = 0;
+  New.Buffer = new char [HistSize];
+  memset(New.Buffer, 0, HistSize);
+  *(int *) New.Buffer = 4;
+  New.Next = 4;
   New.LastValue = DBL_MAX;
   New.HistService = new DimService ((Name+".hist").c_str(), (char *) "C",
-                                          New.HistBuffer, HistSize*sizeof(EvidenceHistoryItem));
-        
+                                          New.Buffer, HistSize);
+
   // Load history buffer from file if existing
   string Filename = New.HistService->getName();
@@ -427 +493 @@
   FILE *File = fopen((HistDir + "/" + Filename).c_str(), "rb");
   if (File != NULL) {
-    fread(&New.HistPointer, sizeof(New.HistPointer), 1, File);
-    fread(New.HistBuffer, sizeof(EvidenceHistoryItem), HistSize, File);
-    fclose(File);
+    // Only load if current buffer size if equal or larger
+    if (FileSize(File) <= HistSize*sizeof(char)+sizeof(New.Next) && FileSize(File) != -1) {
+      fread(&New.Next, sizeof(New.Next), 1, File);
+      fread(New.Buffer, sizeof(char), HistSize, File);
+      fclose(File);
+        }
   }