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 org.slf4j.Logger;
  13 import org.slf4j.LoggerFactory;
  14
  15 /**
  16  * Base class for tracking throughput and computing delays when processing stream of tasks.
  17  *
  18  * <p>The idea is to improve throughput in a typical request-response scenario.
  19  * Multiple "user" threads are submitting requests to a "frontend". The frontend does some
  20  * pre-processing and then sends requests to (usually one) "backend". The backend does the main work
  21  * and replies to the frontend, which reports to the the corresponding user.
  22  * In terms of task processing, user threads are "opening" tasks and frontend is "closing" them.
  23  * Latency of the backend may fluctuate wildly. To avoid backend running out of open tasks,
  24  * frontend should maintain a queue of requests for users to submit tasks to.
  25  * In order to avoid excessive memory consumption, there should be a back-pressure mechanism
  26  * which blocks the user (submit) threads for appropriate durations.
  27  * Users can tolerate moderately delayed responses, but they only tolerate small block (submit)
  28  * times.
  29  *
  30  * <p>An ideal back-pressure algorithm would keep the queue reasonably full,
  31  * while fairly delaying the user threads. In other words, backend idle time should be low,
  32  * as well as user 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, submit thread is supposed to block itself
  38  * when asked to. The delay computation is pessimistic, it expects each participating thread
  39  * to enqueue another task as soon as its delay time allows.
  40  *
  41  * <p>This class is to be used by single frontend. This class is not thread safe,
  42  * the frontend is responsible for guarding against conflicting access.
  43  * Time is measured in ticks (nanoseconds), methods never look at current time,
  44  * relying on {@code now} argument where appropriate.
  45  * This means the sequence of {@code now} argument values is expected to be non-decreasing.
  46  *
  47  * <p>Input data used for tracking is tightly coupled with TransitQueue#recordCompletion arguments.
  48  *
  49  * @author Vratko Polak
  50  */
  51 // TODO: Would bulk methods be less taxing than a loop of single task calls?
  52 abstract class ProgressTracker {
  53     private static final Logger LOG = LoggerFactory.getLogger(ProgressTracker.class);
  54
  55     /**
  56      * When no tasks has been closed yet, this will be used to estimate throughput.
  57      */
  58     private final long defaultTicksPerTask;
  59
  60     /**
  61      * Number of tasks closed so far.
  62      */
  63     private long tasksClosed = 0;
  64
  65     /**
  66      * Number of tasks so far, both open and closed.
  67      */
  68     private long tasksEncountered = 0;
  69
  70     /**
  71      * The most recent tick number when the number of open tasks has become non-positive.
  72      */
  73     private long lastIdle = Long.MIN_VALUE;
  74
  75     /**
  76      * The most recent tick number when a task has been closed.
  77      */
  78     private long lastClosed = Long.MIN_VALUE;
  79
  80     /**
  81      * Tick number when the farthest known wait time is over.
  82      */
  83     private long nearestAllowed = Long.MIN_VALUE;
  84
  85     /**
  86      * Number of ticks elapsed before lastIdle while there was at least one open task.
  87      */
  88     private long elapsedBeforeIdle = 0L;
  89
  90     // Constructors
  91
  92     /**
  93      * Construct an idle tracker with specified ticks per task value to use as default.
  94      *
  95      * @param ticksPerTask value to use as default
  96      */
  97     ProgressTracker(final long ticksPerTask) {
  98         Preconditions.checkArgument(ticksPerTask >= 0);
  99         defaultTicksPerTask = ticksPerTask;
 100     }
 101
 102     /**
 103      * Construct a new tracker suitable for a new task queue related to a "reconnect".
 104      *
 105      * <p>When reconnecting to a new backend, tasks may need to be re-processed by the frontend,
 106      * possibly resulting in a different number of tasks.
 107      * Also, performance of the new backend can be different, but the perforance of the previous backend
 108      * is generally still better estimate than defaults of a brand new tracker.
 109      *
 110      * <p>This "inheritance constructor" creates a new tracker with no open tasks (thus initially idle),
 111      * but other internal values should lead to a balanced performance
 112      * after tasks opened in the source tracker are "replayed" into the new tracker.
 113      *
 114      * <p>In particular, this impementation keeps the number of closed tasks the same,
 115      * and makes it so ticksWorkedPerClosedTask is initially the same as in the old tracker.
 116      *
 117      * @param oldTracker the tracker used for the previously used backend
 118      * @param now tick number corresponding to caller's present
 119      */
 120     ProgressTracker(final ProgressTracker oldTracker, final long now) {
 121         defaultTicksPerTask = oldTracker.defaultTicksPerTask;
 122         tasksEncountered = tasksClosed = oldTracker.tasksClosed;
 123         lastClosed = oldTracker.lastClosed;
 124         // Call cancelDebt explicitly if needed.
 125         nearestAllowed = oldTracker.nearestAllowed;
 126         lastIdle = oldTracker.lastIdle;
 127         elapsedBeforeIdle = oldTracker.elapsedBeforeIdle;
 128         if (!oldTracker.isIdle()) {
 129             transitToIdle(now);
 130         }
 131     }
 132
 133     // "Public" shared access (read-only) accessor-like methods
 134
 135     /**
 136      * Get number of tasks closed so far.
 137      *
 138      * @return number of tasks known to be finished already; the value never decreases
 139      */
 140     final long tasksClosed() {
 141         return tasksClosed;
 142     }
 143
 144     /**
 145      * Get umber of tasks so far, both open and closed.
 146      *
 147      * @return number of tasks encountered so far, open or finished; the value never decreases
 148      */
 149     final long tasksEncountered() {
 150         return tasksEncountered;
 151     }
 152
 153     /**
 154      * Get number of tasks currently open.
 155      *
 156      * @return number of tasks started but not finished yet
 157      */
 158     // TODO: Should we return int?
 159     final long tasksOpen() {
 160         // TODO: Should we check the return value is non-negative?
 161         return tasksEncountered - tasksClosed;
 162     }
 163
 164     /**
 165      * When idle, there are no open tasks so no progress is made.
 166      *
 167      * @return {@code true} if every encountered task is already closed, {@code false} otherwise
 168      */
 169     final boolean isIdle() {
 170         return tasksClosed >= tasksEncountered;
 171     }
 172
 173     /**
 174      * Number of ticks elapsed (before now) since the last closed task while there was at least one open task.
 175      *
 176      * @param now tick number corresponding to caller's present
 177      * @return number of ticks backend is neither idle nor responding
 178      */
 179     final long ticksStalling(final long now) {
 180         return isIdle() ? 0 : Math.max(now, lastClosed) - lastClosed;
 181     }
 182
 183     // Read only protected methods.
 184
 185     /**
 186      * Get the value of default ticks per task this instance was created to use.
 187      *
 188      * @return default ticks per task value
 189      */
 190     protected final long defaultTicksPerTask() {
 191         return defaultTicksPerTask;
 192     }
 193
 194     /**
 195      * One task is roughly estimated to take this long to close.
 196      *
 197      * @param now tick number corresponding to caller's present
 198      * @return total ticks worked divided by closed tasks, or the default value if no closed tasks
 199      */
 200     protected final double ticksWorkedPerClosedTask(final long now) {
 201         if (tasksClosed < 1) {
 202             return defaultTicksPerTask;
 203         }
 204         return (double) ticksWorked(now) / tasksClosed;
 205     }
 206
 207     // Read only private methods.
 208
 209     /**
 210      * Number of ticks elapsed (before now) while there was at least one open task.
 211      *
 212      * @param now tick number corresponding to caller's present
 213      * @return number of ticks there was at least one task open
 214      */
 215     private long ticksWorked(final long now) {
 216         return isIdle() ? elapsedBeforeIdle : Math.max(now, lastIdle) - lastIdle + elapsedBeforeIdle;
 217     }
 218
 219     /**
 220      * Give an estimate of a tick number when there will be no accumulated delays.
 221      *
 222      * <p>The delays accumulated include one more open task.
 223      * Basically, the return value corresponds to openTask() return value,
 224      * but this gives an absolute time, instead of delay relative to now.
 225      *
 226      * @param now tick number corresponding to caller's present
 227      * @return estimated tick number when all threads with opened tasks are done waiting
 228      */
 229     private long estimateAllowed(final long now) {
 230         return Math.max(now, nearestAllowed + estimateIsolatedDelay(now));
 231     }
 232
 233     // State-altering "public" methods.
 234
 235     /**
 236      * Track a task is being closed.
 237      *
 238      * @param now tick number corresponding to caller's present
 239      * @param enqueuedTicks see TransitQueue#recordCompletion
 240      * @param transmitTicks see TransitQueue#recordCompletion
 241      * @param execNanos see TransitQueue#recordCompletion
 242      */
 243     final void closeTask(final long now, final long enqueuedTicks, final long transmitTicks, final long execNanos) {
 244         if (isIdle()) {
 245             LOG.info("Attempted to close a task while no tasks are open");
 246         } else {
 247             unsafeCloseTask(now, enqueuedTicks, transmitTicks, execNanos);
 248         }
 249     }
 250
 251     /**
 252      * Track a task that is being opened.
 253      *
 254      * @param now tick number corresponding to caller's present
 255      * @return number of ticks (nanos) the caller thread should wait before opening another task
 256      */
 257     final long openTask(final long now) {
 258         openTaskWithoutThrottle(now);
 259         return reserveDelay(now);
 260     }
 261
 262     /**
 263      * Set nearestAllowed value to now.
 264      *
 265      * <p>This is useful when new a backend has just connected,
 266      * after a period of no working backend present.
 267      * The accumulated delays should not limit future tasks.
 268      * The queue fullness and the averaged backend performance are kept,
 269      * as they should result in good enough estimations for new tasks.
 270      *
 271      * @param now tick number corresponding to caller's present
 272      */
 273     final void cancelDebt(final long now) {
 274         nearestAllowed = now;
 275     }
 276
 277     // Private state-altering methods.
 278
 279     /**
 280      * Compute the next delay and update nearestAllowed value accordingly.
 281      *
 282      * @param now tick number corresponding to caller's present
 283      * @return number of ticks (nanos) the caller thread should wait before opening another task
 284      */
 285     private long reserveDelay(final long now) {
 286         nearestAllowed = estimateAllowed(now);
 287         return nearestAllowed - now;
 288     }
 289
 290     /**
 291      * Track a task is being closed.
 292      *
 293      * <p>This method does not verify there was any task open.
 294      * This call can empty the collection of open tasks, that special case should be handled.
 295      *
 296      * @param now tick number corresponding to caller's present
 297      * @param enqueuedTicks see TransmitQueue#recordCompletion
 298      * @param transmitTicks see TransmitQueue#recordCompletion
 299      * @param execNanos see TransmitQueue#recordCompletion
 300      */
 301     private void unsafeCloseTask(final long now, final long enqueuedTicks, final long transmitTicks,
 302                 final long execNanos) {
 303         tasksClosed++;
 304         lastClosed = now;
 305         if (isIdle()) {
 306             transitToIdle(now);
 307         }
 308     }
 309
 310     /**
 311      * Track a task is being opened.
 312      *
 313      * <p>This method does not aggregate delays, allowing the caller to sidestep the throttling.
 314      * This call can make the collection of open tasks non-empty, that special case should be handled.
 315      *
 316      * @param now tick number corresponding to caller's present
 317      */
 318     private void openTaskWithoutThrottle(final long now) {
 319         if (isIdle()) {
 320             transitFromIdle(now);
 321         }
 322         tasksEncountered++;
 323     }
 324
 325     /**
 326      * Update lastIdle as a new "last" just hapened.
 327      */
 328     private void transitFromIdle(final long now) {
 329         lastIdle = Math.max(now, lastIdle);
 330     }
 331
 332     /**
 333      * Update elapsedBeforeIdle as the "before" has jast moved.
 334      */
 335     private void transitToIdle(final long now) {
 336         elapsedBeforeIdle += Math.max(0, now - lastIdle);
 337     }
 338
 339     // Protected abstract read-only methods.
 340
 341     /**
 342      * Give an estimate of a fair delay, assuming delays caused by other opened tasks are ignored.
 343      *
 344      * @param now tick number corresponding to caller's present
 345      * @return delay (in ticks) after which another openTask() would be fair to be called by the same thread again
 346      */
 347     protected abstract long estimateIsolatedDelay(long now);
 348 }