cc

Author: fandreuz

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorage.java

@@ -55,6 +55,9 @@ public final class AsynchronousMetricStorage<T extends PointData, U extends Exem
   private final AggregationTemporality aggregationTemporality;
   private final AttributesProcessor attributesProcessor;
   private final MemoryMode memoryMode;
+  private final Aggregator<T, U> originalAggregator;
+
+  private Aggregator<T, U> aggregator;
 
   /**
    * This field is set to 1 less than the actual intended cardinality limit, allowing the last slot
@@ -70,19 +73,16 @@ public final class AsynchronousMetricStorage<T extends PointData, U extends Exem
 
   // Only populated if memoryMode == REUSABLE_DATA
   private final ObjectPool<T> reusablePointsPool;
+  private final ObjectPool<AggregatorHandle<T, U>> reusableHandlesPool;
   private final Function<Attributes, AggregatorHandle<T, U>> handleBuilder;
+  private final BiConsumer<Attributes, AggregatorHandle<T, U>> handleReleaser;
   private final BiConsumer<Attributes, T> pointReleaser;
 
   private final List<T> reusablePointsList = new ArrayList<>();
   // If aggregationTemporality == DELTA, this reference and lastPoints will be swapped at every
   // collection
   private Map<Attributes, T> reusablePointsMap = new PooledHashMap<>();
 
-  // deliberately not volatile because of performance concerns
-  // - which means its eventually consistent
-  private AggregatorHolder<T, U> aggregatorHolder;
-  private final Aggregator<T, U> originalAggregator;
-
   // Time information relative to recording of data in aggregatorHandles, set while calling
   // callbacks
   private long startEpochNanos;
@@ -101,15 +101,16 @@ private AsynchronousMetricStorage(
             .getReader()
             .getAggregationTemporality(metricDescriptor.getSourceInstrument().getType());
     this.memoryMode = registeredReader.getReader().getMemoryMode();
+    this.aggregator = aggregator;
+    this.originalAggregator = aggregator;
     this.attributesProcessor = attributesProcessor;
     this.maxCardinality = maxCardinality - 1;
     this.reusablePointsPool = new ObjectPool<>(aggregator::createReusablePoint);
+    this.reusableHandlesPool = new ObjectPool<>(aggregator::createHandle);
+    this.handleBuilder = ignored -> reusableHandlesPool.borrowObject();
+    this.handleReleaser = (ignored, handle) -> reusableHandlesPool.returnObject(handle);
     this.pointReleaser = (ignored, point) -> reusablePointsPool.returnObject(point);
 
-    this.aggregatorHolder = new AggregatorHolder<>(aggregator);
-    this.originalAggregator = aggregator;
-    this.handleBuilder = ignored -> aggregatorHolder.reusableHandlesPool.borrowObject();
-
     if (memoryMode == REUSABLE_DATA) {
       this.lastPoints = new PooledHashMap<>();
       this.aggregatorHandles = new PooledHashMap<>();
@@ -200,24 +201,20 @@ public MetricData collect(
       InstrumentationScopeInfo instrumentationScopeInfo,
       long startEpochNanos,
       long epochNanos) {
-    AggregatorHolder<T, U> localAggregatorHolder = aggregatorHolder;
-
     Collection<T> result =
         aggregationTemporality == AggregationTemporality.DELTA
             ? collectWithDeltaAggregationTemporality()
             : collectWithCumulativeAggregationTemporality();
 
     // collectWith*AggregationTemporality() methods are responsible for resetting the handle
-    aggregatorHandles.forEach(localAggregatorHolder.handleReleaser);
+    aggregatorHandles.forEach(handleReleaser);
     aggregatorHandles.clear();
 
-    return localAggregatorHolder.aggregator.toMetricData(
+    return aggregator.toMetricData(
         resource, instrumentationScopeInfo, metricDescriptor, result, aggregationTemporality);
   }
 
   private Collection<T> collectWithDeltaAggregationTemporality() {
-    AggregatorHolder<T, U> localAggregatorHolder = aggregatorHolder;
-
     Map<Attributes, T> currentPoints;
     if (memoryMode == REUSABLE_DATA) {
       // deltaPoints computed in the previous collection can be released
@@ -240,7 +237,7 @@ private Collection<T> collectWithDeltaAggregationTemporality() {
             // AggregatorHandle is going to modify the point eventually, but we must persist its
             // value to used it at the next collection (within lastPoints). Thus, we make a copy.
             pointForCurrentPoints = reusablePointsPool.borrowObject();
-            localAggregatorHolder.aggregator.copyPoint(point, pointForCurrentPoints);
+            aggregator.copyPoint(point, pointForCurrentPoints);
           } else {
             pointForCurrentPoints = point;
           }
@@ -259,16 +256,16 @@ private Collection<T> collectWithDeltaAggregationTemporality() {
               // to make sure currentPoint can still be used within lastPoints during the next
               // collection.
               deltaPoint = reusablePointsPool.borrowObject();
-              localAggregatorHolder.aggregator.copyPoint(currentPoint, deltaPoint);
+              aggregator.copyPoint(currentPoint, deltaPoint);
             } else {
               deltaPoint = currentPoint;
             }
           } else {
             if (memoryMode == REUSABLE_DATA) {
-              localAggregatorHolder.aggregator.diffInPlace(lastPoint, currentPoint);
+              aggregator.diffInPlace(lastPoint, currentPoint);
               deltaPoint = lastPoint;
             } else {
-              deltaPoint = localAggregatorHolder.aggregator.diff(lastPoint, currentPoint);
+              deltaPoint = aggregator.diff(lastPoint, currentPoint);
             }
           }
           deltaPoints.add(deltaPoint);
@@ -318,29 +315,27 @@ private Collection<T> collectWithCumulativeAggregationTemporality() {
   @Override
   public void setEnabled(boolean enabled) {
     if (enabled) {
-      if (aggregatorHolder.aggregator == Aggregator.drop()) {
-        aggregatorHolder = new AggregatorHolder<>(originalAggregator);
+      if (aggregator == Aggregator.drop()) {
+        aggregator = originalAggregator;
       }
     } else {
-      aggregatorHolder = new AggregatorHolder<>(Aggregator.drop());
+      aggregator = Aggregator.drop();
+
+      if (memoryMode == REUSABLE_DATA) {
+        aggregatorHandles.forEach(
+            (attributes, handle) -> {
+              handle.aggregateThenMaybeReset(0, 0, Attributes.empty(), /* reset= */ true);
+              reusableHandlesPool.returnObject(handle);
+            });
+        lastPoints.forEach(pointReleaser);
+      }
+      aggregatorHandles.clear();
+      lastPoints.clear();
     }
   }
 
   @Override
   public boolean isEmpty() {
-    return aggregatorHolder.aggregator == Aggregator.drop();
-  }
-
-  private static final class AggregatorHolder<T extends PointData, U extends ExemplarData> {
-
-    private final Aggregator<T, U> aggregator;
-    private final ObjectPool<AggregatorHandle<T, U>> reusableHandlesPool;
-    private final BiConsumer<Attributes, AggregatorHandle<T, U>> handleReleaser;
-
-    private AggregatorHolder(Aggregator<T, U> aggregator) {
-      this.aggregator = aggregator;
-      this.reusableHandlesPool = new ObjectPool<>(aggregator::createHandle);
-      this.handleReleaser = (ignored, handle) -> reusableHandlesPool.returnObject(handle);
-    }
+    return aggregator == Aggregator.drop();
   }
 }