opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/ProgressTracker.java

   1 /*
   2  * Copyright (c) 2016, 2017 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.access.client;
  10
  11 import com.google.common.base.Preconditions;
  12 import javax.annotation.concurrent.NotThreadSafe;
  13 import org.slf4j.Logger;
  14 import org.slf4j.LoggerFactory;
  15
  16 /**
  17  * Base class for tracking throughput and computing delays when processing stream of tasks.
  18  *
  19  * <p>The idea is to improve throughput in a typical request-response scenario.
  20  * A "frontend" is sending requests towards "backend", backend is sending responses back to fronted.
  21  * Both frontend and backend may be realized by multiple Java threads,
  22  * so there may be multiple requests not yet responded to.
  23  * In terms of taks processing, frontend is "opening" tasks and backend is "closing" them.
  24  * Latency of the backend may fluctuate wildly. To avoid backend running out of open tasks,
  25  * there should be a queue of requests frontend can add to.
  26  * In order to avoid excessive memory consumption, there should be a back-pressure mechanism
  27  * which blocks the frontend threads for appropriate durations.
  28  * Frontend can tolerate moderately delayed responses, but it only tolerates small block times.
  29  *
  30  * <p>An ideal back-pressure algorithm would keep the queue reasonably full,
  31  * while fairly delaying the frontend threads. In other words, backend idle time should be low,
  32  * as well as frontend block time dispersion
  33  * (as opposed to block time average, which is dictated by overall performance).
  34  *
  35  * <p>In order for an algorithm to compute reasonable wait times,
  36  * various inputs can be useful, mostly related to timing of various stages of task processing.
  37  * Methods of this class assume "enqueue and wait" usage.
  38  * The delay computation is pessimistic, it expects each participating thread to enqueue another task
  39  * as soon as its delay time allows.
  40  *
  41  * <p>This class is not thread safe, the callers are responsible for guarding against conflicting access.
  42  * Time is measured in ticks (nanos), methods never look at current time, relying on {@code now} argument instead.
  43  * This means the sequence of {$code now} argument values is expected to be non-decreasing.
  44  *
  45  * <p>Input data used for tracking is tightly coupled with TransitQueue#recordCompletion arguments.
  46  *
  47  * @author Vratko Polak
  48  */
  49 // TODO: Would bulk methods be less taxing than a loop of single task calls?
  50 @NotThreadSafe
  51 abstract class ProgressTracker {
  52     private static final Logger LOG = LoggerFactory.getLogger(ProgressTracker.class);
  53
  54     /**
  55      * When no tasks has been closed yet, this will be used to estimate throughput.
  56      */
  57     private final long defaultTicksPerTask;
  58
  59     /**
  60      * Number of tasks closed so far.
  61      */
  62     private long tasksClosed = 0;
  63
  64     /**
  65      * Number of tasks so far, both open and closed.
  66      */
  67     private long tasksEncountered = 0;
  68
  69     /**
  70      * The most recent tick number when the number of open tasks has become non-positive.
  71      */
  72     private long lastIdle = Long.MIN_VALUE;
  73
  74     /**
  75      * The most recent tick number when a task has been closed.
  76      */
  77     private long lastClosed = Long.MIN_VALUE;
  78
  79     /**
  80      * Tick number when the farthest known wait time is over.
  81      */
  82     private long nearestAllowed = Long.MIN_VALUE;
  83
  84     /**
  85      * Number of ticks elapsed before lastIdle while there was at least one open task.
  86      */
  87     private long elapsedBeforeIdle = 0L;
  88
  89     // Constructors
  90
  91     /**
  92      * Construct an idle tracker with specified ticks per task value to use as default.
  93      *
  94      * @param ticksPerTask value to use as default
  95      */
  96     ProgressTracker(final long ticksPerTask) {
  97         Preconditions.checkArgument(ticksPerTask >= 0);
  98         defaultTicksPerTask = ticksPerTask;
  99     }
 100
 101     /**
 102      * Construct a copy of an existing tracker, all future tracking is fully independent.
 103      *
 104      * @param tracker the instance to copy state from
 105      */
 106     ProgressTracker(final ProgressTracker tracker) {
 107         this.defaultTicksPerTask = tracker.defaultTicksPerTask;
 108         this.tasksClosed = tracker.tasksClosed;
 109         this.tasksEncountered = tracker.tasksEncountered;
 110         this.lastClosed = tracker.lastClosed;
 111         this.lastIdle = tracker.lastIdle;
 112         this.nearestAllowed = tracker.nearestAllowed;
 113         this.elapsedBeforeIdle = tracker.elapsedBeforeIdle;
 114     }
 115
 116     // Public shared access (read-only) accessor-like methods
 117
 118     /**
 119      * Get the value of default ticks per task this instance was created to use.
 120      *
 121      * @return default ticks per task value
 122      */
 123     public final long defaultTicksPerTask() {
 124         return defaultTicksPerTask;
 125     }
 126
 127     /**
 128      * Get number of tasks closed so far.
 129      *
 130      * @return number of tasks known to be finished already; the value never decreases
 131      */
 132     public final long tasksClosed() {
 133         return tasksClosed;
 134     }
 135
 136     /**
 137      * Get umber of tasks so far, both open and closed.
 138      *
 139      * @return number of tasks encountered so far, open or finished; the value never decreases
 140      */
 141     public final long tasksEncountered() {
 142         return tasksEncountered;
 143     }
 144
 145     /**
 146      * Get number of tasks currently open.
 147      *
 148      * @return number of tasks started but not finished yet
 149      */
 150     public final long tasksOpen() {
 151         // TODO: Should we check the return value is non-negative?
 152         return tasksEncountered - tasksClosed;
 153     }
 154
 155     /**
 156      * When idle, there are no open tasks so no progress is made.
 157      *
 158      * @return {@code true} if every encountered task is already closed, {@code false} otherwise
 159      */
 160     public boolean isIdle() {
 161         return tasksClosed >= tasksEncountered;
 162     }
 163
 164     /**
 165      * Number of ticks elapsed (before now) since the last closed task while there was at least one open task.
 166      *
 167      * @param now tick number corresponding to caller's present
 168      * @return number of ticks backend is neither idle nor responding
 169      */
 170     public long ticksStalling(final long now) {
 171         return isIdle() ? 0 : Math.max(now, lastClosed) - lastClosed;
 172     }
 173
 174     /**
 175      * Number of ticks elapsed (before now) while there was at least one open task.
 176      *
 177      * @param now tick number corresponding to caller's present
 178      * @return number of ticks there was at least one task open
 179      */
 180     public long ticksWorked(final long now) {
 181         return isIdle() ? elapsedBeforeIdle : Math.max(now, lastIdle) - lastIdle + elapsedBeforeIdle;
 182     }
 183
 184     /**
 185      * One task is roughly estimated to take this long to close.
 186      *
 187      * @param now tick number corresponding to caller's present
 188      * @return total ticks worked divided by closed tasks, or the default value if no closed tasks
 189      */
 190     public double ticksWorkedPerClosedTask(final long now) {
 191         if (tasksClosed < 1) {
 192             return defaultTicksPerTask;
 193         }
 194         return (double) ticksWorked(now) / tasksClosed;
 195     }
 196
 197     /**
 198      * Give an estimate of openTask() return value.
 199      *
 200      * <p>When the returned delay is positive, the caller thread should wait that time before opening additional task.
 201      *
 202      * <p>This method in general takes into account previously assigned delays to avoid overlaps.
 203      *
 204      * @param now tick number corresponding to caller's present
 205      * @return delay (in ticks) after which another openTask() is fair to be called by the same thread again
 206      */
 207     public long estimateDelay(final long now) {
 208         return estimateAllowed(now) - now;
 209     }
 210
 211     /**
 212      * Give an estimate of a tick number when there will be no accumulated delays.
 213      *
 214      * <p>The delays accumulated include one more open task.
 215      * Basically, the return value corresponds to openTask() return value,
 216      * but this gives an absolute time, instead of delay relative to now.
 217      *
 218      * @param now tick number corresponding to caller's present
 219      * @return estimated tick number when all threads with opened tasks are done waiting
 220      */
 221     public long estimateAllowed(final long now) {
 222         return Math.max(now, nearestAllowed) + estimateIsolatedDelay(now);
 223     }
 224
 225     // State-altering public methods.
 226
 227     /**
 228      * Track a task is being closed.
 229      *
 230      * @param now tick number corresponding to caller's present
 231      * @param enqueuedTicks see TransitQueue#recordCompletion
 232      * @param transmitTicks see TransitQueue#recordCompletion
 233      * @param execNanos see TransitQueue#recordCompletion
 234      */
 235     public void closeTask(final long now, final long enqueuedTicks, final long transmitTicks, final long execNanos) {
 236         if (isIdle()) {
 237             LOG.info("Attempted to close a task while no tasks are open");
 238         } else {
 239             protectedCloseTask(now, enqueuedTicks, transmitTicks, execNanos);
 240         }
 241     }
 242
 243     /**
 244      * Track a task that is being opened.
 245      *
 246      * @param now tick number corresponding to caller's present
 247      * @return number of ticks (nanos) the caller thread should wait before opening another task
 248      */
 249     public long openTask(final long now) {
 250         protectedOpenTask(now);
 251         return reserveDelay(now);
 252     }
 253
 254     // Internal state-altering methods. Protected instead of private,
 255     // allowing subclasses to weaken ad-hoc invariants of current implementation.
 256
 257     /**
 258      * Compute the next delay and update nearestAllowed value accordingly.
 259      *
 260      * @param now tick number corresponding to caller's present
 261      * @return number of ticks (nanos) the caller thread should wait before opening another task
 262      */
 263     protected long reserveDelay(final long now) {
 264         nearestAllowed = estimateAllowed(now);
 265         return nearestAllowed - now;
 266     }
 267
 268     /**
 269      * Track a task is being closed.
 270      *
 271      * <p>This method does not verify there was any task open.
 272      * This call can empty the collection of open tasks, that special case should be handled.
 273      *
 274      * @param now tick number corresponding to caller's present
 275      * @param enqueuedTicks see TransmitQueue#recordCompletion
 276      * @param transmitTicks see TransmitQueue#recordCompletion
 277      * @param execNanos see TransmitQueue#recordCompletion
 278      */
 279     protected void protectedCloseTask(final long now, final long enqueuedTicks, final long transmitTicks,
 280                 final long execNanos) {
 281         tasksClosed++;
 282         lastClosed = now;
 283         if (isIdle()) {
 284             elapsedBeforeIdle += now - lastIdle;
 285         }
 286     }
 287
 288     /**
 289      * Track a task is being opened.
 290      *
 291      * <p>This method does not aggregate delays, allowing the caller to sidestep the throttling.
 292      * This call can make the collection of open tasks non-empty, that special case should be handled.
 293      *
 294      * @param now tick number corresponding to caller's present
 295      */
 296     protected void protectedOpenTask(final long now) {
 297         if (isIdle()) {
 298             lastIdle = Math.max(now, lastIdle);
 299         }
 300         tasksEncountered++;
 301     }
 302
 303     /**
 304      * Give an estimate of a fair delay, assuming delays caused by other opened tasks are ignored.
 305      *
 306      * @param now tick number corresponding to caller's present
 307      * @return delay (in ticks) after which another openTask() would be fair to be called by the same thread again
 308      */
 309     abstract long estimateIsolatedDelay(long now);
 310 }