rearchitect so that channelpoolhealthchecker doesn't keep a separate datastructure of entries

Change-Id: I0da4b9bad6470c0cc6b5ae168e318a1ef0620ab4

Author: nicholsl

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPool.java

@@ -42,6 +42,7 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
+import org.checkerframework.checker.units.qual.A;
 
 /**
  * A {@link ManagedChannel} that will send requests round-robin via a set of channels.
@@ -61,6 +62,7 @@ public class BigtableChannelPool extends ManagedChannel {
   private final BigtableChannelPoolSettings settings;
   private final ChannelFactory channelFactory;
   private final ScheduledExecutorService executor;
+  private final ScheduledExecutorService channelHealthProbingExecutor = Executors.newSingleThreadScheduledExecutor();
 
   private final Object entryWriteLock = new Object();
   @VisibleForTesting final AtomicReference<ImmutableList<Entry>> entries = new AtomicReference<>();
@@ -94,7 +96,7 @@ public static BigtableChannelPool create(
     ImmutableList.Builder<Entry> initialListBuilder = ImmutableList.builder();
 
     for (int i = 0; i < settings.getInitialChannelCount(); i++) {
-      initialListBuilder.add(new Entry(channelFactory.createSingleChannel()));
+      initialListBuilder.add(new Entry(channelFactory.createSingleChannel(), channelHealthProbingExecutor));
     }
 
     entries.set(initialListBuilder.build());
@@ -152,6 +154,9 @@ public ManagedChannel shutdown() {
       // shutdownNow will cancel scheduled tasks
       executor.shutdownNow();
     }
+    if (channelHealthProbingExecutor != null){
+      channelHealthProbingExecutor.shutdownNow();
+    }
     return this;
   }
 
@@ -192,6 +197,9 @@ public ManagedChannel shutdownNow() {
     if (executor != null) {
       executor.shutdownNow();
     }
+    if (channelHealthProbingExecutor != null) {
+      channelHealthProbingExecutor.shutdownNow();
+    }
     return this;
   }
 
@@ -318,7 +326,7 @@ private void expand(int desiredSize) {
 
     for (int i = 0; i < desiredSize - localEntries.size(); i++) {
       try {
-        newEntries.add(new Entry(channelFactory.createSingleChannel()));
+        newEntries.add(new Entry(channelFactory.createSingleChannel(), channelHealthProbingExecutor));
       } catch (IOException e) {
         LOG.log(Level.WARNING, "Failed to add channel", e);
       }
@@ -356,7 +364,7 @@ void refresh() {
 
       for (int i = 0; i < newEntries.size(); i++) {
         try {
-          newEntries.set(i, new Entry(channelFactory.createSingleChannel()));
+          newEntries.set(i, new Entry(channelFactory.createSingleChannel(), channelHealthProbingExecutor));
         } catch (IOException e) {
           LOG.log(Level.WARNING, "Failed to refresh channel, leaving old channel", e);
         }
@@ -433,19 +441,40 @@ static class Entry {
 
     private final AtomicInteger maxOutstanding = new AtomicInteger();
 
+    /** Number of probes in flight plus number of probe results. (No-op stub) */
+    AtomicInteger recentProbesSent() {
+      return new AtomicInteger(0);
+    }
+
+    /** Number of recently failed probes. (No-op stub) */
+    AtomicInteger recentlyFailedProbes() {
+      return new AtomicInteger(0);
+    }
+
+    private final AtomicInteger probesInFlight = new AtomicInteger(0);
+
+
     // Flag that the channel should be closed once all of the outstanding RPC complete.
     private final AtomicBoolean shutdownRequested = new AtomicBoolean();
     // Flag that the channel has been closed.
     private final AtomicBoolean shutdownInitiated = new AtomicBoolean();
+    private final ChannelHealthChecker healthChecker;
 
-    private Entry(ManagedChannel channel) {
+    private Entry(ManagedChannel channel, ScheduledExecutorService executor) {
       this.channel = channel;
+      this.healthChecker = new ChannelHealthChecker(this, executor);
     }
 
     ManagedChannel getManagedChannel() {
       return this.channel;
     }
 
+    // Add a getter for the healthChecker
+    public ChannelHealthChecker getHealthChecker() {
+      return healthChecker;
+    }
+
+
     int getAndResetMaxOutstanding() {
       return maxOutstanding.getAndSet(outstandingRpcs.get());
     }
@@ -460,7 +489,7 @@ private boolean retain() {
       // register desire to start RPC
       int currentOutstanding = outstandingRpcs.incrementAndGet();
 
-      // Rough book keeping
+      // Rough bookkeeping
       int prevMax = maxOutstanding.get();
       if (currentOutstanding > prevMax) {
         maxOutstanding.incrementAndGet();
@@ -497,6 +526,9 @@ private void release() {
      */
     private void requestShutdown() {
       shutdownRequested.set(true);
+      if (healthChecker != null) {
+        healthChecker.stop();
+      }
       if (outstandingRpcs.get() == 0) {
         shutdown();
       }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/ChannelHealthChecker.java

@@ -43,7 +43,7 @@ public class ChannelHealthChecker {
   private volatile ScheduledFuture<?> scheduledProbeFuture;
   private final ReadWriteLock probeResultsLock = new ReentrantReadWriteLock();
   private final EvictingQueue<ProbeResult> probeResults;
-  private final AtomicInteger probesInFlight = new AtomicInteger(0);
+  // private final AtomicInteger probesInFlight = new AtomicInteger(0);
 
   /** Inner class to represent the result of a single probe. */
   class ProbeResult {
@@ -84,16 +84,6 @@ void probeFinished(Instant startTime, boolean success) {
     // Method stub, no operation.
   }
 
-  /** Number of probes in flight plus number of probe results. (No-op stub) */
-  private int recentProbesSent() {
-    return 0;
-  }
-
-  /** Number of recently failed probes. (No-op stub) */
-  public int recentlyFailedProbes() {
-    return 0;
-  }
-
   /**
    * Determines if the channel is healthy. (No-op stub)
    *

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/ChannelPoolHealthChecker.java

@@ -31,13 +31,11 @@ public class ChannelPoolHealthChecker {
 
   // Class fields
   private final Supplier<ImmutableList<Entry>> entrySupplier;
-  private Map<Entry, ChannelHealthChecker> healthCheckers;
   private Instant lastEviction;
   private ScheduledExecutorService executor;
 
   /** Constructor for the pool health checker. */
   public ChannelPoolHealthChecker(Supplier<ImmutableList<Entry>> entrySupplier) {
-    this.healthCheckers = Collections.synchronizedMap(new WeakHashMap<>());
     this.entrySupplier = entrySupplier;
     this.lastEviction = Instant.MIN;
     this.executor = Executors.newSingleThreadScheduledExecutor();