Remove field volatile and add some docs as to why

Author: andreidan

server/src/main/java/org/elasticsearch/common/MemoryAccountingBytesRefCounted.java

@@ -13,10 +13,10 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.search.TotalHits;
-import org.elasticsearch.common.MemoryAccountingBytesRefCounted;
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.breaker.CircuitBreakingException;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.core.AbstractRefCounted;
 import org.elasticsearch.core.RefCounted;
 import org.elasticsearch.index.fieldvisitor.LeafStoredFieldLoader;
 import org.elasticsearch.index.fieldvisitor.StoredFieldLoader;
@@ -291,7 +291,7 @@ private HitContext prepareHitContext(
         RankDoc rankDoc,
         CircuitBreaker circuitBreaker,
         boolean submitToCb,
-        IntBooleanFunction memoryUsageAccumulator
+        MemoryUsageAccumulator memoryUsageAccumulator
     ) throws IOException {
         if (nestedDocuments.advance(docId - subReaderContext.docBase) == null) {
             return prepareNonNestedHitContext(
@@ -339,7 +339,7 @@ private static HitContext prepareNonNestedHitContext(
         RankDoc rankDoc,
         CircuitBreaker circuitBreaker,
         boolean submitToCB,
-        IntBooleanFunction memoryUsageAccumulator
+        MemoryUsageAccumulator memoryUsageAccumulator
     ) throws IOException {
         int subDocId = docId - subReaderContext.docBase;
 
@@ -368,7 +368,7 @@ private static HitContext prepareNonNestedHitContext(
                     source = sourceLoader.source(leafStoredFieldLoader, subDocId);
                     int accumulatedInLeaf = memoryUsageAccumulator.apply(source.internalSourceRef().length(), submitToCB);
                     if (submitToCB) {
-                        memAccountingRefCounted.setBytesAndAccount(accumulatedInLeaf, "fetch phase source loader");
+                        memAccountingRefCounted.account(accumulatedInLeaf, "fetch phase source loader");
                     }
                 } catch (CircuitBreakingException e) {
                     hit.decRef();
@@ -400,7 +400,7 @@ private static Supplier<Source> lazyStoredSourceLoader(
         int doc,
         MemoryAccountingBytesRefCounted memAccountingRefCounted,
         boolean submitToCB,
-        IntBooleanFunction memoryUsageAccumulator
+        MemoryUsageAccumulator memoryUsageAccumulator
     ) {
         return () -> {
             StoredFieldLoader rootLoader = profiler.storedFields(StoredFieldLoader.create(true, Collections.emptySet()));
@@ -410,7 +410,7 @@ private static Supplier<Source> lazyStoredSourceLoader(
                 BytesReference source = leafRootLoader.source();
                 int accumulatedInLeaf = memoryUsageAccumulator.apply(source.length(), submitToCB);
                 if (submitToCB) {
-                    memAccountingRefCounted.setBytesAndAccount(accumulatedInLeaf, "lazy fetch phase source loader");
+                    memAccountingRefCounted.account(accumulatedInLeaf, "lazy fetch phase source loader");
                 }
                 return Source.fromBytes(source);
             } catch (IOException e) {
@@ -515,7 +515,45 @@ public String toString() {
     }
 
     @FunctionalInterface
-    private interface IntBooleanFunction {
+    private interface MemoryUsageAccumulator {
         int apply(int i, boolean b);
     }
+
+    /**
+     * A ref counted object that accounts for memory usage in bytes and releases the
+     * accounted memory from the circuit breaker when the reference count reaches zero.
+     */
+    static final class MemoryAccountingBytesRefCounted extends AbstractRefCounted {
+
+        // the bytes that we account for are not volatile because we only accumulate
+        // in the single threaded fetch phase and we release the reference after
+        // we write the response to the network (OutboundHandler). As with all other
+        // SearchHit fields this will be visible to the network thread that'll call #decRef.
+        private int bytes;
+        private final CircuitBreaker breaker;
+
+        private MemoryAccountingBytesRefCounted(CircuitBreaker breaker) {
+            this.breaker = breaker;
+        }
+
+        public static MemoryAccountingBytesRefCounted create(CircuitBreaker breaker) {
+            return new MemoryAccountingBytesRefCounted(breaker);
+        }
+
+        /**
+         * This method increments the local counter for the accounted bytes and submits
+         * the accumulated bytes to the circuit breaker.
+         * This method is not thread-safe and should only be called from the single-threaded
+         * fetch phase.
+         */
+        public void account(int bytes, String label) {
+            this.bytes += bytes;
+            breaker.addEstimateBytesAndMaybeBreak(bytes, label);
+        }
+
+        @Override
+        protected void closeInternal() {
+            breaker.addWithoutBreaking(-bytes);
+        }
+    }
 }

server/src/main/java/org/elasticsearch/search/fetch/FetchPhase.java

@@ -7,11 +7,12 @@
  * License v3.0 only", or the "Server Side Public License, v 1".
  */
 
-package org.elasticsearch.common;
+package org.elasticsearch.search.fetch;
 
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.MockBigArrays;
+import org.elasticsearch.search.fetch.FetchPhase.MemoryAccountingBytesRefCounted;
 import org.elasticsearch.test.ESTestCase;
 
 public class MemoryAccountingBytesRefCountedTests extends ESTestCase {
@@ -26,22 +27,22 @@ public void testNoMemoryAccounted() {
     public void testMemoryAccounted() {
         CircuitBreaker breaker = new MockBigArrays.LimitedBreaker("test", ByteSizeValue.ofGb(1));
         MemoryAccountingBytesRefCounted refCounted = MemoryAccountingBytesRefCounted.create(breaker);
-        refCounted.setBytesAndAccount(10, "test");
+        refCounted.account(10, "test");
         assertEquals(10, breaker.getUsed());
     }
 
     public void testCloseInternalDecrementsBreaker() {
         CircuitBreaker breaker = new MockBigArrays.LimitedBreaker("test", ByteSizeValue.ofGb(1));
         MemoryAccountingBytesRefCounted refCounted = MemoryAccountingBytesRefCounted.create(breaker);
-        refCounted.setBytesAndAccount(10, "test");
+        refCounted.account(10, "test");
         refCounted.decRef();
         assertEquals(0, breaker.getUsed());
     }
 
     public void testBreakerNotDecrementedIfRefsRemaining() {
         CircuitBreaker breaker = new MockBigArrays.LimitedBreaker("test", ByteSizeValue.ofGb(1));
         MemoryAccountingBytesRefCounted refCounted = MemoryAccountingBytesRefCounted.create(breaker);
-        refCounted.setBytesAndAccount(10, "test");
+        refCounted.account(10, "test");
         refCounted.incRef(); // 2 refs
         assertEquals(10, breaker.getUsed());
         refCounted.decRef(); // 1 ref remaining so no decrementing is executed