Merge pull request #11597 from vespa-engine/balder/correct-dispatcher-shutdown

This is the correct way of shutting down the the Dispatcher.

9 files changed, 70 insertions, 18 deletions

diff --git a/container-search/abi-spec.json b/container-search/abi-spec.json
index 0f55f777250..b5fbe235c43 100644
--- a/container-search/abi-spec.json
+++ b/container-search/abi-spec.json
@@ -1953,6 +1953,7 @@
     ],
     "methods": [
       "public void <init>(com.yahoo.search.cluster.NodeManager)",
+      "public void <init>(com.yahoo.search.cluster.NodeManager, boolean)",
       "public com.yahoo.search.cluster.MonitorConfiguration getConfiguration()",
       "public void add(java.lang.Object, boolean)",
       "public com.yahoo.search.cluster.BaseNodeMonitor getNodeMonitor(java.lang.Object)",
@@ -1977,6 +1978,7 @@
     "methods": [
       "public void <init>(com.yahoo.component.ComponentId, java.util.List, boolean)",
       "public void <init>(com.yahoo.component.ComponentId, java.util.List, com.yahoo.search.cluster.Hasher, boolean)",
+      "public void <init>(com.yahoo.component.ComponentId, java.util.List, com.yahoo.search.cluster.Hasher, boolean, boolean)",
       "public final void ping(java.lang.Object, java.util.concurrent.Executor)",
       "protected abstract com.yahoo.prelude.Pong ping(com.yahoo.prelude.Ping, java.lang.Object)",
       "protected java.lang.Object getFirstConnection(com.yahoo.search.cluster.Hasher$NodeList, int, int, com.yahoo.search.Query)",
diff --git a/container-search/src/main/java/com/yahoo/search/cluster/ClusterMonitor.java b/container-search/src/main/java/com/yahoo/search/cluster/ClusterMonitor.java
index a016f7d695c..d4b6279be89 100644
--- a/container-search/src/main/java/com/yahoo/search/cluster/ClusterMonitor.java
+++ b/container-search/src/main/java/com/yahoo/search/cluster/ClusterMonitor.java
@@ -12,6 +12,7 @@ import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -32,15 +33,21 @@ public class ClusterMonitor<T> {
 
     private final MonitorThread monitorThread;
 
-    private volatile boolean shutdown = false;
+    private final AtomicBoolean closed = new AtomicBoolean(false);
 
     /** A map from Node to corresponding MonitoredNode */
     private final Map<T, TrafficNodeMonitor<T>> nodeMonitors = Collections.synchronizedMap(new java.util.LinkedHashMap<>());
 
     public ClusterMonitor(NodeManager<T> manager) {
+        this(manager, true);
+    }
+
+    public ClusterMonitor(NodeManager<T> manager, boolean startPingThread) {
         nodeManager = manager;
         monitorThread = new MonitorThread("search.clustermonitor");
-        monitorThread.start();
+        if (startPingThread) {
+            monitorThread.start();
+        }
     }
 
     /** Returns the configuration of this cluster monitor */
@@ -70,6 +77,7 @@ public class ClusterMonitor<T> {
 
     /** Called from ClusterSearcher/NodeManager when a node failed */
     public synchronized void failed(T node, ErrorMessage error) {
+        if (closed.get()) return; // Do not touch state if close has started.
         TrafficNodeMonitor<T> monitor = nodeMonitors.get(node);
         Boolean wasWorking = monitor.isKnownWorking();
         monitor.failed(error);
@@ -79,6 +87,7 @@ public class ClusterMonitor<T> {
 
     /** Called when a node responded */
     public synchronized void responded(T node) {
+        if (closed.get()) return; // Do not touch state if close has started.
         TrafficNodeMonitor<T> monitor = nodeMonitors.get(node);
         Boolean wasWorking = monitor.isKnownWorking();
         monitor.responded();
@@ -90,10 +99,11 @@ public class ClusterMonitor<T> {
      * Ping all nodes which needs pinging to discover state changes
      */
     public void ping(Executor executor) {
-        for (Iterator<BaseNodeMonitor<T>> i = nodeMonitorIterator(); i.hasNext(); ) {
+        for (Iterator<BaseNodeMonitor<T>> i = nodeMonitorIterator(); i.hasNext() && !closed.get(); ) {
             BaseNodeMonitor<T> monitor= i.next();
             nodeManager.ping(monitor.getNode(), executor); // Cause call to failed or responded
         }
+        if (closed.get()) return; // Do nothing to change state if close has started.
         nodeManager.pingIterationCompleted();
     }
 
@@ -104,15 +114,23 @@ public class ClusterMonitor<T> {
 
     /** Returns a thread-safe snapshot of the NodeMonitors of all added nodes */
     public List<BaseNodeMonitor<T>> nodeMonitors() {
-        synchronized (nodeMonitors) {
-            return new java.util.ArrayList<>(nodeMonitors.values());
-        }
+        return new java.util.ArrayList<>(nodeMonitors.values());
     }
 
     /** Must be called when this goes out of use */
     public void shutdown() {
-        shutdown = true;
-        monitorThread.interrupt();
+        closed.set(true);
+        synchronized (this) {
+            nodeMonitors.clear();
+        }
+        synchronized (nodeManager) {
+            nodeManager.notifyAll();
+        }
+        try {
+            if (monitorThread.isAlive()) {
+                monitorThread.join();
+            }
+        } catch (InterruptedException e) {}
     }
 
     private class MonitorThread extends Thread {
@@ -128,14 +146,16 @@ public class ClusterMonitor<T> {
             // any thread local connections are reused) 2) a new thread will be started to execute
             // new pings when a ping is not responding
             ExecutorService pingExecutor=Executors.newCachedThreadPool(ThreadFactoryFactory.getDaemonThreadFactory("search.ping"));
-            while (!isInterrupted()) {
+            while (!closed.get()) {
                 try {
-                    Thread.sleep(configuration.getCheckInterval());
                     log.finest("Activating ping");
                     ping(pingExecutor);
+                    synchronized (nodeManager) {
+                        nodeManager.wait(configuration.getCheckInterval());
+                    }
                 }
                 catch (Throwable e) {
-                    if (shutdown && e instanceof InterruptedException) {
+                    if (closed.get() && e instanceof InterruptedException) {
                         break;
                     } else if ( ! (e instanceof Exception) ) {
                         log.log(Level.WARNING,"Error in monitor thread, will quit", e);
diff --git a/container-search/src/main/java/com/yahoo/search/cluster/ClusterSearcher.java b/container-search/src/main/java/com/yahoo/search/cluster/ClusterSearcher.java
index dd8b953e3d0..06cbf9a9706 100644
--- a/container-search/src/main/java/com/yahoo/search/cluster/ClusterSearcher.java
+++ b/container-search/src/main/java/com/yahoo/search/cluster/ClusterSearcher.java
@@ -41,7 +41,7 @@ import java.util.concurrent.TimeoutException;
 public abstract class ClusterSearcher<T> extends PingableSearcher implements NodeManager<T> {
 
     private final Hasher<T> hasher;
-    private final ClusterMonitor<T> monitor = new ClusterMonitor<>(this);
+    private final ClusterMonitor<T> monitor;
 
     /**
      * Creates a new cluster searcher
@@ -55,8 +55,13 @@ public abstract class ClusterSearcher<T> extends PingableSearcher implements Nod
     }
 
     public ClusterSearcher(ComponentId id, List<T> connections, Hasher<T> hasher, boolean internal) {
+        this(id, connections, hasher, internal, true);
+    }
+
+    public ClusterSearcher(ComponentId id, List<T> connections, Hasher<T> hasher, boolean internal, boolean startPingThread) {
         super(id);
         this.hasher = hasher;
+        this.monitor = new ClusterMonitor<>(this, startPingThread);
         for (T connection : connections) {
             monitor.add(connection, internal);
             hasher.add(connection);
diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java b/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java
index e078ffa685f..224facd0c5b 100644
--- a/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java
+++ b/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java
@@ -131,8 +131,9 @@ public class Dispatcher extends AbstractComponent {
 
     @Override
     public void deconstruct() {
-        invokerFactory.release();
+        /* The seach cluster must be shutdown first as it uses the invokerfactory. */
         searchCluster.shutDown();
+        invokerFactory.release();
     }
 
     public FillInvoker getFillInvoker(Result result, VespaBackEndSearcher searcher) {
diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/rpc/RpcClient.java b/container-search/src/main/java/com/yahoo/search/dispatch/rpc/RpcClient.java
index ea123b255eb..05ce6d50493 100644
--- a/container-search/src/main/java/com/yahoo/search/dispatch/rpc/RpcClient.java
+++ b/container-search/src/main/java/com/yahoo/search/dispatch/rpc/RpcClient.java
@@ -86,7 +86,9 @@ class RpcClient implements Client {
 
         @Override
         public void close() {
-            target.close();
+            if (target != null) {
+                target.close();
+            }
         }
 
         @Override
diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java b/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java
index 3595a24ca92..fb55e330ebe 100644
--- a/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java
+++ b/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java
@@ -408,8 +408,7 @@ public class SearchCluster implements NodeManager<Node> {
             activeDocuments += n.getActiveDocuments();
         }
         long averageDocumentsInOtherGroups = sumOfActiveDocuments / otherGroups;
-        boolean sufficient = isGroupCoverageSufficient(nodes.size(), nodesInGroup, activeDocuments, averageDocumentsInOtherGroups);
-        return sufficient;
+        return isGroupCoverageSufficient(nodes.size(), nodesInGroup, activeDocuments, averageDocumentsInOtherGroups);
     }
 
     private void trackGroupCoverageChanges(int index, Group group, boolean fullCoverage, long averageDocuments) {
diff --git a/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTestCase.java b/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTestCase.java
index 4011611b049..b5e54b46e5a 100644
--- a/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTestCase.java
+++ b/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTestCase.java
@@ -3,6 +3,8 @@ package com.yahoo.prelude.fastsearch.test;
 
 import com.google.common.collect.ImmutableList;
 import com.yahoo.component.chain.Chain;
+import com.yahoo.container.QrSearchersConfig;
+import com.yahoo.container.handler.VipStatus;
 import com.yahoo.container.protect.Error;
 import com.yahoo.language.simple.SimpleLinguistics;
 import com.yahoo.prelude.fastsearch.ClusterParams;
@@ -12,6 +14,8 @@ import com.yahoo.prelude.fastsearch.SummaryParameters;
 import com.yahoo.search.Query;
 import com.yahoo.search.Result;
 import com.yahoo.search.Searcher;
+import com.yahoo.search.dispatch.Dispatcher;
+import com.yahoo.search.dispatch.rpc.RpcResourcePool;
 import com.yahoo.search.dispatch.searchcluster.Node;
 import com.yahoo.search.grouping.GroupingRequest;
 import com.yahoo.search.grouping.request.AllOperation;
@@ -30,6 +34,7 @@ import java.util.logging.Logger;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 
 /**
@@ -136,4 +141,21 @@ public class FastSearcherTestCase {
             assertForceSinglePassIs(expected, child);
     }
 
+    @Test
+    public void testDispatchReconfig() {
+        String clusterName = "a";
+        var b = new QrSearchersConfig.Builder();
+        var searchClusterB = new QrSearchersConfig.Searchcluster.Builder();
+        searchClusterB.name(clusterName);
+        b.searchcluster(searchClusterB);
+        VipStatus vipStatus = new VipStatus(b.build());
+        List<Node> nodes_1 = ImmutableList.of(new Node(0, "host0", 0));
+        RpcResourcePool rpcPool_1 = new RpcResourcePool(MockDispatcher.toDispatchConfig(nodes_1));
+        Dispatcher dispatch_1 = MockDispatcher.create(nodes_1, rpcPool_1, 1, vipStatus);
+        vipStatus.addToRotation(clusterName);
+        assertTrue(vipStatus.isInRotation());
+        dispatch_1.deconstruct();
+        assertTrue(vipStatus.isInRotation()); //Verify that deconstruct does not touch vipstatus
+    }
+
 }
diff --git a/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/MockDispatcher.java b/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/MockDispatcher.java
index d1e04b26c15..440e3b8d78f 100644
--- a/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/MockDispatcher.java
+++ b/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/MockDispatcher.java
@@ -34,7 +34,7 @@ class MockDispatcher extends Dispatcher {
         super(searchCluster, dispatchConfig, invokerFactory, new MockMetric());
     }
 
-    private static DispatchConfig toDispatchConfig(List<Node> nodes) {
+    static DispatchConfig toDispatchConfig(List<Node> nodes) {
         DispatchConfig.Builder dispatchConfigBuilder = new DispatchConfig.Builder();
         int key = 0;
         for (Node node : nodes) {
diff --git a/container-search/src/test/java/com/yahoo/search/cluster/test/ClusteredConnectionTestCase.java b/container-search/src/test/java/com/yahoo/search/cluster/test/ClusteredConnectionTestCase.java
index aa57e7903f9..c90d2774bd1 100644
--- a/container-search/src/test/java/com/yahoo/search/cluster/test/ClusteredConnectionTestCase.java
+++ b/container-search/src/test/java/com/yahoo/search/cluster/test/ClusteredConnectionTestCase.java
@@ -8,6 +8,7 @@ import com.yahoo.prelude.Pong;
 import com.yahoo.search.Query;
 import com.yahoo.search.Result;
 import com.yahoo.search.cluster.ClusterSearcher;
+import com.yahoo.search.cluster.Hasher;
 import com.yahoo.search.result.ErrorMessage;
 import com.yahoo.search.result.Hit;
 import com.yahoo.search.searchchain.Execution;
@@ -164,7 +165,7 @@ public class ClusteredConnectionTestCase {
     private static class MyBackend extends ClusterSearcher<Connection> {
 
         public MyBackend(ComponentId componentId, List<Connection> connections) {
-            super(componentId,connections, false);
+            super(componentId,connections, new Hasher<>(), false, false);
         }
 
         @Override