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

   1 /*
   2  * Copyright (c) 2016 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 package org.opendaylight.controller.cluster.access.client;
   9
  10 import akka.actor.ActorRef;
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.base.Verify;
  14 import com.google.common.collect.Iterables;
  15 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  16 import java.util.ArrayDeque;
  17 import java.util.Deque;
  18 import java.util.Iterator;
  19 import java.util.Optional;
  20 import java.util.Queue;
  21 import javax.annotation.concurrent.NotThreadSafe;
  22 import org.opendaylight.controller.cluster.access.concepts.Request;
  23 import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope;
  24 import org.opendaylight.controller.cluster.access.concepts.RequestException;
  25 import org.opendaylight.controller.cluster.access.concepts.Response;
  26 import org.opendaylight.controller.cluster.access.concepts.ResponseEnvelope;
  27 import org.slf4j.Logger;
  28 import org.slf4j.LoggerFactory;
  29
  30 /**
  31  * This queue is internally split into two queues for performance reasons, both memory efficiency and copy
  32  * operations.
  33  *
  34  * <p>
  35  * Entries are always appended to the end, but then they are transmitted to the remote end and do not necessarily
  36  * complete in the order in which they were sent -- hence the head of the queue does not increase linearly,
  37  * but can involve spurious removals of non-head entries.
  38  *
  39  * <p>
  40  * For memory efficiency we want to pre-allocate both queues -- which points to ArrayDeque, but that is very
  41  * inefficient when entries are removed from the middle. In the typical case we expect the number of in-flight
  42  * entries to be an order of magnitude lower than the number of enqueued entries, hence the split.
  43  *
  44  * <p>
  45  * Note that in transient case of reconnect, when the backend gives us a lower number of maximum in-flight entries
  46  * than the previous incarnation, we may end up still moving the pending queue -- but that is a very exceptional
  47  * scenario, hence we consciously ignore it to keep the design relatively simple.
  48  *
  49  * <p>
  50  * This class is not thread-safe, as it is expected to be guarded by {@link AbstractClientConnection}.
  51  *
  52  * @author Robert Varga
  53  */
  54 @NotThreadSafe
  55 abstract class TransmitQueue {
  56     static final class Halted extends TransmitQueue {
  57         Halted(final int targetDepth) {
  58             super(targetDepth);
  59         }
  60
  61         @Override
  62         int canTransmitCount(final int inflightSize) {
  63             return 0;
  64         }
  65
  66         @Override
  67         TransmittedConnectionEntry transmit(final ConnectionEntry entry, final long now) {
  68             throw new UnsupportedOperationException("Attempted to transmit on a halted queue");
  69         }
  70     }
  71
  72     static final class Transmitting extends TransmitQueue {
  73         private final BackendInfo backend;
  74         private long nextTxSequence;
  75
  76         Transmitting(final int targetDepth, final BackendInfo backend) {
  77             super(targetDepth);
  78             this.backend = Preconditions.checkNotNull(backend);
  79         }
  80
  81         @Override
  82         int canTransmitCount(final int inflightSize) {
  83             return backend.getMaxMessages() - inflightSize;
  84         }
  85
  86         @Override
  87         TransmittedConnectionEntry transmit(final ConnectionEntry entry, final long now) {
  88             final RequestEnvelope env = new RequestEnvelope(entry.getRequest().toVersion(backend.getVersion()),
  89                 backend.getSessionId(), nextTxSequence++);
  90
  91             final TransmittedConnectionEntry ret = new TransmittedConnectionEntry(entry, env.getSessionId(),
  92                 env.getTxSequence(), now);
  93             backend.getActor().tell(env, ActorRef.noSender());
  94             return ret;
  95         }
  96     }
  97
  98     private static final Logger LOG = LoggerFactory.getLogger(TransmitQueue.class);
  99
 100     private final Deque<TransmittedConnectionEntry> inflight = new ArrayDeque<>();
 101     private final Deque<ConnectionEntry> pending = new ArrayDeque<>();
 102     private final ProgressTracker tracker;
 103     private ReconnectForwarder successor;
 104
 105     TransmitQueue(final int targetDepth) {
 106         tracker = new AveragingProgressTracker(targetDepth);
 107     }
 108
 109     final Iterable<ConnectionEntry> asIterable() {
 110         return Iterables.concat(inflight, pending);
 111     }
 112
 113     final long ticksStalling(final long now) {
 114         return tracker.ticksStalling(now);
 115     }
 116
 117     final boolean hasSuccessor() {
 118         return successor != null;
 119     }
 120
 121     // If a matching request was found, this will track a task was closed.
 122     final Optional<TransmittedConnectionEntry> complete(final ResponseEnvelope<?> envelope, final long now) {
 123         Optional<TransmittedConnectionEntry> maybeEntry = findMatchingEntry(inflight, envelope);
 124         if (maybeEntry == null) {
 125             LOG.debug("Request for {} not found in inflight queue, checking pending queue", envelope);
 126             maybeEntry = findMatchingEntry(pending, envelope);
 127         }
 128
 129         if (maybeEntry == null || !maybeEntry.isPresent()) {
 130             LOG.warn("No request matching {} found, ignoring response", envelope);
 131             return Optional.empty();
 132         }
 133
 134         final TransmittedConnectionEntry entry = maybeEntry.get();
 135         tracker.closeTask(now, entry.getEnqueuedTicks(), entry.getTxTicks(), envelope.getExecutionTimeNanos());
 136
 137         // We have freed up a slot, try to transmit something
 138         tryTransmit(now);
 139
 140         return Optional.of(entry);
 141     }
 142
 143     final void tryTransmit(final long now) {
 144         final int toSend = canTransmitCount(inflight.size());
 145         if (toSend > 0 && !pending.isEmpty()) {
 146             transmitEntries(toSend, now);
 147         }
 148     }
 149
 150     private void transmitEntries(final int maxTransmit, final long now) {
 151         for (int i = 0; i < maxTransmit; ++i) {
 152             final ConnectionEntry e = pending.poll();
 153             if (e == null) {
 154                 LOG.debug("Queue {} transmitted {} requests", this, i);
 155                 return;
 156             }
 157
 158             transmitEntry(e, now);
 159         }
 160
 161         LOG.debug("Queue {} transmitted {} requests", this, maxTransmit);
 162     }
 163
 164     private void transmitEntry(final ConnectionEntry entry, final long now) {
 165         LOG.debug("Queue {} transmitting entry {}", entry);
 166         // We are not thread-safe and are supposed to be externally-guarded,
 167         // hence send-before-record should be fine.
 168         // This needs to be revisited if the external guards are lowered.
 169         inflight.addLast(transmit(entry, now));
 170     }
 171
 172     /**
 173      * Enqueue an entry, possibly also transmitting it.
 174      *
 175      * @return Delay to be forced on the calling thread, in nanoseconds.
 176      */
 177     final long enqueue(final ConnectionEntry entry, final long now) {
 178         if (successor != null) {
 179             successor.forwardEntry(entry, now);
 180             return 0;
 181         }
 182
 183         // XXX: we should place a guard against incorrect entry sequences:
 184         // entry.getEnqueueTicks() should have non-negative difference from the last entry present in the queues
 185
 186         // Reserve an entry before we do anything that can fail
 187         final long delay = tracker.openTask(now);
 188
 189         /*
 190          * This is defensive to make sure we do not do the wrong thing here and reorder messages if we ever happen
 191          * to have available send slots and non-empty pending queue.
 192          */
 193         final int toSend = canTransmitCount(inflight.size());
 194         if (toSend <= 0) {
 195             LOG.trace("Queue is at capacity, delayed sending of request {}", entry.getRequest());
 196             pending.addLast(entry);
 197             return delay;
 198         }
 199
 200         if (pending.isEmpty()) {
 201             transmitEntry(entry, now);
 202             return delay;
 203         }
 204
 205         pending.addLast(entry);
 206         transmitEntries(toSend, now);
 207         return delay;
 208     }
 209
 210     /**
 211      * Return the number of entries which can be transmitted assuming the supplied in-flight queue size.
 212      */
 213     abstract int canTransmitCount(int inflightSize);
 214
 215     abstract TransmittedConnectionEntry transmit(ConnectionEntry entry, long now);
 216
 217     final boolean isEmpty() {
 218         return inflight.isEmpty() && pending.isEmpty();
 219     }
 220
 221     final ConnectionEntry peek() {
 222         final ConnectionEntry ret = inflight.peek();
 223         if (ret != null) {
 224             return ret;
 225         }
 226
 227         return pending.peek();
 228     }
 229
 230     final void poison(final RequestException cause) {
 231         poisonQueue(inflight, cause);
 232         poisonQueue(pending, cause);
 233     }
 234
 235     final void setForwarder(final ReconnectForwarder forwarder, final long now) {
 236         Verify.verify(successor == null, "Successor {} already set on connection {}", successor, this);
 237         successor = Preconditions.checkNotNull(forwarder);
 238         LOG.debug("Connection {} superseded by {}, splicing queue", this, successor);
 239
 240         ConnectionEntry entry = inflight.poll();
 241         while (entry != null) {
 242             successor.forwardEntry(entry, now);
 243             entry = inflight.poll();
 244         }
 245
 246         entry = pending.poll();
 247         while (entry != null) {
 248             successor.forwardEntry(entry, now);
 249             entry = pending.poll();
 250         }
 251     }
 252
 253     final void remove(final long now) {
 254         final TransmittedConnectionEntry txe = inflight.poll();
 255         if (txe == null) {
 256             final ConnectionEntry entry = pending.pop();
 257             tracker.closeTask(now, entry.getEnqueuedTicks(), 0, 0);
 258         } else {
 259             tracker.closeTask(now, txe.getEnqueuedTicks(), txe.getTxTicks(), 0);
 260         }
 261     }
 262
 263     @VisibleForTesting
 264     Deque<TransmittedConnectionEntry> getInflight() {
 265         return inflight;
 266     }
 267
 268     @VisibleForTesting
 269     Deque<ConnectionEntry> getPending() {
 270         return pending;
 271     }
 272
 273     /*
 274      * We are using tri-state return here to indicate one of three conditions:
 275      * - if a matching entry is found, return an Optional containing it
 276      * - if a matching entry is not found, but it makes sense to keep looking at other queues, return null
 277      * - if a conflicting entry is encountered, indicating we should ignore this request, return an empty Optional
 278      */
 279     @SuppressFBWarnings(value = "NP_OPTIONAL_RETURN_NULL",
 280             justification = "Returning null Optional is documented in the API contract.")
 281     private static Optional<TransmittedConnectionEntry> findMatchingEntry(final Queue<? extends ConnectionEntry> queue,
 282             final ResponseEnvelope<?> envelope) {
 283         // Try to find the request in a queue. Responses may legally come back in a different order, hence we need
 284         // to use an iterator
 285         final Iterator<? extends ConnectionEntry> it = queue.iterator();
 286         while (it.hasNext()) {
 287             final ConnectionEntry e = it.next();
 288             final Request<?, ?> request = e.getRequest();
 289             final Response<?, ?> response = envelope.getMessage();
 290
 291             // First check for matching target, or move to next entry
 292             if (!request.getTarget().equals(response.getTarget())) {
 293                 continue;
 294             }
 295
 296             // Sanity-check logical sequence, ignore any out-of-order messages
 297             if (request.getSequence() != response.getSequence()) {
 298                 LOG.debug("Expecting sequence {}, ignoring response {}", request.getSequence(), envelope);
 299                 return Optional.empty();
 300             }
 301
 302             // Check if the entry has (ever) been transmitted
 303             if (!(e instanceof TransmittedConnectionEntry)) {
 304                 return Optional.empty();
 305             }
 306
 307             final TransmittedConnectionEntry te = (TransmittedConnectionEntry) e;
 308
 309             // Now check session match
 310             if (envelope.getSessionId() != te.getSessionId()) {
 311                 LOG.debug("Expecting session {}, ignoring response {}", te.getSessionId(), envelope);
 312                 return Optional.empty();
 313             }
 314             if (envelope.getTxSequence() != te.getTxSequence()) {
 315                 LOG.warn("Expecting txSequence {}, ignoring response {}", te.getTxSequence(), envelope);
 316                 return Optional.empty();
 317             }
 318
 319             LOG.debug("Completing request {} with {}", request, envelope);
 320             it.remove();
 321             return Optional.of(te);
 322         }
 323
 324         return null;
 325     }
 326
 327     private static void poisonQueue(final Queue<? extends ConnectionEntry> queue, final RequestException cause) {
 328         for (ConnectionEntry e : queue) {
 329             final Request<?, ?> request = e.getRequest();
 330             LOG.trace("Poisoning request {}", request, cause);
 331             e.complete(request.toRequestFailure(cause));
 332         }
 333         queue.clear();
 334     }
 335 }