HADOOP-19712 S3A: Deadlock in EvaluatingStatisticsMap.entryset()

Reworked how entrySet() and values() work, using .forEach()
iterators after reviewing what ConcurrentHashMap does internally;
it does a (safe) traverse.

Add EvaluatingStatisticsMap.forEach() implementation which maps the passed
in BiConsumer down to the evaluators.forEach, evaluating each value as it goes.

Use that in IOStatisticsBinding.snapshot() code.

Tests for all this.

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/impl/EvaluatingStatisticsMap.java

@@ -19,12 +19,15 @@
 package org.apache.hadoop.fs.statistics.impl;
 
 import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiConsumer;
 import java.util.function.Function;
-import java.util.stream.Collectors;
 
 /**
  * A map of functions which can be invoked to dynamically
@@ -132,11 +135,10 @@ public Set<String> keySet() {
    */
   @Override
   public Collection<E> values() {
-    Set<Entry<String, Function<String, E>>> evalEntries =
-        evaluators.entrySet();
-    return evalEntries.parallelStream().map((e) ->
-        e.getValue().apply(e.getKey()))
-        .collect(Collectors.toList());
+    List<E> result = new ArrayList<>(size());
+    evaluators.forEach((k, f) ->
+        result.add(f.apply(k)));
+    return result;
   }
 
   /**
@@ -149,22 +151,37 @@ public Map<String, E> snapshot() {
 
   /**
    * Creating the entry set forces an evaluation of the functions.
-   *
+   * <p>
+   * Not synchronized, though thread safe.
+   * <p>
    * This is not a snapshot, so if the evaluators actually return
    * references to mutable objects (e.g. a MeanStatistic instance)
    * then that value may still change.
    *
-   * The evaluation may be parallelized.
    * @return an evaluated set of values
    */
   @Override
-  public synchronized Set<Entry<String, E>> entrySet() {
-    Set<Entry<String, Function<String, E>>> evalEntries =
-        evaluators.entrySet();
-    Set<Entry<String, E>> r = evalEntries.parallelStream().map((e) ->
-        new EntryImpl<>(e.getKey(), e.getValue().apply(e.getKey())))
-        .collect(Collectors.toSet());
-    return r;
+  public Set<Entry<String, E>> entrySet() {
+    Set<Entry<String, E>> result = new LinkedHashSet<>(size());
+    evaluators.forEach((key, evaluator) -> {
+      final E current = evaluator.apply(key);
+      result.add(new EntryImpl<>(key, current));
+    });
+    return result;
+  }
+
+
+  /**
+   * Hand down to the foreach iterator of the evaluators, by evaluating as each
+   * entry is processed and passing that in to the {@code action} consumer.
+   * @param action consumer of entries.
+   */
+  @Override
+  public void forEach(final BiConsumer<? super String, ? super E> action) {
+    BiConsumer<String, Function<String, E>> biConsumer = (key, value) -> {
+      action.accept(key, value.apply(key));
+    };
+    evaluators.forEach(biConsumer);
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/impl/IOStatisticsBinding.java

@@ -170,9 +170,8 @@ private static <E> Map<String, E> copyMap(
     // we have to clone the values so that they aren't
     // bound to the original values
     dest.clear();
-    source.entrySet()
-        .forEach(entry ->
-            dest.put(entry.getKey(), copyFn.apply(entry.getValue())));
+    source.forEach((key, current) ->
+            dest.put(key, copyFn.apply(current)));
     return dest;
   }
 

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/statistics/TestIOStatisticsStore.java

@@ -18,6 +18,12 @@
 
 package org.apache.hadoop.fs.statistics;
 
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -174,4 +180,59 @@ public void testNegativeCounterIncrementIgnored() throws Throwable {
         .isEqualTo(2);
   }
 
+  @Test
+  public void testForeach() throws Throwable {
+
+    final IOStatisticsStore store = iostatisticsStore()
+        .withCounters(COUNT, "c1", "c2")
+        .withGauges(GAUGE)
+        .withMinimums(MIN)
+        .withMaximums(MAX)
+        .withMeanStatistics(MEAN)
+        .build();
+    store.setCounter(COUNT, 10);
+    store.setCounter("c1", 1);
+    store.setCounter("c2", 2);
+
+    // get the counter map, which is evaluated on demand
+    final Map<String, Long> counters = store.counters();
+    LongAdder entryCount = new LongAdder();
+    LongAdder sum = new LongAdder();
+
+    // apply the foreach iteration
+    counters.forEach((k, v) -> {
+      entryCount.increment();
+      sum.add(v);
+    });
+    Assertions.assertThat(entryCount.longValue())
+        .describedAs("entry count")
+        .isEqualTo(3);
+    Assertions.assertThat(sum.longValue())
+        .describedAs("sum of values")
+        .isEqualTo(13);
+
+    // keyset is as expected
+    final Set<String> keys = counters.keySet();
+    Assertions.assertThat(keys)
+        .describedAs("keys")
+        .hasSize(3)
+        .contains("c1", "c2", COUNT);
+
+    // values are as expected
+    final Collection<Long> values = counters.values();
+    Assertions.assertThat(values)
+        .describedAs("values")
+        .hasSize(3)
+        .contains(10L, 1L, 2L);
+
+    // entries will all be evaluated
+    final Set<Map.Entry<String, Long>> entries = counters.entrySet();
+    entryCount.reset();
+    sum.reset();
+    entries.forEach(e -> {
+          entryCount.increment();
+          sum.add(e.getValue());
+        });
+  }
+
 }