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