7dc150e403dc283a2181aad081fb1757afc690ea
[controller.git] / 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     // Updated from actor thread only
54     private long lastProgress;
55
56     private volatile RequestException poisoned;
57
58     // Do not allow subclassing outside of this package
59     AbstractClientConnection(final ClientActorContext context, final Long cookie,
60             final TransmitQueue queue) {
61         this.context = Preconditions.checkNotNull(context);
62         this.cookie = Preconditions.checkNotNull(cookie);
63         this.queue = Preconditions.checkNotNull(queue);
64         this.lastProgress = readTime();
65     }
66
67     // Do not allow subclassing outside of this package
68     AbstractClientConnection(final AbstractClientConnection<T> oldConnection) {
69         this.context = oldConnection.context;
70         this.cookie = oldConnection.cookie;
71         this.lastProgress = oldConnection.lastProgress;
72         this.queue = new TransmitQueue.Halted();
73     }
74
75     public final ClientActorContext context() {
76         return context;
77     }
78
79     public final @Nonnull Long cookie() {
80         return cookie;
81     }
82
83     public final ActorRef localActor() {
84         return context.self();
85     }
86
87     /**
88      * Send a request to the backend and invoke a specified callback when it finishes. This method is safe to invoke
89      * from any thread.
90      *
91      * @param request Request to send
92      * @param callback Callback to invoke
93      */
94     public final void sendRequest(final Request<?, ?> request, final Consumer<Response<?, ?>> callback) {
95         final RequestException maybePoison = poisoned;
96         if (maybePoison != null) {
97             throw new IllegalStateException("Connection " + this + " has been poisoned", maybePoison);
98         }
99
100         final ConnectionEntry entry = new ConnectionEntry(request, callback, readTime());
101
102         lock.lock();
103         try {
104             queue.enqueue(entry, entry.getEnqueuedTicks());
105         } finally {
106             lock.unlock();
107         }
108     }
109
110     public abstract Optional<T> getBackendInfo();
111
112     final Iterable<ConnectionEntry> startReplay() {
113         lock.lock();
114         return queue.asIterable();
115     }
116
117     @GuardedBy("lock")
118     final void finishReplay(final ReconnectForwarder forwarder) {
119         queue.setForwarder(forwarder, readTime());
120         lock.unlock();
121     }
122
123     @GuardedBy("lock")
124     final void setForwarder(final ReconnectForwarder forwarder) {
125         queue.setForwarder(forwarder, readTime());
126     }
127
128     @GuardedBy("lock")
129     abstract ClientActorBehavior<T> reconnectConnection(ClientActorBehavior<T> current);
130
131     private long readTime() {
132         return context.ticker().read();
133     }
134
135     final void enqueueEntry(final ConnectionEntry entry, final long now) {
136         lock.lock();
137         try {
138             queue.enqueue(entry, now);
139         } finally {
140             lock.unlock();
141         }
142     }
143
144     /**
145      * Schedule a timer to fire on the actor thread after a delay.
146      *
147      * @param delay Delay, in nanoseconds
148      */
149     private void scheduleTimer(final FiniteDuration delay) {
150         LOG.debug("{}: scheduling timeout in {}", context.persistenceId(), delay);
151         context.executeInActor(this::runTimer, delay);
152     }
153
154     /**
155      * Check this queue for timeout and initiate reconnection if that happened. If the queue has not made progress
156      * in {@link #NO_PROGRESS_TIMEOUT_NANOS} nanoseconds, it will be aborted.
157      *
158      * @param current Current behavior
159      * @return Next behavior to use
160      */
161     @VisibleForTesting
162     final ClientActorBehavior<T> runTimer(final ClientActorBehavior<T> current) {
163         final Optional<FiniteDuration> delay;
164
165         lock.lock();
166         try {
167             final long now = readTime();
168             if (!queue.isEmpty()) {
169                 final long ticksSinceProgress = now - lastProgress;
170                 if (ticksSinceProgress >= NO_PROGRESS_TIMEOUT_NANOS) {
171                     LOG.error("Queue {} has not seen progress in {} seconds, failing all requests", this,
172                         TimeUnit.NANOSECONDS.toSeconds(ticksSinceProgress));
173
174                     lockedPoison(new NoProgressException(ticksSinceProgress));
175                     current.removeConnection(this);
176                     return current;
177                 }
178             }
179
180             // Requests are always scheduled in sequence, hence checking for timeout is relatively straightforward.
181             // Note we use also inquire about the delay, so we can re-schedule if needed, hence the unusual tri-state
182             // return convention.
183             delay = lockedCheckTimeout(now);
184             if (delay == null) {
185                 // We have timed out. There is no point in scheduling a timer
186                 return reconnectConnection(current);
187             }
188         } finally {
189             lock.unlock();
190         }
191
192         if (delay.isPresent()) {
193             // If there is new delay, schedule a timer
194             scheduleTimer(delay.get());
195         }
196
197         return current;
198     }
199
200     @VisibleForTesting
201     final Optional<FiniteDuration> checkTimeout(final long now) {
202         lock.lock();
203         try {
204             return lockedCheckTimeout(now);
205         } finally {
206             lock.unlock();
207         }
208     }
209
210     /*
211      * We are using tri-state return here to indicate one of three conditions:
212      * - if there is no timeout to schedule, return Optional.empty()
213      * - if there is a timeout to schedule, return a non-empty optional
214      * - if this connections has timed out, return null
215      */
216     @SuppressFBWarnings(value = "NP_OPTIONAL_RETURN_NULL",
217             justification = "Returning null Optional is documented in the API contract.")
218     @GuardedBy("lock")
219     private Optional<FiniteDuration> lockedCheckTimeout(final long now) {
220         final ConnectionEntry head = queue.peek();
221         if (head == null) {
222             return Optional.empty();
223         }
224
225         final long delay = head.getEnqueuedTicks() - now + REQUEST_TIMEOUT_NANOS;
226         if (delay <= 0) {
227             LOG.debug("Connection {} timed out", this);
228             return null;
229         }
230
231         return Optional.of(FiniteDuration.apply(delay, TimeUnit.NANOSECONDS));
232     }
233
234     final void poison(final RequestException cause) {
235         lock.lock();
236         try {
237             lockedPoison(cause);
238         } finally {
239             lock.unlock();
240         }
241     }
242
243     @GuardedBy("lock")
244     private void lockedPoison(final RequestException cause) {
245         poisoned = cause;
246         queue.poison(cause);
247     }
248
249     @VisibleForTesting
250     final RequestException poisoned() {
251         return poisoned;
252     }
253
254     final void receiveResponse(final ResponseEnvelope<?> envelope) {
255         final long now = readTime();
256
257         final Optional<TransmittedConnectionEntry> maybeEntry;
258         lock.lock();
259         try {
260             maybeEntry = queue.complete(envelope, now);
261         } finally {
262             lock.unlock();
263         }
264
265         if (maybeEntry.isPresent()) {
266             final TransmittedConnectionEntry entry = maybeEntry.get();
267             LOG.debug("Completing {} with {}", entry, envelope);
268             entry.complete(envelope.getMessage());
269         }
270
271         lastProgress = readTime();
272     }
273 }