Replace LongAdder in metric-like logic with ThreadLocalCounter

patch by Dmitry Konstantinov; reviewed by Stefan Miklosovic for CASSANDRA-21400

CHANGES.txt

@@ -1,4 +1,5 @@
 6.0-alpha2
+ * Replace LongAdder in metric-like logic with ThreadLocalCounter (CASSANDRA-21400)
  * BTreeRow.hasLiveData: avoid Cell iteration if there are no cell tombstones (CASSANDRA-21363)
  * Reduce memory allocations in row merge logic (CASSANDRA-21359)
  * Restore option to avoid hint transfer during decommission (CASSANDRA-21341)

src/java/org/apache/cassandra/io/sstable/filter/BloomFilterTracker.java

@@ -18,35 +18,36 @@
 package org.apache.cassandra.io.sstable.filter;
 
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.LongAdder;
+
+import org.apache.cassandra.metrics.ThreadLocalCounter;
 
 public class BloomFilterTracker
 {
-    private final LongAdder falsePositiveCount = new LongAdder();
-    private final LongAdder truePositiveCount = new LongAdder();
-    private final LongAdder trueNegativeCount = new LongAdder();
+    private final ThreadLocalCounter falsePositiveCount = new ThreadLocalCounter();
+    private final ThreadLocalCounter truePositiveCount = new ThreadLocalCounter();
+    private final ThreadLocalCounter trueNegativeCount = new ThreadLocalCounter();
     private final AtomicLong lastFalsePositiveCount = new AtomicLong();
     private final AtomicLong lastTruePositiveCount = new AtomicLong();
     private final AtomicLong lastTrueNegativeCount = new AtomicLong();
 
     public void addFalsePositive()
     {
-        falsePositiveCount.increment();
+        falsePositiveCount.inc();
     }
 
     public void addTruePositive()
     {
-        truePositiveCount.increment();
+        truePositiveCount.inc();
     }
 
     public void addTrueNegative()
     {
-        trueNegativeCount.increment();
+        trueNegativeCount.inc();
     }
 
     public long getFalsePositiveCount()
     {
-        return falsePositiveCount.sum();
+        return falsePositiveCount.getCount();
     }
 
     public long getRecentFalsePositiveCount()
@@ -58,7 +59,7 @@ public class BloomFilterTracker
 
     public long getTruePositiveCount()
     {
-        return truePositiveCount.sum();
+        return truePositiveCount.getCount();
     }
 
     public long getRecentTruePositiveCount()
@@ -70,7 +71,7 @@ public class BloomFilterTracker
 
     public long getTrueNegativeCount()
     {
-        return trueNegativeCount.sum();
+        return trueNegativeCount.getCount();
     }
 
     public long getRecentTrueNegativeCount()

src/java/org/apache/cassandra/io/sstable/keycache/KeyCache.java

@@ -18,8 +18,6 @@
 
 package org.apache.cassandra.io.sstable.keycache;
 
-import java.util.concurrent.atomic.LongAdder;
-
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
@@ -29,6 +27,7 @@ import org.slf4j.LoggerFactory;
 import org.apache.cassandra.cache.InstrumentingCache;
 import org.apache.cassandra.cache.KeyCacheKey;
 import org.apache.cassandra.io.sstable.AbstractRowIndexEntry;
+import org.apache.cassandra.metrics.ThreadLocalCounter;
 
 /**
  * A simple wrapper of possibly global cache with local metrics.
@@ -40,8 +39,8 @@ public class KeyCache
     private final static Logger logger = LoggerFactory.getLogger(KeyCache.class);
 
     private final InstrumentingCache<KeyCacheKey, AbstractRowIndexEntry> cache;
-    private final LongAdder hits = new LongAdder();
-    private final LongAdder requests = new LongAdder();
+    private final ThreadLocalCounter hits = new ThreadLocalCounter();
+    private final ThreadLocalCounter requests = new ThreadLocalCounter();
 
     public KeyCache(@Nullable InstrumentingCache<KeyCacheKey, AbstractRowIndexEntry> cache)
     {
@@ -50,12 +49,12 @@ public class KeyCache
 
     public long getHits()
     {
-        return cache != null ? hits.sum() : 0;
+        return cache != null ? hits.getCount() : 0;
     }
 
     public long getRequests()
     {
-        return cache != null ? requests.sum() : 0;
+        return cache != null ? requests.getCount() : 0;
     }
 
     public void put(@Nonnull KeyCacheKey cacheKey, @Nonnull AbstractRowIndexEntry info)
@@ -74,10 +73,10 @@ public class KeyCache
 
         if (updateStats)
         {
-            requests.increment();
+            requests.inc();
             AbstractRowIndexEntry r = cache.get(key);
             if (r != null)
-                hits.increment();
+                hits.inc();
             return r;
         }
         else

src/java/org/apache/cassandra/metrics/ThreadLocalCounter.java

@@ -73,4 +73,9 @@ public class ThreadLocalCounter extends com.codahale.metrics.Counter implements
     {
         return ThreadLocalMetrics.getCount(metricId);
     }
+
+    public void reset()
+    {
+        ThreadLocalMetrics.getCountAndReset(metricId);
+    }
 }

src/java/org/apache/cassandra/transport/ServerConnection.java

@@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.auth.IAuthenticator;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.metrics.ThreadLocalCounter;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.QueryState;
 
@@ -41,7 +42,7 @@ public class ServerConnection extends Connection
     private volatile IAuthenticator.SaslNegotiator saslNegotiator;
     private final ClientState clientState;
     private volatile ConnectionStage stage;
-    public final Counter requests = new Counter();
+    public final Counter requests = new ThreadLocalCounter();
 
     ServerConnection(Channel channel, ProtocolVersion version, Connection.Tracker tracker)
     {

src/java/org/apache/cassandra/utils/memory/BufferPool.java

@@ -33,7 +33,6 @@ import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
-import java.util.concurrent.atomic.LongAdder;
 import java.util.function.BiPredicate;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
@@ -48,6 +47,7 @@ import org.slf4j.LoggerFactory;
 import org.apache.cassandra.concurrent.Shutdownable;
 import org.apache.cassandra.io.compress.BufferType;
 import org.apache.cassandra.metrics.BufferPoolMetrics;
+import org.apache.cassandra.metrics.ThreadLocalCounter;
 import org.apache.cassandra.utils.NoSpamLogger;
 import org.apache.cassandra.utils.Shared;
 import org.apache.cassandra.utils.concurrent.Ref;
@@ -150,12 +150,12 @@ public class BufferPool
     /**
      * Size of unpooled buffer being allocated outside of buffer pool in bytes.
      */
-    private final LongAdder overflowMemoryUsage = new LongAdder();
+    private final ThreadLocalCounter overflowMemoryUsage = new ThreadLocalCounter();
 
     /**
      * Size of buffer being used in bytes, including pooled buffer and unpooled buffer.
      */
-    private final LongAdder memoryInUse = new LongAdder();
+    private final ThreadLocalCounter memoryInUse = new ThreadLocalCounter();
 
     /**
      * Size of allocated buffer pool slabs in bytes
@@ -264,7 +264,7 @@ public class BufferPool
 
     private void updateOverflowMemoryUsage(int size)
     {
-        overflowMemoryUsage.add(size);
+        overflowMemoryUsage.inc(size);
     }
 
     public void setRecycleWhenFreeForCurrentThread(boolean recycleWhenFree)
@@ -277,7 +277,7 @@ public class BufferPool
      */
     public long sizeInBytes()
     {
-        return memoryAllocated.get() + overflowMemoryUsage.longValue();
+        return memoryAllocated.get() + overflowMemoryUsage.getCount();
     }
 
     /**
@@ -285,7 +285,7 @@ public class BufferPool
      */
     public long usedSizeInBytes()
     {
-        return memoryInUse.longValue() + overflowMemoryUsage.longValue();
+        return memoryInUse.getCount() + overflowMemoryUsage.getCount();
     }
 
     /**
@@ -293,7 +293,7 @@ public class BufferPool
      */
     public long overflowMemoryInBytes()
     {
-        return overflowMemoryUsage.longValue();
+        return overflowMemoryUsage.getCount();
     }
 
     /**
@@ -825,7 +825,7 @@ public class BufferPool
             else
             {
                 put(buffer, chunk);
-                memoryInUse.add(-size);
+                memoryInUse.dec(size);
             }
         }
 
@@ -892,7 +892,7 @@ public class BufferPool
 
             chunk.freeUnusedPortion(buffer);
             // Calculate the actual freed bytes which may be different from `size` when pooling is involved
-            memoryInUse.add(buffer.capacity() - originalCapacity);
+            memoryInUse.inc(buffer.capacity() - originalCapacity);
         }
 
         public ByteBuffer get(int size)
@@ -951,7 +951,7 @@ public class BufferPool
             if (ret != null)
             {
                 metrics.hits.mark();
-                memoryInUse.add(ret.capacity());
+                memoryInUse.inc(ret.capacity());
             }
             else
             {