Merge pull request #27 from CODARcode/sm_release

Fix for HBOS and COPOD: Handling cases when the number of function runtimes in each IO step is Zero (XGC run)

Author: giltirn

include/chimbuko/ad/ADOutlier.hpp

@@ -238,7 +238,7 @@ namespace chimbuko {
 
   private:
     double m_alpha; /**< Used to prevent log2 overflow */
-    double m_threshold; /**< Threshold used to filter anomalies in HBOS*/
+    double m_threshold; /**< Threshold used to filter anomalies in COPOD*/
     bool m_use_global_threshold; /**< Flag to use global threshold*/
     //double m_threshold; /** sync with global threshold */
     OutlierStatistic m_statistic; /** Which statistic to use for outlier detection */
@@ -253,13 +253,13 @@ namespace chimbuko {
   public:
 
     /**
-     * @brief Construct a new ADOutlierHBOS object
+     * @brief Construct a new ADOutlierCOPOD object
      *
      */
     ADOutlierCOPOD(OutlierStatistic stat = ExclusiveRuntime, double threshold = 0.99, bool use_global_threshold = true);
 
     /**
-     * @brief Destroy the ADOutlierHBOS object
+     * @brief Destroy the ADOutlierCOPOD object
      *
      */
     ~ADOutlierCOPOD();
@@ -308,11 +308,12 @@ namespace chimbuko {
 
   private:
     double m_alpha; /**< Used to prevent log2 overflow */
-    double m_threshold; /**< Threshold used to filter anomalies in HBOS*/
+    double m_threshold; /**< Threshold used to filter anomalies in COPOD*/
     bool m_use_global_threshold; /**< Flag to use global threshold*/
-    //double m_threshold; /** sync with global threshold */
-    OutlierStatistic m_statistic; /** Which statistic to use for outlier detection */
+    //double m_threshold; /**< sync with global threshold */
+    OutlierStatistic m_statistic; /**< Which statistic to use for outlier detection */
 
+    std::unordered_map<unsigned long, int> m_skewness; /**< skewness for each function*/
   };
 
 

scripts/launch/run_services.sh

@@ -61,18 +61,18 @@ var_dir=${base}/chimbuko/vars
 bp_dir=${base}/chimbuko/adios2
 
 #Check tau adios2 path is writable; by default this is ${bp_dir} but it can be overridden by users, eg for offline analysis
-touch ${TAU_ADIOS2_PATH}/write_check
-if [[ $? == 1 ]]; then
-    echo "Chimbuko Services: Could not write to ADIOS2 output path ${TAU_ADIOS2_PATH}, check permissions"
-    exit 1
-fi
-rm -f ${TAU_ADIOS2_PATH}/write_check
+#touch ${TAU_ADIOS2_PATH}/write_check
+#if [[ $? == 1 ]]; then
+#    echo "Chimbuko Services: Could not write to ADIOS2 output path ${TAU_ADIOS2_PATH}, check permissions"
+#    exit 1
+#fi
+#rm -f ${TAU_ADIOS2_PATH}/write_check
 
 #Get head node IP
 if command -v ip &> /dev/null
 then
     ip=$(ip -4 addr show ${service_node_iface} | grep -oP '(?<=inet\s)\d+(\.\d+){3}')
-elif command -v iconfig &> /dev/null
+elif command -v ifconfig &> /dev/null
 then
     ip=$(ifconfig 2>&1 ${service_node_iface} | grep -E -o 'inet [0-9.]+' | awk '{print $2}')
 else

scripts/provdb_analyze.py

@@ -89,6 +89,19 @@ def generateIndex(interface, keys, collection):
 
     return results
 
+#Write the index to disk
+def writeIndex(filename, index):
+    f = open(filename,"w")
+    json.dump(index, f)
+    f.close()
+
+#Read the index from disk
+def readIndex(filename):
+    f = open(filename,"r")
+    index = json.load(f)
+    f.close()
+    return index
+
 #Get an event using the index tuple obtained from the provided index
 def getEventByID(interface, index, idx_tuple):
     collection = index['__collection']
@@ -119,15 +132,53 @@ def getRecordsByKeyValue(interface, index, key, value, sort_key = 'entry'):
         lst.sort(key=lambda x: x[sort_key])
     return lst
 
-#Generate a string containing a summary of an event:  (process, rank, thread, function name, runtime, runtime/avg)
+#Generate a string containing a summary of an event:  (process, rank, thread, function name, step, exclusive runtime, inclusive runtime, outlier score, outlier severity)
 #If the event was a GPU event, the thread index will be replaced by GPU${DEVICE}/${CONTEXT}/${STREAM}
 def summarizeEvent(event):
     thr_str = "{}".format(event['tid'])
     if event['is_gpu_event']:
         thr_str = "GPU{}/{}/{}".format(event['gpu_location']['device'],event['gpu_location']['context'],event['gpu_location']['stream'])    
-    return "{} {} {} {} {}s {}".format(event['pid'],event['rid'],  thr_str, event['func'], float(event['runtime_exclusive'])/1e6, float(event['runtime_exclusive'])/float(event['func_stats']['mean']) )
-
-
+    return "pid={} rid={} tid={} func=\"{}\" step={} excl={}s tot={}s score={} severity={}".format(event['pid'],event['rid'],  thr_str, event['func'], event['io_step'], float(event['runtime_exclusive'])/1e6, float(event['runtime_total'])/1e6, event['outlier_score'], event['outlier_severity'])
+                                             
+
+#Get the function profile information for application index 'app'
+#Returns a dictionary indexed by the function name with entries
+#    'excl_time_tot' : the total exclusive time (in seconds) spent in the function over all ranks and threads
+#    'count' : the number of times the function was encountered over all ranks and threads
+#    'excl_time_mean' : the average exclusive time (in seconds) spent in the function over all occurences
+#    'frac_time' : the amount of time spent in the function over all occurences relative to the total runtime spent in all threads and ranks
+def getFunctionProfile(interface, app):
+        profile = interface.getGlobalDB().filter('func_stats', "function($f){return $f[\"app\"] == %d;}" % app)
+        total_time = 0 #sum of exclusive times is total time
+        fprofile = {}
+        for f in profile:
+            #Don't currently sum times but can reconstruct from count and mean
+            j = json.loads(f)
+            count = int(j['exclusive']['count'])
+            excl_mean = float(j['exclusive']['mean']) / 1e6   #seconds
+            excl_tot = count * excl_mean
+            fname = j['name']
+            fprofile[fname] = {}
+            fprofile[fname]['excl_time_tot'] = excl_tot
+            fprofile[fname]['count'] = count
+            fprofile[fname]['excl_time_mean'] = excl_mean
+            total_time += excl_tot
+
+
+        for f in fprofile.keys():
+            fprofile[f]['frac_time'] = fprofile[f]['excl_time_tot']/total_time
+
+        return fprofile
+
+#Print summary information for each function sorted by the total exclusive runtime spent in the function
+def summarizeProfile(fprofile):
+    fordered = sorted(fprofile.keys(), key=lambda x: fprofile[x]['excl_time_tot'], reverse=True)
+    for f in fordered:
+        finfo = fprofile[f]
+        print("(name='%s') (total excl. time=%es) (count=%d) (avg. excl. time=%es) (runtime fraction=%e)" % (f,finfo['excl_time_tot'],finfo['count'],finfo['excl_time_mean'],finfo['frac_time']) )
+
+
+    
 if __name__ == '__main__':
     argc = len(sys.argv)
     if(argc != 2):

scripts/provdb_interact.py

@@ -12,8 +12,9 @@
 from pysonata.client import SonataClient
 from pysonata.admin import SonataAdmin
 
-class provDBshard:
-    def __openCollection(self, coll_name, create=False):
+#Base class for a Sonata client connection to a database
+class provDBclientBase:
+    def _openCollection(self, coll_name, create=False):
         if(self.database.exists(coll_name) == False):
             if(create == True):
                 print("Shard %s creating '%s' collection" % (self.db_name, coll_name) )
@@ -23,16 +24,28 @@ def __openCollection(self, coll_name, create=False):
                 sys.exit(1)
         return self.database.open(coll_name)
 
+    #Apply a jx9 filter to a collection
+    def filter(self, which_coll, query):
+        return self.getCollection(which_coll).filter(query)
+
+    def execute(self, code, variables):
+        return self.database.execute(code,variables)
 
+    def store(self, which_coll, record):
+        return self.getCollection(which_coll).store(record, commit=True)
+
+
+#Sonata client connection to a database shard
+class provDBshard(provDBclientBase):
     def __init__(self,client,address,db_name,create=False):
         #Open database as a client
         self.db_name = db_name
         self.database = client.open(address, 0, db_name)
 
         #Initialize collections
-        self.anomalies = self.__openCollection('anomalies',create=create)
-        self.normalexecs = self.__openCollection('normalexecs',create=create)
-        self.metadata = self.__openCollection('metadata',create=create)
+        self.anomalies = self._openCollection('anomalies',create=create)
+        self.normalexecs = self._openCollection('normalexecs',create=create)
+        self.metadata = self._openCollection('metadata',create=create)
 
     def __del__(self):
         del self.anomalies
@@ -53,16 +66,34 @@ def getCollection(self, which_coll):
             sys.exit(1)
         return col
 
-    #Apply a jx9 filter to a collection
-    def filter(self, which_coll, query):
-        return self.getCollection(which_coll).filter(query)
 
-    def execute(self, code, variables):
-        return self.database.execute(code,variables)
 
-    def store(self, which_coll, record):
-        return self.getCollection(which_coll).store(record, commit=True)
-    
+class provDBglobal(provDBclientBase):
+    def __init__(self,client,address,db_name,create=False):
+        #Open database as a client
+        self.db_name = db_name
+        self.database = client.open(address, 0, db_name)
+
+        #Initialize collections
+        self.func_stats = self._openCollection('func_stats',create=create)
+        self.counter_stats = self._openCollection('counter_stats',create=create)
+
+    def __del__(self):
+        del self.func_stats
+        del self.counter_stats
+
+    def getCollection(self, which_coll):
+        col = None
+        if which_coll == "func_stats":
+            col = self.func_stats
+        elif which_coll == "counter_stats":
+            col = self.counter_stats
+        else:
+            print("Invalid collection")
+            sys.exit(1)
+        return col
+        
+
 
 
 class provDBinterface:
@@ -99,6 +130,27 @@ def __init__(self,engine,filename,nshards,create=False):
 
             self.db_shards.append( provDBshard(self.client, self.address, db_name, create=create) )
 
+        #Connect to global database if available
+        db_name = "provdb.global"
+        self.db_global_name = db_name
+        db_file = re.sub(r'\%d',"global",filename)
+        if os.path.exists(db_file) == False:
+            if create == True:
+                print("Creating global database as %s from file %s" % (db_name,db_file))
+                self.admin.create_database(self.address, 0, db_name, 'unqlite', "{ \"path\" : \"%s\" }" % db_file)
+                self.use_global_db = True                                
+            else:
+                self.use_global_db = False
+        else:
+            print("Attaching global database as %s from file %s" % (db_name,db_file))
+            self.admin.attach_database(self.address, 0, db_name, 'unqlite', "{ \"path\" : \"%s\" }" % db_file)
+            self.use_global_db = True
+
+        if self.use_global_db:
+            self.db_global = provDBglobal(self.client, self.address, db_name, create=create)
+        else:
+            self.db_global = None
+            
     def getNshards(self):
         return len(self.db_shards)
 
@@ -108,11 +160,20 @@ def getShard(self, i):
     def getShards(self):
         return self.db_shards
 
+    #Get the global database client. Returns None if the global database doesn't exit
+    def getGlobalDB(self):
+        return self.db_global
+    
     def __del__(self):
-        del self.db_shards
-        del self.client
+        del self.db_shards        
         for n in self.db_names:
             self.admin.detach_database(self.address, 0, n)
+            
+        if self.use_global_db:
+            del self.db_global
+            self.admin.detach_database(self.address, 0, self.db_global_name)
+
+        del self.client
         del self.admin
         del self.provider