Acquire lock before starting schedulers and threads
package com.yahoo.vespa.hosted.node.admin.nodeadmin;
import com.yahoo.concurrent.ThreadFactoryFactory;
+import com.yahoo.concurrent.classlock.ClassLock;
+import com.yahoo.concurrent.classlock.ClassLocking;
+import com.yahoo.concurrent.classlock.LockInterruptException;
import com.yahoo.log.LogLevel;
import com.yahoo.vespa.hosted.node.admin.ContainerNodeSpec;
import com.yahoo.vespa.hosted.node.admin.maintenance.StorageMaintainer;
@@ -48,16 +51,17 @@ public class NodeAdminStateUpdater {
private final Logger log = Logger.getLogger(NodeAdminStateUpdater.class.getName());
private final ScheduledExecutorService specVerifierScheduler =
Executors.newScheduledThreadPool(1, ThreadFactoryFactory.getDaemonThreadFactory("specverifier"));
- private Thread loopThread;
+ private final Thread loopThread;
private final NodeRepository nodeRepository;
private final Orchestrator orchestrator;
- private final StorageMaintainer storageMaintainer;
private final NodeAdmin nodeAdmin;
private final Clock clock;
private final String dockerHostHostName;
private final Duration nodeAdminConvergeStateInterval;
+ private final ClassLocking classLocking;
+ private ClassLock classLock;
private Instant lastTick;
public NodeAdminStateUpdater(
@@ -67,15 +71,33 @@ public class NodeAdminStateUpdater {
NodeAdmin nodeAdmin,
String dockerHostHostName,
Clock clock,
- Duration nodeAdminConvergeStateInterval) {
+ Duration nodeAdminConvergeStateInterval,
+ ClassLocking classLocking) {
this.nodeRepository = nodeRepository;
this.orchestrator = orchestrator;
- this.storageMaintainer = storageMaintainer;
this.nodeAdmin = nodeAdmin;
this.dockerHostHostName = dockerHostHostName;
this.clock = clock;
this.nodeAdminConvergeStateInterval = nodeAdminConvergeStateInterval;
+ this.classLocking = classLocking;
this.lastTick = clock.instant();
+
+ this.loopThread = new Thread(() -> {
+ try {
+ classLock = classLocking.tryLock(NodeAdminStateUpdater.class, () -> !terminated.get());
+ } catch (LockInterruptException e) {
+ return;
+ }
+
+ nodeAdmin.start();
+ specVerifierScheduler.scheduleWithFixedDelay(() ->
+ updateHardwareDivergence(storageMaintainer), 5, 60, TimeUnit.MINUTES);
+
+ while (! terminated.get()) {
+ tick();
+ }
+ });
+ this.loopThread.setName("tick-NodeAdminStateUpdater");
}
public enum State { RESUMED, SUSPENDED_NODE_ADMIN, SUSPENDED}
@@ -257,29 +279,21 @@ public class NodeAdminStateUpdater {
}
public void start() {
- if (loopThread != null) {
- throw new RuntimeException("Can not restart NodeAdminStateUpdater");
- }
-
- loopThread = new Thread(() -> {
- while (! terminated.get()) tick();
- });
- loopThread.setName("tick-NodeAdminStateUpdater");
loopThread.start();
-
- nodeAdmin.start();
- specVerifierScheduler.scheduleWithFixedDelay(() ->
- updateHardwareDivergence(storageMaintainer), 5, 60, TimeUnit.MINUTES);
}
public void stop() {
- specVerifierScheduler.shutdown();
if (!terminated.compareAndSet(false, true)) {
throw new RuntimeException("Can not re-stop a node agent.");
}
+ synchronized (classLocking) {
+ classLocking.notifyAll();
+ }
+
// First we need to stop NodeAdminStateUpdater thread to make sure no new NodeAgents are spawned
signalWorkToBeDone();
+ specVerifierScheduler.shutdown();
do {
try {
@@ -292,5 +306,9 @@ public class NodeAdminStateUpdater {
// Finally, stop NodeAdmin and all the NodeAgents
nodeAdmin.stop();
+
+ if (classLock != null) {
+ classLock.close();
+ }
}
}
