Merge pull request #7865 from vespa-engine/bjorncs/make-thread-safer

Bjorncs/make thread safer

1 files changed, 31 insertions, 13 deletions

diff --git a/config/src/main/java/com/yahoo/config/subscription/ConfigSubscriber.java b/config/src/main/java/com/yahoo/config/subscription/ConfigSubscriber.java
index 6a7fba90998..d10e7299eeb 100644
--- a/config/src/main/java/com/yahoo/config/subscription/ConfigSubscriber.java
+++ b/config/src/main/java/com/yahoo/config/subscription/ConfigSubscriber.java
@@ -29,10 +29,11 @@ import static java.util.stream.Collectors.toList;
  */
 public class ConfigSubscriber {
 
-    private Logger log = Logger.getLogger(getClass().getName());
+    private final Logger log = Logger.getLogger(getClass().getName());
     private State state = State.OPEN;
     protected List<ConfigHandle<? extends ConfigInstance>> subscriptionHandles = new ArrayList<>();
     private final ConfigSource source;
+    private final Object monitor = new Object();
 
     /** The last complete config generation received by this */
     private long generation = -1;
@@ -110,8 +111,10 @@ public class ConfigSubscriber {
     }
 
     protected void checkStateBeforeSubscribe() {
-        if (state != State.OPEN)
-            throw new IllegalStateException("Adding subscription after calling nextConfig() is not allowed");
+        synchronized (monitor) {
+            if (state != State.OPEN)
+                throw new IllegalStateException("Adding subscription after calling nextConfig() is not allowed");
+        }
     }
 
     protected void subscribeAndHandleErrors(ConfigSubscription<?> sub, ConfigKey<?> configKey, ConfigHandle<?> handle, TimingValues timingValues) {
@@ -230,10 +233,12 @@ public class ConfigSubscriber {
      *         that at lest one of them has changed if <code>requireChange</code> is true), false otherwise
      */
     private boolean acquireSnapshot(long timeoutInMillis, boolean requireChange) {
-        if (state == State.CLOSED) return false;
+        synchronized (monitor) {
+            if (state == State.CLOSED) return false;
+            state = State.FROZEN;
+        }
         long started = System.currentTimeMillis();
         long timeLeftMillis = timeoutInMillis;
-        state = State.FROZEN;
         boolean anyConfigChanged = false;
         boolean allGenerationsChanged = true;
         boolean allGenerationsTheSame = true;
@@ -269,8 +274,10 @@ public class ConfigSubscriber {
             // This indicates the clients will possibly reconfigure their services, so "reset" changed-logic in subscriptions.
             // Also if appropriate update the changed flag on the handler, which clients use.
             markSubsChangedSeen(currentGen);
-            internalRedeploy = internalRedeployOnly;
-            generation = currentGen;
+            synchronized (monitor) {
+                internalRedeploy = internalRedeployOnly;
+                generation = currentGen;
+            }
         }
         return reconfigDue;
     }
@@ -307,7 +314,9 @@ public class ConfigSubscriber {
      * Closes all open {@link ConfigSubscription}s
      */
     public void close() {
-        state = State.CLOSED;
+        synchronized (monitor) {
+            state = State.CLOSED;
+        }
         for (ConfigHandle<? extends ConfigInstance> h : subscriptionHandles) {
             h.subscription().close();
         }
@@ -326,7 +335,10 @@ public class ConfigSubscriber {
 
     @Override
     public String toString() {
-        String ret = "Subscriber state:" + state;
+        String ret;
+        synchronized (monitor) {
+            ret = "Subscriber state:" + state;
+        }
         for (ConfigHandle<?> h : subscriptionHandles) {
             ret = ret + "\n" + h.toString();
         }
@@ -349,7 +361,9 @@ public class ConfigSubscriber {
     }
 
     protected State state() {
-        return state;
+        synchronized (monitor) {
+            return state;
+        }
     }
 
     /**
@@ -382,7 +396,9 @@ public class ConfigSubscriber {
     }
 
     public boolean isClosed() {
-        return state == State.CLOSED;
+        synchronized (monitor) {
+            return state == State.CLOSED;
+        }
     }
 
     /**
@@ -429,14 +445,16 @@ public class ConfigSubscriber {
      * @return the current generation of configs known by this subscriber
      */
     public long getGeneration() {
-        return generation;
+        synchronized (monitor) {
+            return generation;
+        }
     }
 
     /**
      * Whether the current config generation received by this was due to a system-internal redeploy,
      * not an application package change
      */
-    public boolean isInternalRedeploy() { return internalRedeploy; }
+    public boolean isInternalRedeploy() { synchronized (monitor) { return internalRedeploy; } }
 
     /**
      * Convenience interface for clients who only subscribe to one config. Implement this, and pass it to {@link ConfigSubscriber#subscribe(SingleSubscriber, Class, String)}.