Fix for Likwid collector (#95)

* Run LIKWID in separate thread and check metric type

* Change LIKWID collector documentation to use 'type' instead of 'scope'

* Re-initialize LIKWID after one read is missing due to lock toggle

collectors/likwidMetric.go

@@ -28,6 +28,7 @@ import (
 	lp "github.com/ClusterCockpit/cc-metric-collector/pkg/ccMetric"
 	topo "github.com/ClusterCockpit/cc-metric-collector/pkg/ccTopology"
 	"github.com/NVIDIA/go-nvml/pkg/dl"
+	"golang.design/x/thread"
 )
 
 const (
@@ -71,18 +72,20 @@ type LikwidCollectorConfig struct {
 
 type LikwidCollector struct {
 	metricCollector
-	cpulist      []C.int
+	cpulist       []C.int
-	cpu2tid      map[int]int
+	cpu2tid       map[int]int
-	sock2tid     map[int]int
+	sock2tid      map[int]int
-	metrics      map[C.int]map[string]int
+	metrics       map[C.int]map[string]int
-	groups       []C.int
+	groups        []C.int
-	config       LikwidCollectorConfig
+	config        LikwidCollectorConfig
-	gmresults    map[int]map[string]float64
+	gmresults     map[int]map[string]float64
-	basefreq     float64
+	basefreq      float64
-	running      bool
+	running       bool
-	initialized  bool
+	initialized   bool
-	likwidGroups map[C.int]LikwidEventsetConfig
+	needs_reinit bool
-	lock         sync.Mutex
+	likwidGroups  map[C.int]LikwidEventsetConfig
+	lock          sync.Mutex
+	measureThread thread.Thread
 }
 
 type LikwidMetric struct {
@@ -92,6 +95,18 @@ type LikwidMetric struct {
 	group_idx int
 }
 
+func checkMetricType(t string) bool {
+	valid := map[string]bool{
+		"node":         true,
+		"socket":       true,
+		"hwthread":     true,
+		"core":         true,
+		"memoryDomain": true,
+	}
+	_, ok := valid[t]
+	return ok
+}
+
 func eventsToEventStr(events map[string]string) string {
 	elist := make([]string, 0)
 	for k, v := range events {
@@ -179,6 +194,7 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
 	m.name = "LikwidCollector"
 	m.parallel = false
 	m.initialized = false
+	m.needs_reinit = true
 	m.running = false
 	m.config.AccessMode = LIKWID_DEF_ACCESSMODE
 	m.config.LibraryPath = LIKWID_LIB_NAME
@@ -239,7 +255,7 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
 			}
 			for _, metric := range evset.Metrics {
 				// Try to evaluate the metric
-				if testLikwidMetricFormula(metric.Calc, params) {
+				if testLikwidMetricFormula(metric.Calc, params) && checkMetricType(metric.Type) {
 					// Add the computable metric to the parameter list for the global metrics
 					globalParams = append(globalParams, metric.Name)
 					totalMetrics++
@@ -257,6 +273,9 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
 		if !testLikwidMetricFormula(metric.Calc, globalParams) {
 			cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed")
 			metric.Calc = ""
+		} else if !checkMetricType(metric.Type) {
+			cclog.ComponentError(m.name, "Metric", metric.Name, "has invalid type")
+			metric.Calc = ""
 		} else {
 			totalMetrics++
 		}
@@ -268,6 +287,7 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
 		cclog.ComponentError(m.name, err.Error())
 		return err
 	}
+	m.measureThread = thread.New()
 	m.init = true
 	return nil
 }
@@ -281,6 +301,7 @@ func (m *LikwidCollector) takeMeasurement(evset LikwidEventsetConfig, interval t
 		if ret != 0 {
 			var err error = nil
 			var skip bool = false
+			cclog.ComponentDebug(m.name, "Setup returns", ret)
 			if ret == -37 {
 				skip = true
 			} else {
@@ -289,6 +310,7 @@ func (m *LikwidCollector) takeMeasurement(evset LikwidEventsetConfig, interval t
 			m.lock.Unlock()
 			return skip, err
 		}
+		m.running = true
 		ret = C.perfmon_startCounters()
 		if ret != 0 {
 			var err error = nil
@@ -301,7 +323,7 @@ func (m *LikwidCollector) takeMeasurement(evset LikwidEventsetConfig, interval t
 			m.lock.Unlock()
 			return skip, err
 		}
-		m.running = true
+		ret = C.perfmon_readCounters()
 		time.Sleep(interval)
 		m.running = false
 		ret = C.perfmon_stopCounters()
@@ -316,6 +338,24 @@ func (m *LikwidCollector) takeMeasurement(evset LikwidEventsetConfig, interval t
 			m.lock.Unlock()
 			return skip, err
 		}
+		m.running = false
+		runtime := float64(C.perfmon_getLastTimeOfGroup(evset.gid))
+		// Go over events and get the results
+		for eidx, counter := range evset.eorder {
+			gctr := C.GoString(counter)
+			for _, tid := range m.cpu2tid {
+				res := C.perfmon_getLastResult(evset.gid, C.int(eidx), C.int(tid))
+				fres := float64(res)
+				if m.config.InvalidToZero && (math.IsNaN(fres) || math.IsInf(fres, 0)) {
+					cclog.ComponentDebug(m.name, "Sanitize", gctr, "to zero")
+					fres = 0.0
+				}
+				evset.results[tid][gctr] = fres
+			}
+		}
+		for _, tid := range m.cpu2tid {
+			evset.results[tid]["time"] = runtime
+		}
 	}
 	m.lock.Unlock()
 	return false, nil
@@ -325,19 +365,8 @@ func (m *LikwidCollector) takeMeasurement(evset LikwidEventsetConfig, interval t
 func (m *LikwidCollector) calcEventsetMetrics(evset LikwidEventsetConfig, interval time.Duration, output chan lp.CCMetric) error {
 	invClock := float64(1.0 / m.basefreq)
 
-	// Go over events and get the results
+	for _, tid := range m.cpu2tid {
-	for eidx, counter := range evset.eorder {
+		evset.results[tid]["inverseClock"] = invClock
-		gctr := C.GoString(counter)
-		for _, tid := range m.cpu2tid {
-			res := C.perfmon_getLastResult(evset.gid, C.int(eidx), C.int(tid))
-			fres := float64(res)
-			if m.config.InvalidToZero && (math.IsNaN(fres) || math.IsInf(fres, 0)) {
-				fres = 0.0
-			}
-			evset.results[tid][gctr] = fres
-			evset.results[tid]["time"] = interval.Seconds()
-			evset.results[tid]["inverseClock"] = invClock
-		}
 	}
 
 	// Go over the event set metrics, derive the value out of the event:counter values and send it
@@ -431,6 +460,28 @@ func (m *LikwidCollector) calcGlobalMetrics(interval time.Duration, output chan
 	return nil
 }
 
+func (m *LikwidCollector) ReInit() error {
+	C.perfmon_finalize()
+	ret := C.perfmon_init(C.int(len(m.cpulist)), &m.cpulist[0])
+	if ret != 0 {
+		return nil
+	}
+	for i, evset := range m.config.Eventsets {
+		var gid C.int
+		if len(evset.Events) > 0 {
+			//skip := false
+			likwidGroup := genLikwidEventSet(evset)
+			gid = C.perfmon_addEventSet(likwidGroup.estr)
+			if gid >= 0 {
+				likwidGroup.gid = gid
+				likwidGroup.internal = i
+				m.likwidGroups[gid] = likwidGroup
+			}
+		}
+	}
+	return nil
+}
+
 func (m *LikwidCollector) LateInit() error {
 	var ret C.int
 	if m.initialized {
@@ -445,6 +496,9 @@ func (m *LikwidCollector) LateInit() error {
 			os.Setenv("PATH", m.config.DaemonPath+":"+p)
 		}
 		C.HPMmode(1)
+		for _, c := range m.cpulist {
+			C.HPMaddThread(c)
+		}
 	}
 	cclog.ComponentDebug(m.name, "initialize LIKWID topology")
 	ret = C.topology_init()
@@ -468,48 +522,53 @@ func (m *LikwidCollector) LateInit() error {
 	m.basefreq = getBaseFreq()
 	cclog.ComponentDebug(m.name, "BaseFreq", m.basefreq)
 
-	cclog.ComponentDebug(m.name, "initialize LIKWID perfmon module")
+	if m.needs_reinit {
-	ret = C.perfmon_init(C.int(len(m.cpulist)), &m.cpulist[0])
+		m.ReInit()
-	if ret != 0 {
+		m.needs_reinit = false
-		var err error = nil
-		C.topology_finalize()
-		if ret != -22 {
-			err = errors.New("failed to initialize LIKWID perfmon")
-			cclog.ComponentError(m.name, err.Error())
-		} else {
-			err = errors.New("access to LIKWID perfmon locked")
-		}
-		return err
 	}
 
-	// While adding the events, we test the metrics whether they can be computed at all
+	// cclog.ComponentDebug(m.name, "initialize LIKWID perfmon module")
-	for i, evset := range m.config.Eventsets {
+	// ret = C.perfmon_init(C.int(len(m.cpulist)), &m.cpulist[0])
-		var gid C.int
+	// if ret != 0 {
-		if len(evset.Events) > 0 {
+	// 	var err error = nil
-			skip := false
+	// 	C.topology_finalize()
-			likwidGroup := genLikwidEventSet(evset)
+	// 	if ret != -22 {
-			for _, g := range m.likwidGroups {
+	// 		err = errors.New("failed to initialize LIKWID perfmon")
-				if likwidGroup.go_estr == g.go_estr {
+	// 		cclog.ComponentError(m.name, err.Error())
-					skip = true
+	// 	} else {
-					break
+	// 		err = errors.New("access to LIKWID perfmon locked")
-				}
+	// 	}
-			}
+	// 	return err
-			if skip {
+	// }
-				continue
-			}
-			// Now we add the list of events to likwid
-			gid = C.perfmon_addEventSet(likwidGroup.estr)
-			if gid >= 0 {
-				likwidGroup.gid = gid
-				likwidGroup.internal = i
-				m.likwidGroups[gid] = likwidGroup
-			}
-		} else {
-			cclog.ComponentError(m.name, "Invalid Likwid eventset config, no events given")
-			continue
-		}
 
-	}
+	// // While adding the events, we test the metrics whether they can be computed at all
+	// for i, evset := range m.config.Eventsets {
+	// 	var gid C.int
+	// 	if len(evset.Events) > 0 {
+	// 		//skip := false
+	// 		likwidGroup := genLikwidEventSet(evset)
+	// 		// for _, g := range m.likwidGroups {
+	// 		// 	if likwidGroup.go_estr == g.go_estr {
+	// 		// 		skip = true
+	// 		// 		break
+	// 		// 	}
+	// 		// }
+	// 		// if skip {
+	// 		// 	continue
+	// 		// }
+	// 		// Now we add the list of events to likwid
+	// 		gid = C.perfmon_addEventSet(likwidGroup.estr)
+	// 		if gid >= 0 {
+	// 			likwidGroup.gid = gid
+	// 			likwidGroup.internal = i
+	// 			m.likwidGroups[gid] = likwidGroup
+	// 		}
+	// 	} else {
+	// 		cclog.ComponentError(m.name, "Invalid Likwid eventset config, no events given")
+	// 		continue
+	// 	}
+
+	// }
 
 	// If no event set could be added, shut down LikwidCollector
 	if len(m.likwidGroups) == 0 {
@@ -540,38 +599,48 @@ func (m *LikwidCollector) Read(interval time.Duration, output chan lp.CCMetric)
 		return
 	}
 
-	if !m.initialized {
+	m.measureThread.Call(func() {
-		m.lock.Lock()
+		if !m.initialized {
-		err = m.LateInit()
+			m.lock.Lock()
-		if err != nil {
+			err = m.LateInit()
+			if err != nil {
+				m.lock.Unlock()
+				cclog.ComponentError(m.name, "lateinit failed")
+				return
+			}
+			m.initialized = true
 			m.lock.Unlock()
-			return
+			skip = true
 		}
-		m.initialized = true
-		m.lock.Unlock()
-	}
 
-	if m.initialized && !skip {
+		if m.initialized && !skip {
-		for _, evset := range m.likwidGroups {
+			time := interval
-			if !skip {
+			for _, evset := range m.likwidGroups {
-				// measure event set 'i' for 'interval' seconds
+				if !skip {
-				skip, err = m.takeMeasurement(evset, interval)
+					// measure event set 'i' for 'interval' seconds
-				if err != nil {
+					skip, err = m.takeMeasurement(evset, interval)
-					cclog.ComponentError(m.name, err.Error())
+					if err != nil {
-					return
+						cclog.ComponentError(m.name, err.Error())
+						return
+					}
+				}
+
+				if !skip {
+					// read measurements and derive event set metrics
+					m.calcEventsetMetrics(evset, time, output)
 				}
 			}
 
 			if !skip {
-				// read measurements and derive event set metrics
+				// use the event set metrics to derive the global metrics
-				m.calcEventsetMetrics(evset, interval, output)
+				m.calcGlobalMetrics(time, output)
+			}
+			if skip {
+				m.needs_reinit = true
+				m.initialized = false
 			}
 		}
-		if !skip {
+	})
-			// use the event set metrics to derive the global metrics
-			m.calcGlobalMetrics(interval, output)
-		}
-	}
 }
 
 func (m *LikwidCollector) Close() {

collectors/likwidMetric.md

@@ -41,7 +41,7 @@ The `likwid` collector is probably the most complicated collector. The LIKWID li
 
 The `likwid` configuration consists of two parts, the `eventsets` and `globalmetrics`:
 - An event set list itself has two parts, the `events` and a set of derivable `metrics`. Each of the `events` is a `counter:event` pair in LIKWID's syntax. The `metrics` are a list of formulas to derive the metric value from the measurements of the `events`' values. Each metric has a name, the formula, a type and a publish flag. There is an optional `unit` field. Counter names can be used like variables in the formulas, so `PMC0+PMC1` sums the measurements for the both events configured in the counters `PMC0` and `PMC1`. You can optionally use `time` for the measurement time and `inverseClock` for `1.0/baseCpuFrequency`. The type tells the LikwidCollector whether it is a metric for each hardware thread (`cpu`) or each CPU socket (`socket`). You may specify a unit for the metric with `unit`. The last one is the publishing flag. It tells the LikwidCollector whether a metric should be sent to the router or is only used internally to compute a global metric.
-- The `globalmetrics` are metrics which require data from multiple event set measurements to be derived. The inputs are the metrics in the event sets. Similar to the metrics in the event sets, the global metrics are defined by a name, a formula, a scope and a publish flag. See event set metrics for details. The only difference is that there is no access to the raw event measurements anymore but only to the metrics. Also `time` and `inverseClock` cannot be used anymore. So, the idea is to derive a metric in the `eventsets` section and reuse it in the `globalmetrics` part. If you need a metric only for deriving the global metrics, disable forwarding of the event set metrics (`"publish": false`). **Be aware** that the combination might be misleading because the "behavior" of a metric changes over time and the multiple measurements might count different computing phases. Similar to the metrics in the eventset, you can specify a metric unit with the `unit` field.
+- The `globalmetrics` are metrics which require data from multiple event set measurements to be derived. The inputs are the metrics in the event sets. Similar to the metrics in the event sets, the global metrics are defined by a name, a formula, a type and a publish flag. See event set metrics for details. The only difference is that there is no access to the raw event measurements anymore but only to the metrics. Also `time` and `inverseClock` cannot be used anymore. So, the idea is to derive a metric in the `eventsets` section and reuse it in the `globalmetrics` part. If you need a metric only for deriving the global metrics, disable forwarding of the event set metrics (`"publish": false`). **Be aware** that the combination might be misleading because the "behavior" of a metric changes over time and the multiple measurements might count different computing phases. Similar to the metrics in the eventset, you can specify a metric unit with the `unit` field.
 
 Additional options:
 - `force_overwrite`: Same as setting `LIKWID_FORCE=1`. In case counters are already in-use, LIKWID overwrites their configuration to do its measurements
@@ -50,20 +50,20 @@ Additional options:
 - `accessdaemon_path`: Folder of the accessDaemon `likwid-accessD` (like `/usr/local/sbin`)
 - `liblikwid_path`: Location of `liblikwid.so` including file name like `/usr/local/lib/liblikwid.so`
 
-### Available metric scopes
+### Available metric types
 
 Hardware performance counters are scattered all over the system nowadays. A counter coveres a specific part of the system. While there are hardware thread specific counter for CPU cycles, instructions and so on, some others are specific for a whole CPU socket/package. To address that, the LikwidCollector provides the specification of a `type` for each metric.
 
 - `hwthread` : One metric per CPU hardware thread with the tags `"type" : "hwthread"` and `"type-id" : "$hwthread_id"`
 - `socket` : One metric per CPU socket/package with the tags `"type" : "socket"` and `"type-id" : "$socket_id"`
 
-**Note:** You cannot specify `socket` scope for a metric that is measured at `hwthread` scope, so some kind of expert knowledge or lookup work in the [Likwid Wiki](https://github.com/RRZE-HPC/likwid/wiki) is required. Get the scope of each counter from the *Architecture* pages and as soon as one counter in a metric is socket-specific, the whole metric is socket-specific.
+**Note:** You cannot specify `socket` type for a metric that is measured at `hwthread` type, so some kind of expert knowledge or lookup work in the [Likwid Wiki](https://github.com/RRZE-HPC/likwid/wiki) is required. Get the type of each counter from the *Architecture* pages and as soon as one counter in a metric is socket-specific, the whole metric is socket-specific.
 
 As a guideline:
-- All counters `FIXCx`, `PMCy` and `TMAz` have the scope `hwthread`
+- All counters `FIXCx`, `PMCy` and `TMAz` have the type `hwthread`
-- All counters names containing `BOX` have the scope `socket`
+- All counters names containing `BOX` have the type `socket`
-- All `PWRx` counters have scope `socket`, except `"PWR1" : "RAPL_CORE_ENERGY"` has `hwthread` scope
+- All `PWRx` counters have type `socket`, except `"PWR1" : "RAPL_CORE_ENERGY"` has `hwthread` type
-- All `DFCx` counters have scope `socket`
+- All `DFCx` counters have type `socket`
 
 ### Help with the configuration
 
@@ -93,7 +93,7 @@ $ scripts/likwid_perfgroup_to_cc_config.py ICX MEM_DP
       "name": "Runtime (RDTSC) [s]",
       "publish": true,
       "unit": "seconds"
-      "scope": "hwthread"
+      "type": "hwthread"
     },
     {
       "..." : "..."
@@ -245,7 +245,7 @@ METRICS                          ->   "metrics": [
 IPC   PMC0/PMC1                  ->     {
                                  ->       "name" : "IPC",
                                  ->       "calc" : "PMC0/PMC1",
-                                 ->       "scope": "hwthread",
+                                 ->       "type": "hwthread",
                                  ->       "publish": true
                                  ->     }
                                  ->   ]