perf: Memory optimization for RangeCache (#65)

Motivation

Currently, RangeCache will implicitly retain a reference to the RangeResponse KeyValue arrays used to initially populate/refresh the cache via the promise held in its startFuture field.

This ref is there only to propagate cancellation to the etcd range request but can take up significant memory once completed if the cache is large.

Note this is mostly the size of the object array itself since the contained objects will usually be referenced from the cache's map. However over time unnecessary references could remain if/when the corresponding KeyValues are deleted. The array's "shallow" footprint might also be non-negligible for very large caches.

Modifications

- Change the promise used for the start future to hold an explicit reference to the range request future, so that it can be nulled as soon as that other future completes.
- Remove an incorrect assertion
- Clean up a weird @nullable import

Author: njhill

src/main/java/com/ibm/etcd/client/utils/RangeCache.java

@@ -44,15 +44,13 @@
 
 import javax.annotation.concurrent.GuardedBy;
 
-import org.checkerframework.checker.nullness.qual.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.google.common.collect.Iterators;
 import com.google.common.util.concurrent.AbstractFuture;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
 import com.google.protobuf.ByteString;
 import com.ibm.etcd.client.EtcdClient;
 import com.ibm.etcd.client.FutureListener;
@@ -210,22 +208,46 @@ protected ListenableFuture<Boolean> fullRefreshCache() {
                         .map(ResponseOp::getResponseRange)
                         .collect(Collectors.toList()), directExecutor());
         }
-        
-        final SettableFuture<Boolean> promise = new SettableFuture<Boolean>() {
+
+        final class StartPromise extends SettableFuture<Boolean> implements Runnable {
+            // Reference to parent future (etcd range request) held here just
+            // to be able to propagate cancellation if that happens before it completes.
+            volatile ListenableFuture<?> parent;
+            StartPromise(ListenableFuture<?> parent) {
+                this.parent = parent;
+                parent.addListener(this, directExecutor());
+            }
             @Override
-            protected void interruptTask() {
-                if (rrfut.cancel(true)) {
+            public void run() {
+                // Ensure that we remove our reference to the "parent" future as soon
+                // as it completes, since RangeCache stores this promise in the
+                // startFuture field, and we want to allow the original future and its
+                // possibly-large value to be garbage collected.
+                parent = null;
+            }
+            @Override
+            protected void afterDone() {
+                if (!wasInterrupted()) {
+                    return;
+                }
+                // propagate cancellation
+                final ListenableFuture<?> p = parent;
+                if (p != null && p.cancel(true)) {
                     return;
                 }
                 Watch theWatch;
                 synchronized(RangeCache.this) {
                     theWatch = watch;
+                    closed = true;
                 }
                 if (theWatch != null) {
                     theWatch.close();
                 }
             }
-        };
+        }
+
+        final SettableFuture<Boolean> promise = new StartPromise(rrfut);
+
         Futures.addCallback(rrfut, (FutureListener<List<RangeResponse>>) (rrs, err) -> {
             if (rrs != null) try {
                 setupWatch(rrs, firstTime, promise);
@@ -323,7 +345,6 @@ public void onError(Throwable t) {
                     promise.setException(new CancellationException());
                     return;
                 }
-                assert startFuture == promise;
                 boolean isDone = promise.isDone();
                 if (isDone && promise.isCancelled()) {
                     return;
@@ -651,7 +672,7 @@ public boolean keyExistsRemote(ByteString key) {
     /**
      * Stores result of put operations
      */
-    public static class PutResult {
+    public static final class PutResult {
         private final boolean succ;
         private final KeyValue kv;
         public PutResult(boolean success, KeyValue kv) {
@@ -1020,7 +1041,7 @@ public boolean isClosed() {
 
     static class SettableFuture<V> extends AbstractFuture<V> {
         @Override
-        public boolean set(@Nullable V value) {
+        public boolean set(V value) {
             return super.set(value);
         }
         @Override