SEV garbage_collect revisited, automatic upgrade from sev_version 1 to 2

sev/exe/sev_server/src/sev_server.cpp

@@ -659,27 +659,7 @@ int sev_server::check_histitems( sev_sMsgHistItems *msg, unsigned int size)
   printf( "----  Node up (%d) ----\n", nid);
   return 1;
 }
-/*
-int sev_server::receive_histdata( sev_sMsgHistDataStore *msg, unsigned int size)
-{
-  sev_sHistData *dp = (sev_sHistData *)&msg->Data;
-  pwr_tTime time;
 
-  while ( (char *)dp - (char *)msg < (int)size) {
-    sev_refid sevid(dp->sevid);
-    iterator_refid it = m_refid.find( sevid);
-    if ( it == m_refid.end()) {
-      dp = (sev_sHistData *)((char *)dp + sizeof( *dp) - sizeof(dp->data) +  dp->size);
-      continue;
-    }
-    unsigned int idx = it->second;
-    time = net_NetTimeToTime( &msg->Time);
-    m_db->store_value( &m_sts, idx, 0, time, &dp->data, dp->size);
-    dp = (sev_sHistData *)((char *)dp + sizeof( *dp) - sizeof(dp->data) +  dp->size);
-  }
-  return 1;
-}
-*/
 int sev_server::receive_histdata( sev_sMsgHistDataStore *msg, unsigned int size)
 {
   pwr_tStatus sts;
@@ -704,9 +684,6 @@ int sev_server::receive_histdata( sev_sMsgHistDataStore *msg, unsigned int size)
   return 1;
 }
 
-
-
-
 int sev_server::send_histdata( qcom_sQid tgt, sev_sMsgHistDataGetRequest *rmsg, unsigned int size)
 {
   pwr_tTime *tbuf;
@@ -885,11 +862,11 @@ void sev_server::garbage_item( int idx)
 
   if( m_db->m_items[idx].attrnum > 1 ) {
     m_db->delete_old_objectdata( &m_sts, m_db->m_items[idx].tablename, 
-                                 m_db->m_items[idx].options, limit);
+                                 m_db->m_items[idx].options, limit, m_db->m_items[idx].scantime, (float)sev_cGarbageCycle);
   }
   else {
     m_db->delete_old_data( &m_sts, m_db->m_items[idx].tablename, 
-         m_db->m_items[idx].options, limit);
+         m_db->m_items[idx].options, limit, m_db->m_items[idx].scantime, (float)sev_cGarbageCycle);
   }
 }
 

sev/exe/sev_server/src/sev_server.h

@@ -59,12 +59,9 @@ class sev_server {
 
   sev_server() : m_server_status(0), m_msg_id(0) {}
 
-//  typedef map<sev_refid, unsigned int>::iterator iterator_refid;
-
   pwr_tStatus m_sts;
   pwr_tStatus m_server_status;
   vector<sev_node> m_nodes;
-//  map<sev_refid, unsigned int> m_refid;
 
   tree_sTable *m_refid;
 

sev/lib/sev/src/sev_db.h

@@ -93,9 +93,8 @@ class sev_db {
 			  unsigned int *bsize) { return 0;}
   virtual int get_items( pwr_tStatus *sts) { return 0;}
   virtual int delete_old_data( pwr_tStatus *sts, char *tablename, 
-			       pwr_tMask options, pwr_tTime limit) { return 0;}
+			       pwr_tMask options, pwr_tTime limit, pwr_tFloat32 scantime, pwr_tFloat32 garbagecycle) { return 0;}
 
-  virtual char *oid_to_table( pwr_tOid oid, char *aname) { return 0;}
   virtual int check_objectitem( pwr_tStatus *sts, char *tablename, pwr_tOid oid, char *oname, char *aname, 
                          pwr_tDeltaTime storagetime, 
                          char *description, pwr_tFloat32 scantime, 
@@ -112,7 +111,7 @@ class sev_db {
   virtual int check_objectitemattr( pwr_tStatus *sts, char *tablename, pwr_tOid oid, char *aname, char *oname, 
 																	  pwr_eType type, unsigned int size, unsigned int *idx) { return 0;}
   virtual int delete_old_objectdata( pwr_tStatus *sts, char *tablename, 
-                                     pwr_tMask options, pwr_tTime limit) { return 0;}
+                                     pwr_tMask options, pwr_tTime limit, pwr_tFloat32 scantime, pwr_tFloat32 garbagecycle) { return 0;}
   virtual int get_objectvalues( pwr_tStatus *sts, sev_item *item,
                                 unsigned int size, pwr_tTime *starttime, pwr_tTime *endtime, 
                                 int maxsize, pwr_tTime **tbuf, void **vbuf, unsigned int *bsize) { return 0;}

sev/lib/sev/src/sev_dbms.cpp

@@ -271,7 +271,7 @@ MYSQL *sev_dbms_env::createDb(void)
   rc = mysql_query( m_con, query);
   if (rc) printf( "Create items table: %s\n", mysql_error(m_con));
 
-  updateDB_to_SevVersion2();
+  createSevVersion2Tables();
 
   return con;
 }
@@ -310,7 +310,7 @@ int sev_dbms_env::checkAndUpdateVersion(unsigned int version)
   //add code for new versions here
   if(old_version < 2 ) {
     printf("Updating database tables to sev version 2\n");
-    updateDB_to_SevVersion2();
+    updateDBToSevVersion2();
   }
 
   if(old_version != version) {
@@ -325,7 +325,70 @@ int sev_dbms_env::checkAndUpdateVersion(unsigned int version)
   return 1;
 }
 
-int sev_dbms_env::updateDB_to_SevVersion2(void)
+int sev_dbms_env::updateDBToSevVersion2(void)
+{
+  createSevVersion2Tables();
+
+  int rc;
+  char query[300];
+
+  sprintf( query, "select id,tablename,vid,oix from items order by id");
+
+  rc = mysql_query( con(), query);
+  if (rc) {
+    printf("In %s row %d:\n", __FILE__, __LINE__);
+    printf( "%s: %s\n", __FUNCTION__, mysql_error(con()));
+    return 0;
+  }
+
+  MYSQL_ROW row;
+  MYSQL_RES *result = mysql_store_result( con());
+  if ( !result) {
+    printf("In %s row %d:\n", __FILE__, __LINE__);
+    printf( "GetValues Result Error\n");
+    return 0;
+  }
+  int rows = mysql_num_rows( result);
+
+  vector<sev_item> itemsVec;
+  for ( int i = 0; i < rows; i++) {
+    sev_item item;
+    row = mysql_fetch_row( result);
+    if (!row) break;
+    item.id = atoi( row[0]);
+    strncpy( item.tablename, row[1], sizeof(item.tablename));
+    item.oid.vid = strtoul( row[2], 0, 10);
+    item.oid.oix = strtoul( row[3], 0, 10);
+    itemsVec.push_back( item);
+  }
+  mysql_free_result( result);
+
+  char newTableName[256];
+  for(size_t i=0; i < itemsVec.size(); i++) {
+    sprintf(newTableName, "%s_%d", sev_dbms::oid_to_table(itemsVec[i].oid, (char*)""), itemsVec[i].id);
+
+    printf("UPDATE TO SEV_VERSION 2: Renaming table %s to %s \n", itemsVec[i].tablename, newTableName);
+    errh_Info("UPDATE TO SEV_VERSION 2: Renaming table %s to %s", itemsVec[i].tablename, newTableName);
+
+    sprintf(query, "RENAME TABLE %s to %s", itemsVec[i].tablename, newTableName);
+    rc = mysql_query( con(), query);
+    if (rc) {
+      printf("In %s row %d:\n", __FILE__, __LINE__);
+      printf( "%s: %s\n", __FUNCTION__, mysql_error(con()));
+      return 0;
+    }
+
+    sprintf(query, "update items set tablename='%s' where id=%d", newTableName, itemsVec[i].id);
+    rc = mysql_query( con(), query);
+    if (rc) {
+      printf("In %s row %d:\n", __FILE__, __LINE__);
+      printf( "%s: %s\n", __FUNCTION__, mysql_error(con()));
+      return 0;
+    }
+  }
+  return 1;
+}
+int sev_dbms_env::createSevVersion2Tables(void)
 {
   char query[400];
 
@@ -1622,7 +1685,7 @@ int sev_dbms::delete_item( pwr_tStatus *sts, pwr_tOid oid, char *aname)
 }
 
 int sev_dbms::delete_old_data( pwr_tStatus *sts, char *tablename, 
-			       pwr_tMask options, pwr_tTime limit)
+			       pwr_tMask options, pwr_tTime limit, pwr_tFloat32 scantime, pwr_tFloat32 garbagecycle)
 {
   char query[300];
   char timstr[40];
@@ -1633,7 +1696,13 @@ int sev_dbms::delete_old_data( pwr_tStatus *sts, char *tablename,
 
 
   if(options & pwr_mSevOptionsMask_ReadOptimized) {
-    pwr_tUInt64 nbRowsToClean = 2000;
+    //TODO Change 86400 to sev_cGarbageCycle
+    //We scan trough as many rows that we write between two garbage cycles plus some extra rows
+    //to be sure that the table not grows to much.
+    pwr_tUInt64 nbRowsToClean = 86400*2; //default value
+    if(scantime > 0.0)
+      nbRowsToClean = (pwr_tUInt64)((garbagecycle/scantime) + 10000.0);
+
     nbRowsToClean += get_minFromIntegerColumn(tablename, (char *)"id");
     if ( options & pwr_mSevOptionsMask_PosixTime)
       sprintf( query, "delete from %s where id < %llu and time < %ld;",
@@ -2517,7 +2586,7 @@ int sev_dbms::get_objectitemattributes( pwr_tStatus *sts, sev_item *item, char *
 }
 
 int sev_dbms::delete_old_objectdata( pwr_tStatus *sts, char *tablename, 
-			                               pwr_tMask options, pwr_tTime limit)
+			                               pwr_tMask options, pwr_tTime limit, pwr_tFloat32 scantime, pwr_tFloat32 garbagecycle)
 {
   char query[300];
   char timstr[40];
@@ -2528,7 +2597,12 @@ int sev_dbms::delete_old_objectdata( pwr_tStatus *sts, char *tablename,
 
 
   if(options & pwr_mSevOptionsMask_ReadOptimized) {
-    pwr_tUInt64 nbRowsToClean = 2000;
+    //We scan trough as many rows that we write between two garbage cycles plus some extra rows
+    //to be sure that the table not grows to much.
+    pwr_tUInt64 nbRowsToClean = 86400*2; //default value
+    if(scantime > 0.0)
+      nbRowsToClean = (pwr_tUInt64)((garbagecycle/scantime) + 10000.0);
+
     nbRowsToClean += get_minFromIntegerColumn(tablename, (char *)"sev__id");
     if ( options & pwr_mSevOptionsMask_PosixTime)
       sprintf( query, "delete from %s where sev__id < %llu and sev__time < %ld;",

sev/lib/sev/src/sev_dbms.h

@@ -68,7 +68,8 @@ class sev_dbms_env
 	   const char *dbName, unsigned int port, const char *socket);
 
   int checkAndUpdateVersion(unsigned int version);
-  int updateDB_to_SevVersion2(void);
+  int updateDBToSevVersion2(void);
+  int createSevVersion2Tables(void);
   MYSQL *createDb(void);
   MYSQL *openDb(void);
   bool exists() { return m_exists;}    
@@ -125,7 +126,7 @@ class sev_dbms : public sev_db {
 		  pwr_tTime *starttime, 
 		  pwr_tTime *endtime, int maxsize, pwr_tTime **tbuf, void **vbuf, unsigned int *bsize);
   int delete_old_data( pwr_tStatus *sts, char *tablename, 
-		       pwr_tMask options, pwr_tTime limit);
+		       pwr_tMask options, pwr_tTime limit, pwr_tFloat32 scantime, pwr_tFloat32 garbagecycle);
   int delete_item( pwr_tStatus *sts, pwr_tOid oid, char *aname);
 
   int get_items( pwr_tStatus *sts);
@@ -137,7 +138,7 @@ class sev_dbms : public sev_db {
 		  char *description, char *unit, pwr_tFloat32 scantime, 
 		  pwr_tFloat32 deadband, pwr_tMask options);
   int remove_item( pwr_tStatus *sts, pwr_tOid oid, char *aname);
-  char *oid_to_table( pwr_tOid oid, char *aname);
+  static char *oid_to_table( pwr_tOid oid, char *aname);
   char *pwrtype_to_type( pwr_eType type, unsigned int size);
   static int timestr_to_time( char *tstr, pwr_tTime *ts);
   int check_objectitem( pwr_tStatus *sts, char *tablename, pwr_tOid oid, char *oname, char *aname, 
@@ -161,7 +162,7 @@ class sev_dbms : public sev_db {
   int check_objectitemattr( pwr_tStatus *sts, char *tablename, pwr_tOid oid, char *aname, char *oname, 
 																	pwr_eType type, unsigned int size, unsigned int *idx);
   int delete_old_objectdata( pwr_tStatus *sts, char *tablename, 
-                             pwr_tMask options, pwr_tTime limit);
+                             pwr_tMask options, pwr_tTime limit, pwr_tFloat32 scantime, pwr_tFloat32 garbagecycle);
   int check_deadband(pwr_eType type, unsigned int size, pwr_tFloat32 deadband, void *value, void *oldvalue);
   int get_objectvalues( pwr_tStatus *sts, sev_item *item, unsigned int size, pwr_tTime *starttime, pwr_tTime *endtime, 
 			                  int maxsize, pwr_tTime **tbuf, void **vbuf, unsigned int *bsize);

src/exe/rt_sevhistmon/src/rt_sevhistmon.cpp

@@ -961,6 +961,7 @@ int rt_sevhistmon::send_itemlist( pwr_tNid nid)
   put.msg_id = m_msg_id++;
 
   if ( !qcom_Put( &sts, &tgt, &put)) {
+    printf("rt_sevhistmon: ItemList send ERROR sts:%d\n", sts);
     qcom_Free( &sts, put.data);
     return 0;
   }
@@ -980,6 +981,7 @@ int rt_sevhistmon::send_itemlist( pwr_tNid nid)
       set_status();
     }
   }
+  printf("rt_sevhistmon: ItemList sent\n");
   return 1;
 }
 
@@ -1008,7 +1010,11 @@ int rt_sevhistmon::mainloop()
       case sev_cMsgClass:
         switch ( get.type.s) {
           case sev_eMsgType_NodeUp:
+            printf("rt_sevhistmon: Node up recevied\n");
+            send_itemlist( get.sender.nid);
+            break;
           case sev_eMsgType_HistItemsRequest:
+            printf("rt_sevhistmon: HistitemsRequest recevied\n");
             send_itemlist( get.sender.nid);
             break;
           case sev_eMsgType_ServerStatus: