chore(spanner): optimize lock contention and skipped tablet reporting

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/ChannelFinder.java

@@ -17,6 +17,7 @@
 package com.google.cloud.spanner.spi.v1;
 
 import com.google.api.core.InternalApi;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.spanner.v1.BeginTransactionRequest;
 import com.google.spanner.v1.CacheUpdate;
 import com.google.spanner.v1.CommitRequest;
@@ -34,6 +35,9 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Predicate;
@@ -47,11 +51,20 @@
 @InternalApi
 public final class ChannelFinder {
   private static final Predicate<String> NO_EXCLUDED_ENDPOINTS = address -> false;
+  private static final ExecutorService CACHE_UPDATE_POOL =
+      Executors.newCachedThreadPool(
+          r -> {
+            Thread t = new Thread(r, "spanner-cache-update");
+            t.setDaemon(true);
+            return t;
+          });
 
   private final Object updateLock = new Object();
   private final AtomicLong databaseId = new AtomicLong();
   private final KeyRecipeCache recipeCache = new KeyRecipeCache();
   private final KeyRangeCache rangeCache;
+  private final Executor cacheUpdateExecutor =
+      MoreExecutors.newSequentialExecutor(CACHE_UPDATE_POOL);
   @Nullable private final EndpointLifecycleManager lifecycleManager;
   @Nullable private final String finderKey;
 
@@ -112,6 +125,10 @@ public void update(CacheUpdate update) {
     }
   }
 
+  public void updateAsync(CacheUpdate update) {
+    cacheUpdateExecutor.execute(() -> update(update));
+  }
+
   public ChannelEndpoint findServer(ReadRequest.Builder reqBuilder) {
     return findServer(reqBuilder, preferLeader(reqBuilder.getTransaction()), NO_EXCLUDED_ENDPOINTS);
   }

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManager.java

@@ -73,6 +73,13 @@ class EndpointLifecycleManager {
    */
   private static final int MAX_TRANSIENT_FAILURE_COUNT = 3;
 
+  private enum EvictionReason {
+    TRANSIENT_FAILURE,
+    SHUTDOWN,
+    IDLE,
+    STALE
+  }
+
   /** Per-endpoint lifecycle state. */
   static final class EndpointState {
     final String address;
@@ -95,6 +102,7 @@ static final class EndpointState {
 
   private final ChannelEndpointCache endpointCache;
   private final Map<String, EndpointState> endpoints = new ConcurrentHashMap<>();
+  private final Set<String> transientFailureEvictedAddresses = ConcurrentHashMap.newKeySet();
 
   /**
    * Active addresses reported by each ChannelFinder, keyed by database id.
@@ -235,6 +243,7 @@ void updateActiveAddresses(String finderKey, Set<String> activeAddresses) {
       for (Set<String> addresses : activeAddressesPerFinder.values()) {
         allActive.addAll(addresses);
       }
+      transientFailureEvictedAddresses.retainAll(allActive);
 
       // Evict managed endpoints not referenced by any finder.
       List<String> stale = new ArrayList<>();
@@ -276,6 +285,7 @@ void unregisterFinder(String finderKey) {
       for (Set<String> addresses : activeAddressesPerFinder.values()) {
         allActive.addAll(addresses);
       }
+      transientFailureEvictedAddresses.retainAll(allActive);
 
       List<String> stale = new ArrayList<>();
       for (String address : endpoints.keySet()) {
@@ -412,6 +422,7 @@ private void probe(String address) {
         case READY:
           state.lastReadyAt = clock.instant();
           state.consecutiveTransientFailures = 0;
+          transientFailureEvictedAddresses.remove(address);
           break;
 
         case IDLE:
@@ -439,13 +450,13 @@ private void probe(String address) {
                 Level.FINE,
                 "Evicting endpoint {0}: {1} consecutive TRANSIENT_FAILURE probes",
                 new Object[] {address, state.consecutiveTransientFailures});
-            evictEndpoint(address);
+            evictEndpoint(address, EvictionReason.TRANSIENT_FAILURE);
           }
           break;
 
         case SHUTDOWN:
           logger.log(Level.FINE, "Probe for {0}: channel SHUTDOWN, evicting endpoint", address);
-          evictEndpoint(address);
+          evictEndpoint(address, EvictionReason.SHUTDOWN);
           break;
 
         default:
@@ -482,16 +493,26 @@ void checkIdleEviction() {
     }
 
     for (String address : toEvict) {
-      evictEndpoint(address);
+      evictEndpoint(address, EvictionReason.IDLE);
     }
   }
 
   /** Evicts an endpoint: stops probing, removes from tracking, shuts down the channel. */
   private void evictEndpoint(String address) {
+    evictEndpoint(address, EvictionReason.STALE);
+  }
+
+  /** Evicts an endpoint and records whether it should still be reported as unhealthy. */
+  private void evictEndpoint(String address, EvictionReason reason) {
     logger.log(Level.FINE, "Evicting endpoint {0}", address);
 
     stopProbing(address);
     endpoints.remove(address);
+    if (reason == EvictionReason.TRANSIENT_FAILURE) {
+      transientFailureEvictedAddresses.add(address);
+    } else {
+      transientFailureEvictedAddresses.remove(address);
+    }
     endpointCache.evict(address);
   }
 
@@ -526,6 +547,10 @@ boolean isManaged(String address) {
     return endpoints.containsKey(address);
   }
 
+  boolean wasRecentlyEvictedTransientFailure(String address) {
+    return transientFailureEvictedAddresses.contains(address);
+  }
+
   /** Returns the endpoint state for testing. */
   @VisibleForTesting
   EndpointState getEndpointState(String address) {
@@ -558,6 +583,7 @@ void shutdown() {
       }
     }
     endpoints.clear();
+    transientFailureEvictedAddresses.clear();
 
     scheduler.shutdown();
     try {

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyAwareChannel.java

@@ -798,25 +798,25 @@ public void onMessage(ResponseT message) {
       if (message instanceof PartialResultSet) {
         PartialResultSet response = (PartialResultSet) message;
         if (response.hasCacheUpdate() && call.channelFinder != null) {
-          call.channelFinder.update(response.getCacheUpdate());
+          call.channelFinder.updateAsync(response.getCacheUpdate());
         }
         transactionId = transactionIdFromMetadata(response);
       } else if (message instanceof ResultSet) {
         ResultSet response = (ResultSet) message;
         if (response.hasCacheUpdate() && call.channelFinder != null) {
-          call.channelFinder.update(response.getCacheUpdate());
+          call.channelFinder.updateAsync(response.getCacheUpdate());
         }
         transactionId = transactionIdFromMetadata(response);
       } else if (message instanceof Transaction) {
         Transaction response = (Transaction) message;
         if (response.hasCacheUpdate() && call.channelFinder != null) {
-          call.channelFinder.update(response.getCacheUpdate());
+          call.channelFinder.updateAsync(response.getCacheUpdate());
         }
         transactionId = transactionIdFromTransaction(response);
       } else if (message instanceof CommitResponse) {
         CommitResponse response = (CommitResponse) message;
         if (response.hasCacheUpdate() && call.channelFinder != null) {
-          call.channelFinder.update(response.getCacheUpdate());
+          call.channelFinder.updateAsync(response.getCacheUpdate());
         }
       }
       if (transactionId != null) {