Merge pull request #26915 from vespa-engine/bjorncs/jdisc-timeout

Bjorncs/jdisc timeout

2 files changed, 46 insertions, 47 deletions

diff --git a/jdisc_core/src/main/java/com/yahoo/jdisc/Request.java b/jdisc_core/src/main/java/com/yahoo/jdisc/Request.java
index 7b63eba86b5..da05d468888 100644
--- a/jdisc_core/src/main/java/com/yahoo/jdisc/Request.java
+++ b/jdisc_core/src/main/java/com/yahoo/jdisc/Request.java
@@ -36,6 +36,7 @@ import java.util.concurrent.TimeUnit;
  */
 public class Request extends AbstractResource {
 
+    private final Object monitor = new Object();
     private final Map<String, Object> context = Collections.synchronizedMap(new HashMap<>());
     private final HeaderFields headers = new HeaderFields();
     private final Container container;
@@ -44,8 +45,8 @@ public class Request extends AbstractResource {
     private final long creationTime;
     private final boolean serverRequest;
     private final URI uri;
-    private volatile boolean cancel = false;
-    private BindingMatch<RequestHandler> bindingMatch;
+    private boolean cancel = false;
+    private volatile BindingMatch<RequestHandler> bindingMatch;
     private TimeoutManager timeoutManager;
     private Long timeout;
 
@@ -217,12 +218,14 @@ public class Request extends AbstractResource {
      */
     public void setTimeoutManager(TimeoutManager timeoutManager) {
         Objects.requireNonNull(timeoutManager, "timeoutManager");
-        if (this.timeoutManager != null) {
-            throw new IllegalStateException("Timeout manager already set.");
-        }
-        this.timeoutManager = timeoutManager;
-        if (timeout != null) {
-            timeoutManager.scheduleTimeout(this);
+        synchronized (monitor) {
+            if (this.timeoutManager != null) {
+                throw new IllegalStateException("Timeout manager already set.");
+            }
+            this.timeoutManager = timeoutManager;
+            if (timeout != null) {
+                timeoutManager.scheduleTimeout(this);
+            }
         }
     }
 
@@ -233,7 +236,7 @@ public class Request extends AbstractResource {
      * @see #setTimeoutManager(TimeoutManager)
      */
     public TimeoutManager getTimeoutManager() {
-        return timeoutManager;
+        synchronized (monitor) { return timeoutManager; }
     }
 
     /**
@@ -252,9 +255,11 @@ public class Request extends AbstractResource {
      * @see #timeRemaining(TimeUnit)
      */
     public void setTimeout(long timeout, TimeUnit unit) {
-        this.timeout = unit.toMillis(timeout);
-        if (timeoutManager != null) {
-            timeoutManager.scheduleTimeout(this);
+        synchronized (monitor) {
+            this.timeout = unit.toMillis(timeout);
+            if (timeoutManager != null) {
+                timeoutManager.scheduleTimeout(this);
+            }
         }
     }
 
@@ -267,10 +272,12 @@ public class Request extends AbstractResource {
      * @see #setTimeout(long, TimeUnit)
      */
     public Long getTimeout(TimeUnit unit) {
-        if (timeout == null) {
-            return null;
+        synchronized (monitor) {
+            if (timeout == null) {
+                return null;
+            }
+            return unit.convert(timeout, TimeUnit.MILLISECONDS);
         }
-        return unit.convert(timeout, TimeUnit.MILLISECONDS);
     }
 
     /**
@@ -281,10 +288,12 @@ public class Request extends AbstractResource {
      * @return The number of time units left until this Request times out, or <em>null</em>.
      */
     public Long timeRemaining(TimeUnit unit) {
-        if (timeout == null) {
-            return null;
+        synchronized (monitor) {
+            if (timeout == null) {
+                return null;
+            }
+            return unit.convert(timeout - (container().currentTimeMillis() - creationTime), TimeUnit.MILLISECONDS);
         }
-        return unit.convert(timeout - (container().currentTimeMillis() - creationTime), TimeUnit.MILLISECONDS);
     }
 
     /**
@@ -324,11 +333,13 @@ public class Request extends AbstractResource {
      * @see #setTimeout(long, TimeUnit)
      */
     public boolean isCancelled() {
-        if (cancel) {
-            return true;
-        }
-        if (timeout != null && timeRemaining(TimeUnit.MILLISECONDS) <= 0) {
-            return true;
+        synchronized (monitor) {
+            if (cancel) {
+                return true;
+            }
+            if (timeout != null && timeRemaining(TimeUnit.MILLISECONDS) <= 0) {
+                return true;
+            }
         }
         if (parent != null && parent.isCancelled()) {
             return true;
@@ -343,11 +354,13 @@ public class Request extends AbstractResource {
      * @see #isCancelled()
      */
     public void cancel() {
-        if (cancel) return;
+        synchronized (monitor) {
+            if (cancel) return;
 
-        if (timeoutManager != null && timeout != null)
-            timeoutManager.unscheduleTimeout(this);
-        cancel = true;
+            if (timeoutManager != null && timeout != null)
+                timeoutManager.unscheduleTimeout(this);
+            cancel = true;
+        }
     }
 
     /**
diff --git a/jdisc_core/src/main/java/com/yahoo/jdisc/core/TimeoutManagerImpl.java b/jdisc_core/src/main/java/com/yahoo/jdisc/core/TimeoutManagerImpl.java
index d7b652b80f7..7a3898b2946 100644
--- a/jdisc_core/src/main/java/com/yahoo/jdisc/core/TimeoutManagerImpl.java
+++ b/jdisc_core/src/main/java/com/yahoo/jdisc/core/TimeoutManagerImpl.java
@@ -19,7 +19,6 @@ import java.util.Queue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -30,10 +29,9 @@ public class TimeoutManagerImpl {
 
     private static final ContentChannel IGNORED_CONTENT = new IgnoredContent();
     private static final Logger log = Logger.getLogger(TimeoutManagerImpl.class.getName());
-    private final ScheduledQueue [] schedules = new ScheduledQueue[Runtime.getRuntime().availableProcessors()];
+    private final ScheduledQueue scheduler;
     private final Thread thread;
     private final Timer timer;
-    private final AtomicInteger nextScheduler = new AtomicInteger(0);
     private final AtomicBoolean done = new AtomicBoolean(false);
 
     @Inject
@@ -41,11 +39,7 @@ public class TimeoutManagerImpl {
         this.thread = factory.newThread(new ManagerTask());
         this.thread.setName(getClass().getName());
         this.timer = timer;
-
-        long now = timer.currentTimeMillis();
-        for (int i = 0; i < schedules.length; ++i) {
-            schedules[i] = new ScheduledQueue(now);
-        }
+        this.scheduler = new ScheduledQueue(timer.currentTimeMillis());
     }
 
     public void start() {
@@ -66,13 +60,7 @@ public class TimeoutManagerImpl {
         return new ManagedRequestHandler(handler);
     }
 
-    synchronized int queueSize() {
-        int sum = 0;
-        for (ScheduledQueue schedule : schedules) {
-            sum += schedule.queueSize();
-        }
-        return sum;
-    }
+    synchronized int queueSize() { return scheduler.queueSize(); }
 
     Timer timer() {
         return timer;
@@ -80,9 +68,7 @@ public class TimeoutManagerImpl {
 
     void checkTasks(long currentTimeMillis) {
         Queue<Object> queue = new LinkedList<>();
-        for (ScheduledQueue schedule : schedules) {
-            schedule.drainTo(currentTimeMillis, queue);
-        }
+        scheduler.drainTo(currentTimeMillis, queue);
         while (!queue.isEmpty()) {
             TimeoutHandler timeoutHandler = (TimeoutHandler)queue.poll();
             invokeTimeout(timeoutHandler.requestHandler, timeoutHandler.request, timeoutHandler);
@@ -92,7 +78,7 @@ public class TimeoutManagerImpl {
     private void invokeTimeout(RequestHandler requestHandler, Request request, ResponseHandler responseHandler) {
         try {
             requestHandler.handleTimeout(request, responseHandler);
-        } catch (RuntimeException e) {
+        } catch (Exception e) {
             log.log(Level.WARNING, "Ignoring exception thrown by " + requestHandler.getClass().getName() +
                                    " in timeout manager.", e);
         }
@@ -204,7 +190,7 @@ public class TimeoutManagerImpl {
                 return;
             }
             if (timeoutQueueEntry == null) {
-                timeoutQueueEntry = schedules[(nextScheduler.incrementAndGet() & 0xffff) % schedules.length].newEntry(this);
+                timeoutQueueEntry = scheduler.newEntry(this);
             }
             timeoutQueueEntry.scheduleAt(request.creationTime(TimeUnit.MILLISECONDS) + request.getTimeout(TimeUnit.MILLISECONDS));
         }