HDFS-15975. Use LongAdder instead of AtomicLong

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableCounterLong.java

@@ -23,7 +23,7 @@
 import org.apache.hadoop.metrics2.MetricsInfo;
 import org.apache.hadoop.metrics2.MetricsRecordBuilder;
 
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
 
 /**
  * A mutable long counter
@@ -32,11 +32,11 @@
 @InterfaceStability.Evolving
 public class MutableCounterLong extends MutableCounter {
 
-  private AtomicLong value = new AtomicLong();
+  private final LongAdder value = new LongAdder();
 
   public MutableCounterLong(MetricsInfo info, long initValue) {
     super(info);
-    this.value.set(initValue);
+    this.value.add(initValue);
   }
 
   @Override
@@ -49,12 +49,12 @@ public void incr() {
    * @param delta of the increment
    */
   public void incr(long delta) {
-    value.addAndGet(delta);
+    value.add(delta);
     setChanged();
   }
 
   public long value() {
-    return value.get();
+    return value.longValue();
   }
 
   @Override

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSHedgedReadMetrics.java

@@ -19,7 +19,7 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
 
 /**
  * The client-side metrics for hedged read feature.
@@ -28,20 +28,20 @@
  */
 @InterfaceAudience.Private
 public class DFSHedgedReadMetrics {
-  public final AtomicLong hedgedReadOps = new AtomicLong();
-  public final AtomicLong hedgedReadOpsWin = new AtomicLong();
-  public final AtomicLong hedgedReadOpsInCurThread = new AtomicLong();
+  public final LongAdder hedgedReadOps = new LongAdder();
+  public final LongAdder hedgedReadOpsWin = new LongAdder();
+  public final LongAdder hedgedReadOpsInCurThread = new LongAdder();
 
   public void incHedgedReadOps() {
-    hedgedReadOps.incrementAndGet();
+    hedgedReadOps.increment();
   }
 
   public void incHedgedReadOpsInCurThread() {
-    hedgedReadOpsInCurThread.incrementAndGet();
+    hedgedReadOpsInCurThread.increment();
   }
 
   public void incHedgedReadWins() {
-    hedgedReadOpsWin.incrementAndGet();
+    hedgedReadOpsWin.increment();
   }
 
   public long getHedgedReadOps() {

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOpsCountStatistics.java

@@ -26,7 +26,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.NoSuchElementException;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
 
 /**
  * This storage statistics tracks how many times each DFS operation was issued.
@@ -141,21 +141,21 @@ public static OpType fromSymbol(String symbol) {
 
   public static final String NAME = "DFSOpsCountStatistics";
 
-  private final Map<OpType, AtomicLong> opsCount = new EnumMap<>(OpType.class);
+  private final Map<OpType, LongAdder> opsCount = new EnumMap<>(OpType.class);
 
   public DFSOpsCountStatistics() {
     super(NAME);
     for (OpType opType : OpType.values()) {
-      opsCount.put(opType, new AtomicLong(0));
+      opsCount.put(opType, new LongAdder());
     }
   }
 
   public void incrementOpCounter(OpType op) {
-    opsCount.get(op).addAndGet(1);
+    opsCount.get(op).increment();
   }
 
   private class LongIterator implements Iterator<LongStatistic> {
-    private Iterator<Entry<OpType, AtomicLong>> iterator =
+    private final Iterator<Entry<OpType, LongAdder>> iterator =
         opsCount.entrySet().iterator();
 
     @Override
@@ -168,9 +168,9 @@ public LongStatistic next() {
       if (!iterator.hasNext()) {
         throw new NoSuchElementException();
       }
-      final Entry<OpType, AtomicLong> entry = iterator.next();
+      final Entry<OpType, LongAdder> entry = iterator.next();
       return new LongStatistic(entry.getKey().getSymbol(),
-          entry.getValue().get());
+          entry.getValue().longValue());
     }
 
     @Override
@@ -192,7 +192,7 @@ public Iterator<LongStatistic> getLongStatistics() {
   @Override
   public Long getLong(String key) {
     final OpType type = OpType.fromSymbol(key);
-    return type == null ? null : opsCount.get(type).get();
+    return type == null ? null : opsCount.get(type).longValue();
   }
 
   @Override
@@ -202,8 +202,8 @@ public boolean isTracked(String key) {
 
   @Override
   public void reset() {
-    for (AtomicLong count : opsCount.values()) {
-      count.set(0);
+    for (LongAdder count : opsCount.values()) {
+      count.reset();
     }
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetCache.java

@@ -43,6 +43,7 @@
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
 
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang3.time.DurationFormatUtils;
@@ -143,11 +144,11 @@ public boolean shouldAdvertise() {
   /**
    * Number of cache commands that could not be completed successfully
    */
-  final AtomicLong numBlocksFailedToCache = new AtomicLong(0);
+  final LongAdder numBlocksFailedToCache = new LongAdder();
   /**
    * Number of uncache commands that could not be completed successfully
    */
-  final AtomicLong numBlocksFailedToUncache = new AtomicLong(0);
+  final LongAdder numBlocksFailedToUncache = new LongAdder();
 
   public FsDatasetCache(FsDatasetImpl dataset) throws IOException {
     this.dataset = dataset;
@@ -278,7 +279,7 @@ synchronized void cacheBlock(long blockId, String bpid,
       LOG.debug("Block with id {}, pool {} already exists in the "
               + "FsDatasetCache with state {}", blockId, bpid, prevValue.state
       );
-      numBlocksFailedToCache.incrementAndGet();
+      numBlocksFailedToCache.increment();
       return;
     }
     mappableBlockMap.put(key, new Value(null, State.CACHING));
@@ -301,7 +302,7 @@ synchronized void uncacheBlock(String bpid, long blockId) {
       LOG.debug("Block with id {}, pool {} does not need to be uncached, "
           + "because it is not currently in the mappableBlockMap.", blockId,
           bpid);
-      numBlocksFailedToUncache.incrementAndGet();
+      numBlocksFailedToUncache.increment();
       return;
     }
     switch (prevValue.state) {
@@ -331,7 +332,7 @@ synchronized void uncacheBlock(String bpid, long blockId) {
     default:
       LOG.debug("Block with id {}, pool {} does not need to be uncached, "
           + "because it is in state {}.", blockId, bpid, prevValue.state);
-      numBlocksFailedToUncache.incrementAndGet();
+      numBlocksFailedToUncache.increment();
       break;
     }
   }
@@ -482,7 +483,7 @@ public void run() {
           LOG.debug("Caching of {} was aborted.  We are now caching only {} "
                   + "bytes in total.", key, cacheLoader.getCacheUsed());
           IOUtils.closeQuietly(mappableBlock);
-          numBlocksFailedToCache.incrementAndGet();
+          numBlocksFailedToCache.increment();
 
           synchronized (FsDatasetCache.this) {
             mappableBlockMap.remove(key);
@@ -607,11 +608,11 @@ public long getCacheCapacity() {
   }
 
   public long getNumBlocksFailedToCache() {
-    return numBlocksFailedToCache.get();
+    return numBlocksFailedToCache.longValue();
   }
 
   public long getNumBlocksFailedToUncache() {
-    return numBlocksFailedToUncache.get();
+    return numBlocksFailedToUncache.longValue();
   }
 
   public long getNumBlocksCached() {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java

@@ -2463,7 +2463,7 @@ private void cacheBlock(String bpid, long blockId) {
         success = true;
       } finally {
         if (!success) {
-          cacheManager.numBlocksFailedToCache.incrementAndGet();
+          cacheManager.numBlocksFailedToCache.increment();
         }
       }
       blockFileName = info.getBlockURI().toString();

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java

@@ -28,7 +28,7 @@
 import java.util.EnumSet;
 import java.util.Iterator;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
@@ -183,7 +183,7 @@ private enum State {
 
   // these are statistics counters.
   private long numTransactions;        // number of transactions
-  private final AtomicLong numTransactionsBatchedInSync = new AtomicLong();
+  private final LongAdder numTransactionsBatchedInSync = new LongAdder();
   private long totalTimeTransactions;  // total time for all transactions
   private NameNodeMetrics metrics;
 
@@ -731,7 +731,7 @@ protected void logSync(long mytxid) {
       if (metrics != null) { // Metrics non-null only when used inside name node
         metrics.addSync(elapsed);
         metrics.incrTransactionsBatchedInSync(editsBatchedInSync);
-        numTransactionsBatchedInSync.addAndGet(editsBatchedInSync);
+        numTransactionsBatchedInSync.add(editsBatchedInSync);
       }
 
     } finally {
@@ -771,7 +771,7 @@ private void printStatistics(boolean force) {
         .append(" Total time for transactions(ms): ")
         .append(totalTimeTransactions)
         .append(" Number of transactions batched in Syncs: ")
-        .append(numTransactionsBatchedInSync.get())
+        .append(numTransactionsBatchedInSync.longValue())
         .append(" Number of syncs: ")
         .append(editLogStream.getNumSync())
         .append(" SyncTimes(ms): ")
@@ -1404,7 +1404,7 @@ private void startLogSegment(final long segmentTxId, int layoutVersion)
 
     numTransactions = 0;
     totalTimeTransactions = 0;
-    numTransactionsBatchedInSync.set(0L);
+    numTransactionsBatchedInSync.reset();
 
     // TODO no need to link this back to storage anymore!
     // See HDFS-2174.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystemLock.java

@@ -22,6 +22,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Supplier;
@@ -113,12 +114,12 @@ public Long initialValue() {
    * The number of time the read lock
    * has been held longer than the threshold.
    */
-  private final AtomicLong numReadLockLongHold = new AtomicLong(0);
+  private final LongAdder numReadLockLongHold = new LongAdder();
   /**
    * The number of time the write lock
    * has been held for longer than the threshold.
    */
-  private final AtomicLong numWriteLockLongHold = new AtomicLong(0);
+  private final LongAdder numWriteLockLongHold = new LongAdder();
 
   @VisibleForTesting
   static final String OP_NAME_OTHER = "OTHER";
@@ -192,7 +193,7 @@ public void readUnlock(String opName,
     final long readLockIntervalMs =
         TimeUnit.NANOSECONDS.toMillis(readLockIntervalNanos);
     if (needReport && readLockIntervalMs >= this.readLockReportingThresholdMs) {
-      numReadLockLongHold.incrementAndGet();
+      numReadLockLongHold.increment();
       String lockReportInfo = null;
       boolean done = false;
       while (!done) {
@@ -309,7 +310,7 @@ private void writeUnlock(String opName, boolean suppressWriteLockReport,
     LogAction logAction = LogThrottlingHelper.DO_NOT_LOG;
     if (needReport &&
         writeLockIntervalMs >= this.writeLockReportingThresholdMs) {
-      numWriteLockLongHold.incrementAndGet();
+      numWriteLockLongHold.increment();
       if (longestWriteLockHeldInfo.getIntervalMs() <= writeLockIntervalMs) {
         String lockReportInfo = lockReportInfoSupplier != null ? " (" +
             lockReportInfoSupplier.get() + ")" : "";
@@ -382,7 +383,7 @@ public int getQueueLength() {
    * has been held longer than the threshold
    */
   public long getNumOfReadLockLongHold() {
-    return numReadLockLongHold.get();
+    return numReadLockLongHold.longValue();
   }
 
   /**
@@ -393,7 +394,7 @@ public long getNumOfReadLockLongHold() {
    * has been held longer than the threshold.
    */
   public long getNumOfWriteLockLongHold() {
-    return numWriteLockLongHold.get();
+    return numWriteLockLongHold.longValue();
   }
 
   /**

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestPread.java

@@ -401,9 +401,9 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
     DFSClient dfsClient = fileSys.getClient();
     DFSHedgedReadMetrics metrics = dfsClient.getHedgedReadMetrics();
     // Metrics instance is static, so we need to reset counts from prior tests.
-    metrics.hedgedReadOps.set(0);
-    metrics.hedgedReadOpsWin.set(0);
-    metrics.hedgedReadOpsInCurThread.set(0);
+    metrics.hedgedReadOps.reset();
+    metrics.hedgedReadOpsWin.reset();
+    metrics.hedgedReadOpsInCurThread.reset();
 
     try {
       Path file1 = new Path("hedgedReadMaxOut.dat");
@@ -590,7 +590,7 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
     String filename = "/hedgedReadMaxOut.dat";
     DFSHedgedReadMetrics metrics = dfsClient.getHedgedReadMetrics();
     // Metrics instance is static, so we need to reset counts from prior tests.
-    metrics.hedgedReadOps.set(0);
+    metrics.hedgedReadOps.reset();
     try {
       Path file = new Path(filename);
       output = fileSys.create(file, (short) 2);