Version 0.1.2

Author: alexander-yakushev

README.md

@@ -35,12 +35,11 @@ The first argument to MeterThread's constructor is a LongConsumer callback that
 will be called with every new measurement of allocation rate in bytes/sec.
 
 The second (optional) argument is the measurement interval in milliseconds (1000
-by default). Note that the callback might not be called on each interval,
-because we can reliably measure the allocation rate only if the garbage
-collection hasn't been performed between two iterations. The default interval
-should work fine most of the time; unless your allocation rate is so high that
-the GC is triggered every second or two — in that case, in order not to lose
-most of the measurements, try to set the interval to a lower value.
+by default). Regardless of the interval length, the reported rate will be
+normalized to per-second value. Note that the callback might not be called on
+each interval, because under certain conditions we can't reliably measure the
+allocation rate (if both the garbage collection happened and a few of the
+heavy-allocating threads died).
 
 Here's an example of combining jvm-alloc-rate-meter with [Dropwizard
 Metrics](https://github.com/dropwizard/metrics):
@@ -54,7 +53,7 @@ import java.util.concurrent.TimeUnit;
 // ...
 
 Histogram hist = new Histogram(new SlidingTimeWindowArrayReservoir(10, TimeUnit.SECONDS));
-MeterThread mt = new MeterThread((r) -> hist.update(r));
+MeterThread mt = new MeterThread(hist::update);
 mt.start()
 
 // Now, you can forward this histogram to Graphite, or check the values manually, e.g.:

boot.properties

@@ -0,0 +1,5 @@
+#http://boot-clj.com
+#Tue Jun 26 19:08:05 EEST 2018
+BOOT_CLOJURE_NAME=org.clojure/clojure
+BOOT_CLOJURE_VERSION=1.10.0
+BOOT_VERSION=2.8.2

build.boot

@@ -1,6 +1,6 @@
 (task-options!
  pom {:project     'com.clojure-goes-fast/jvm-alloc-rate-meter
-      :version     "0.1.1"
+      :version     "0.1.2"
       :description "Measure JVM heap allocation rate in real time"
       :url         "http://github.com/clojure-goes-fast/jvm-alloc-rate-meter"
       :scm         {:url "http://github.com/clojure-goes-fast/jvm-alloc-rate-meter"}

src/jvm_alloc_rate_meter/MeterThread.java

@@ -1,5 +1,6 @@
 package jvm_alloc_rate_meter;
 
+import com.sun.management.ThreadMXBean;
 import java.util.function.LongConsumer;
 import java.lang.management.GarbageCollectorMXBean;
 import java.lang.management.ManagementFactory;
@@ -22,26 +23,48 @@ public MeterThread(LongConsumer callback, int intervalMs) {
         setDaemon(true);
     }
 
+    // Basically, we have two ways of measuring the allocation rate.
+    // First relies on checking heap usage between two points of time and
+    // substracting. This is accurate, but works only if the GC didn't trigger
+    // in the meantime.
+    // Second works by taking allocation stats by each alive thread. This is
+    // more reliable, but potentially less accurate (because threads can go
+    // away, and we lose their allocation stats).
+    // The idea is to use the first approach if GC didn't happen, and the second
+    // one if it did.
+
     public void run() {
-        long lastTime = 0;
+        long lastTime = 0, lastHeapUsage = -1, lastGcCounts = -1, lastThreadAllocated = -1;
         try {
-            long prevUsage = -1, prevGcCounts = -1;
             while (doRun) {
-                long usage = usedHeap();
+                long heapUsage = usedHeap();
                 long gcCounts = gcCounts();
-                long ts = System.currentTimeMillis();
+                long threadAllocated = allocatedByAllThreads();
+                long time = System.currentTimeMillis();
+
+                double multiplier = 1000.0 / (time - lastTime);
+                long deltaUsage = heapUsage - lastHeapUsage;
+                long deltaThreadAllocated = threadAllocated - lastThreadAllocated;
 
-                if ((gcCounts == prevGcCounts) && (usage >= prevUsage)) {
-                    long deltaTime = ts - lastTime;
-                    long rate = Math.round((usage - prevUsage) * (1000.0 / deltaTime));
-                    callback.accept(rate);
+                if (lastTime != 0) {
+                    if ((gcCounts == lastGcCounts) && (deltaUsage >= 0)) {
+                        long rate = Math.round(deltaUsage * multiplier);
+                        callback.accept(rate);
+                    } else if (deltaThreadAllocated >= 0) {
+                        long rate = Math.round(deltaThreadAllocated * multiplier);
+                        callback.accept(rate);
+                    } else {
+                        // Apparently, neither approach did well, just skip this
+                        // iteration.
+                    }
                 }
 
                 Thread.sleep(intervalMs);
 
-                prevUsage = usage;
-                prevGcCounts = gcCounts;
-                lastTime = ts;
+                lastTime = time;
+                lastHeapUsage = heapUsage;
+                lastGcCounts = gcCounts;
+                lastThreadAllocated = threadAllocated;
             }
         } catch (InterruptedException e) {
             System.err.println("MeterThread terminating...");
@@ -64,4 +87,16 @@ private static long gcCounts() {
         }
         return total;
     }
+
+    private static long allocatedByAllThreads() {
+        ThreadMXBean bean = (ThreadMXBean)ManagementFactory.getThreadMXBean();
+        long[] ids = bean.getAllThreadIds();
+        long[] allocatedBytes = bean.getThreadAllocatedBytes(ids);
+        long result = 0;
+        // This is not correct because we will lose allocation data from threads
+        // that died. Oh well.
+        for (long abytes : allocatedBytes)
+            result += abytes;
+        return result;
+    }
 }

src/jvm_alloc_rate_meter/core.clj

@@ -13,7 +13,7 @@
   (let [iw (MeterThread. (reify LongConsumer
                            (accept [_ v]
                              (callback-fn v)))
-                      (or interval-ms 1000))]
+                         (or interval-ms 1000))]
     (.start iw)
     #(.terminate iw)))