opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/AbstractClientConnection.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 edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  14 import java.util.Optional;
  15 import java.util.concurrent.TimeUnit;
  16 import java.util.concurrent.locks.Lock;
  17 import java.util.concurrent.locks.ReentrantLock;
  18 import java.util.function.Consumer;
  19 import javax.annotation.Nonnull;
  20 import javax.annotation.concurrent.GuardedBy;
  21 import javax.annotation.concurrent.NotThreadSafe;
  22 import org.opendaylight.controller.cluster.access.concepts.Request;
  23 import org.opendaylight.controller.cluster.access.concepts.RequestException;
  24 import org.opendaylight.controller.cluster.access.concepts.Response;
  25 import org.opendaylight.controller.cluster.access.concepts.ResponseEnvelope;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28 import scala.concurrent.duration.FiniteDuration;
  29
  30 /**
  31  * Base class for a connection to the backend. Responsible to queueing and dispatch of requests toward the backend.
  32  * Can be in three conceptual states: Connecting, Connected and Reconnecting, which are represented by public final
  33  * classes exposed from this package.
  34  *
  35  * @author Robert Varga
  36  */
  37 @NotThreadSafe
  38 public abstract class AbstractClientConnection<T extends BackendInfo> {
  39     private static final Logger LOG = LoggerFactory.getLogger(AbstractClientConnection.class);
  40
  41     // Keep these constants in nanoseconds, as that prevents unnecessary conversions in the fast path
  42     @VisibleForTesting
  43     static final long NO_PROGRESS_TIMEOUT_NANOS = TimeUnit.MINUTES.toNanos(15);
  44     @VisibleForTesting
  45     static final long REQUEST_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(30);
  46
  47     private final Lock lock = new ReentrantLock();
  48     private final ClientActorContext context;
  49     @GuardedBy("lock")
  50     private final TransmitQueue queue;
  51     private final Long cookie;
  52
  53     private volatile RequestException poisoned;
  54     private long lastProgress;
  55
  56     // Do not allow subclassing outside of this package
  57     AbstractClientConnection(final ClientActorContext context, final Long cookie,
  58             final TransmitQueue queue) {
  59         this.context = Preconditions.checkNotNull(context);
  60         this.cookie = Preconditions.checkNotNull(cookie);
  61         this.queue = Preconditions.checkNotNull(queue);
  62         this.lastProgress = readTime();
  63     }
  64
  65     // Do not allow subclassing outside of this package
  66     AbstractClientConnection(final AbstractClientConnection<T> oldConnection) {
  67         this.context = oldConnection.context;
  68         this.cookie = oldConnection.cookie;
  69         this.lastProgress = oldConnection.lastProgress;
  70         this.queue = new TransmitQueue.Halted();
  71     }
  72
  73     public final ClientActorContext context() {
  74         return context;
  75     }
  76
  77     public final @Nonnull Long cookie() {
  78         return cookie;
  79     }
  80
  81     public final ActorRef localActor() {
  82         return context.self();
  83     }
  84
  85     /**
  86      * Send a request to the backend and invoke a specified callback when it finishes. This method is safe to invoke
  87      * from any thread.
  88      *
  89      * @param request Request to send
  90      * @param callback Callback to invoke
  91      */
  92     public final void sendRequest(final Request<?, ?> request, final Consumer<Response<?, ?>> callback) {
  93         final RequestException maybePoison = poisoned;
  94         if (maybePoison != null) {
  95             throw new IllegalStateException("Connection " + this + " has been poisoned", maybePoison);
  96         }
  97
  98         final ConnectionEntry entry = new ConnectionEntry(request, callback, readTime());
  99
 100         lock.lock();
 101         try {
 102             queue.enqueue(entry, entry.getEnqueuedTicks());
 103         } finally {
 104             lock.unlock();
 105         }
 106     }
 107
 108     public abstract Optional<T> getBackendInfo();
 109
 110     final Iterable<ConnectionEntry> startReplay() {
 111         lock.lock();
 112         return queue.asIterable();
 113     }
 114
 115     @GuardedBy("lock")
 116     final void finishReplay(final ReconnectForwarder forwarder) {
 117         queue.setForwarder(forwarder, readTime());
 118         lock.unlock();
 119     }
 120
 121     @GuardedBy("lock")
 122     final void setForwarder(final ReconnectForwarder forwarder) {
 123         queue.setForwarder(forwarder, readTime());
 124     }
 125
 126     @GuardedBy("lock")
 127     abstract ClientActorBehavior<T> reconnectConnection(ClientActorBehavior<T> current);
 128
 129     private long readTime() {
 130         return context.ticker().read();
 131     }
 132
 133     final void enqueueEntry(final ConnectionEntry entry, final long now) {
 134         lock.lock();
 135         try {
 136             queue.enqueue(entry, now);
 137         } finally {
 138             lock.unlock();
 139         }
 140     }
 141
 142     /**
 143      * Schedule a timer to fire on the actor thread after a delay.
 144      *
 145      * @param delay Delay, in nanoseconds
 146      */
 147     private void scheduleTimer(final FiniteDuration delay) {
 148         LOG.debug("{}: scheduling timeout in {}", context.persistenceId(), delay);
 149         context.executeInActor(this::runTimer, delay);
 150     }
 151
 152     /**
 153      * Check this queue for timeout and initiate reconnection if that happened. If the queue has not made progress
 154      * in {@link #NO_PROGRESS_TIMEOUT_NANOS} nanoseconds, it will be aborted.
 155      *
 156      * @param current Current behavior
 157      * @return Next behavior to use
 158      */
 159     @VisibleForTesting
 160     final ClientActorBehavior<T> runTimer(final ClientActorBehavior<T> current) {
 161         final Optional<FiniteDuration> delay;
 162
 163         lock.lock();
 164         try {
 165             final long now = readTime();
 166             if (!queue.isEmpty()) {
 167                 final long ticksSinceProgress = now - lastProgress;
 168                 if (ticksSinceProgress >= NO_PROGRESS_TIMEOUT_NANOS) {
 169                     LOG.error("Queue {} has not seen progress in {} seconds, failing all requests", this,
 170                         TimeUnit.NANOSECONDS.toSeconds(ticksSinceProgress));
 171
 172                     lockedPoison(new NoProgressException(ticksSinceProgress));
 173                     current.removeConnection(this);
 174                     return current;
 175                 }
 176             }
 177
 178             // Requests are always scheduled in sequence, hence checking for timeout is relatively straightforward.
 179             // Note we use also inquire about the delay, so we can re-schedule if needed, hence the unusual tri-state
 180             // return convention.
 181             delay = lockedCheckTimeout(now);
 182             if (delay == null) {
 183                 // We have timed out. There is no point in scheduling a timer
 184                 return reconnectConnection(current);
 185             }
 186         } finally {
 187             lock.unlock();
 188         }
 189
 190         if (delay.isPresent()) {
 191             // If there is new delay, schedule a timer
 192             scheduleTimer(delay.get());
 193         }
 194
 195         return current;
 196     }
 197
 198     @VisibleForTesting
 199     final Optional<FiniteDuration> checkTimeout(final long now) {
 200         lock.lock();
 201         try {
 202             return lockedCheckTimeout(now);
 203         } finally {
 204             lock.unlock();
 205         }
 206     }
 207
 208     /*
 209      * We are using tri-state return here to indicate one of three conditions:
 210      * - if there is no timeout to schedule, return Optional.empty()
 211      * - if there is a timeout to schedule, return a non-empty optional
 212      * - if this connections has timed out, return null
 213      */
 214     @SuppressFBWarnings(value = "NP_OPTIONAL_RETURN_NULL",
 215             justification = "Returning null Optional is documented in the API contract.")
 216     @GuardedBy("lock")
 217     private Optional<FiniteDuration> lockedCheckTimeout(final long now) {
 218         final ConnectionEntry head = queue.peek();
 219         if (head == null) {
 220             return Optional.empty();
 221         }
 222
 223         final long delay = head.getEnqueuedTicks() - now + REQUEST_TIMEOUT_NANOS;
 224         if (delay <= 0) {
 225             LOG.debug("Connection {} timed out", this);
 226             return null;
 227         }
 228
 229         return Optional.of(FiniteDuration.apply(delay, TimeUnit.NANOSECONDS));
 230     }
 231
 232     final void poison(final RequestException cause) {
 233         lock.lock();
 234         try {
 235             lockedPoison(cause);
 236         } finally {
 237             lock.unlock();
 238         }
 239     }
 240
 241     @GuardedBy("lock")
 242     private void lockedPoison(final RequestException cause) {
 243         poisoned = cause;
 244         queue.poison(cause);
 245     }
 246
 247     @VisibleForTesting
 248     final RequestException poisoned() {
 249         return poisoned;
 250     }
 251
 252     final void receiveResponse(final ResponseEnvelope<?> envelope) {
 253         final long now = readTime();
 254
 255         lock.lock();
 256         try {
 257             queue.complete(envelope, now);
 258         } finally {
 259             lock.unlock();
 260         }
 261
 262         lastProgress = readTime();
 263     }
 264 }