Fixes calculation of average thread blocked time and average thread waited time (#118)

* Enables thread contention monitoring. Fixes calculation of average thread blocked time and average thread waited time

Signed-off-by: Surya Sashank Nistala <[email protected]>

* check thread contention monitoring enabled/disabled via PA setting

Signed-off-by: Surya Sashank Nistala <[email protected]>

* remove test generated files

Signed-off-by: Surya Sashank Nistala <[email protected]>

Author: eirsep

src/main/java/org/opensearch/performanceanalyzer/collectors/OSMetricsCollector.java

@@ -45,7 +45,8 @@ public void collectMetrics(long startTime) {
         SchedMetricsGenerator schedMetricsGenerator = osMetricsGenerator.getSchedMetricsGenerator();
         schedMetricsGenerator.addSample();
 
-        Map<Long, ThreadList.ThreadState> threadStates = ThreadList.getNativeTidMap();
+        Map<Long, ThreadList.ThreadState> threadStates =
+                ThreadList.getNativeTidMap(getThreadContentionMonitoringEnabled());
 
         DiskIOMetricsGenerator diskIOMetricsGenerator =
                 osMetricsGenerator.getDiskIOMetricsGenerator();

src/main/java/org/opensearch/performanceanalyzer/collectors/PerformanceAnalyzerMetricsCollector.java

@@ -33,6 +33,7 @@ enum State {
     private String collectorName;
     protected StringBuilder value;
     protected State state;
+    private boolean threadContentionMonitoringEnabled;
 
     protected PerformanceAnalyzerMetricsCollector(int timeInterval, String collectorName) {
         this.timeInterval = timeInterval;
@@ -92,4 +93,12 @@ public State getState() {
     public void setState(State state) {
         this.state = state;
     }
+
+    public void setThreadContentionMonitoringEnabled(boolean enabled) {
+        this.threadContentionMonitoringEnabled = enabled;
+    }
+
+    public boolean getThreadContentionMonitoringEnabled() {
+        return threadContentionMonitoringEnabled;
+    }
 }

src/main/java/org/opensearch/performanceanalyzer/collectors/ScheduledMetricCollectorsExecutor.java

@@ -25,6 +25,7 @@ public class ScheduledMetricCollectorsExecutor extends Thread {
     private static final int COLLECTOR_THREAD_KEEPALIVE_SECS = 1000;
     private final boolean checkFeatureDisabledFlag;
     private boolean paEnabled = false;
+    private boolean threadContentionMonitoringEnabled = false;
     private int minTimeIntervalToSleep = Integer.MAX_VALUE;
     private Map<PerformanceAnalyzerMetricsCollector, Long> metricsCollectors;
 
@@ -52,7 +53,19 @@ public synchronized boolean getEnabled() {
         return paEnabled;
     }
 
+    public synchronized void setThreadContentionMonitoringEnabled(final boolean enabled) {
+        metricsCollectors
+                .keySet()
+                .forEach(collector -> collector.setThreadContentionMonitoringEnabled(enabled));
+        threadContentionMonitoringEnabled = enabled;
+    }
+
+    private synchronized boolean getThreadContentionMonitoringEnabled() {
+        return threadContentionMonitoringEnabled;
+    }
+
     public void addScheduledMetricCollector(PerformanceAnalyzerMetricsCollector task) {
+        task.setThreadContentionMonitoringEnabled(getThreadContentionMonitoringEnabled());
         metricsCollectors.put(task, System.currentTimeMillis() + task.getTimeInterval());
         if (task.getTimeInterval() < minTimeIntervalToSleep) {
             minTimeIntervalToSleep = task.getTimeInterval();

src/main/java/org/opensearch/performanceanalyzer/jvm/ThreadList.java

@@ -117,8 +117,13 @@ public String toString() {
      * acquire this lock and move on if we could not get it.
      *
      * @return A hashmap of threadId to threadState.
+     * @param threadContentionMonitoringEnabled
      */
-    public static Map<Long, ThreadState> getNativeTidMap() {
+    public static Map<Long, ThreadState> getNativeTidMap(
+            boolean threadContentionMonitoringEnabled) {
+        if (threadBean.isThreadContentionMonitoringSupported()) {
+            threadBean.setThreadContentionMonitoringEnabled(threadContentionMonitoringEnabled);
+        }
         if (vmAttachLock.tryLock()) {
             try {
                 // Thread dumps are expensive and therefore we make sure that at least
@@ -264,6 +269,9 @@ private static void parseThreadInfo(final ThreadInfo info) {
                         1.0e-3
                                 * (t.blockedTime - oldt.blockedTime)
                                 / (t.blockedCount - oldt.blockedCount);
+            } else if (t.blockedCount == oldt.blockedCount && t.blockedTime > oldt.blockedTime) {
+                t.avgBlockedTime =
+                        1.0e-3 * (t.blockedTime - oldt.blockedTime + oldt.avgBlockedTime);
             } else {
                 CircularLongArray arr = ThreadHistory.blockedTidHistoryMap.get(t.nativeTid);
                 // NOTE: this is an upper bound
@@ -276,6 +284,8 @@ private static void parseThreadInfo(final ThreadInfo info) {
                         1.0e-3
                                 * (t.waitedTime - oldt.waitedTime)
                                 / (t.waitedCount - oldt.waitedCount);
+            } else if (t.waitedCount == oldt.waitedCount && t.waitedTime > oldt.waitedTime) {
+                t.avgWaitedTime = 1.0e-3 * (t.waitedTime - oldt.waitedTime + oldt.avgWaitedTime);
             } else {
                 CircularLongArray arr = ThreadHistory.waitedTidHistoryMap.get(t.nativeTid);
                 // NOTE: this is an upper bound

src/test/java/org/opensearch/performanceanalyzer/jvm/ThreadListTests.java

@@ -36,7 +36,7 @@ private static void runOnce() throws InterruptedException {
         String params[] = new String[0];
         while (true) {
             ThreadList.runThreadDump(OSGlobals.getPid(), params);
-            ThreadList.LOGGER.info(ThreadList.getNativeTidMap().values());
+            ThreadList.LOGGER.info(ThreadList.getNativeTidMap(false).values());
 
             /*GCMetrics.runOnce();
             HeapMetrics.runOnce();