add safety and boundness on thread

Author: rahul2393

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

@@ -18,6 +18,7 @@
 
 import com.google.api.core.InternalApi;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.spanner.v1.BeginTransactionRequest;
 import com.google.spanner.v1.CacheUpdate;
 import com.google.spanner.v1.CommitRequest;
@@ -36,8 +37,10 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
@@ -51,13 +54,9 @@
 @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 static final int MAX_CACHE_UPDATE_THREADS =
+      Math.max(2, Runtime.getRuntime().availableProcessors());
+  private static final ExecutorService CACHE_UPDATE_POOL = createCacheUpdatePool();
 
   private final Object updateLock = new Object();
   private final AtomicLong databaseId = new AtomicLong();
@@ -90,6 +89,22 @@ void useDeterministicRandom() {
     rangeCache.useDeterministicRandom();
   }
 
+  private static ExecutorService createCacheUpdatePool() {
+    ThreadPoolExecutor executor =
+        new ThreadPoolExecutor(
+            MAX_CACHE_UPDATE_THREADS,
+            MAX_CACHE_UPDATE_THREADS,
+            30L,
+            TimeUnit.SECONDS,
+            new LinkedBlockingQueue<>(),
+            new ThreadFactoryBuilder()
+                .setDaemon(true)
+                .setNameFormat("spanner-cache-update-%d")
+                .build());
+    executor.allowCoreThreadTimeOut(true);
+    return executor;
+  }
+
   public void update(CacheUpdate update) {
     synchronized (updateLock) {
       long currentId = databaseId.get();

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

@@ -114,16 +114,24 @@ public void addRanges(CacheUpdate cacheUpdate) {
       touchedGroups.get(i).update(cacheUpdate.getGroup(i));
     }
 
-    writeLock.lock();
     try {
-      for (Range rangeIn : cacheUpdate.getRangeList()) {
-        replaceRangeIfNewer(rangeIn);
-      }
-      for (CachedGroup group : touchedGroups) {
-        unref(group);
+      writeLock.lock();
+      try {
+        for (Range rangeIn : cacheUpdate.getRangeList()) {
+          replaceRangeIfNewer(rangeIn);
+        }
+      } finally {
+        writeLock.unlock();
       }
     } finally {
-      writeLock.unlock();
+      writeLock.lock();
+      try {
+        for (int i = touchedGroups.size() - 1; i >= 0; i--) {
+          unref(touchedGroups.get(i));
+        }
+      } finally {
+        writeLock.unlock();
+      }
     }
   }