Cleanup warnings
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / databroker / actors / dds / AbstractProxyTransaction.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.databroker.actors.dds;
9
10 import akka.actor.ActorRef;
11 import com.google.common.base.MoreObjects;
12 import com.google.common.base.Optional;
13 import com.google.common.base.Preconditions;
14 import com.google.common.base.Verify;
15 import com.google.common.collect.Iterables;
16 import com.google.common.util.concurrent.CheckedFuture;
17 import com.google.common.util.concurrent.ListenableFuture;
18 import com.google.common.util.concurrent.SettableFuture;
19 import java.util.ArrayDeque;
20 import java.util.Deque;
21 import java.util.Iterator;
22 import java.util.concurrent.CountDownLatch;
23 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
24 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
25 import java.util.function.Consumer;
26 import javax.annotation.Nonnull;
27 import javax.annotation.Nullable;
28 import javax.annotation.concurrent.GuardedBy;
29 import javax.annotation.concurrent.NotThreadSafe;
30 import org.opendaylight.controller.cluster.access.client.ConnectionEntry;
31 import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
32 import org.opendaylight.controller.cluster.access.commands.ClosedTransactionException;
33 import org.opendaylight.controller.cluster.access.commands.IncrementTransactionSequenceRequest;
34 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
35 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
36 import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
37 import org.opendaylight.controller.cluster.access.commands.TransactionCommitSuccess;
38 import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest;
39 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
40 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitSuccess;
41 import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
42 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
43 import org.opendaylight.controller.cluster.access.concepts.Request;
44 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
45 import org.opendaylight.controller.cluster.access.concepts.Response;
46 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
47 import org.opendaylight.mdsal.common.api.ReadFailedException;
48 import org.opendaylight.yangtools.concepts.Identifiable;
49 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
50 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
51 import org.slf4j.Logger;
52 import org.slf4j.LoggerFactory;
53
54 /**
55  * Class translating transaction operations towards a particular backend shard.
56  *
57  * <p>
58  * This class is not safe to access from multiple application threads, as is usual for transactions. Internal state
59  * transitions coming from interactions with backend are expected to be thread-safe.
60  *
61  * <p>
62  * This class interacts with the queueing mechanism in ClientActorBehavior, hence once we arrive at a decision
63  * to use either a local or remote implementation, we are stuck with it. We can re-evaluate on the next transaction.
64  *
65  * @author Robert Varga
66  */
67 abstract class AbstractProxyTransaction implements Identifiable<TransactionIdentifier> {
68     /**
69      * Marker object used instead of read-type of requests, which are satisfied only once. This has a lower footprint
70      * and allows compressing multiple requests into a single entry.
71      */
72     @NotThreadSafe
73     private static final class IncrementSequence {
74         private final long sequence;
75         private long delta = 0;
76
77         IncrementSequence(final long sequence) {
78             this.sequence = sequence;
79         }
80
81         long getDelta() {
82             return delta;
83         }
84
85         long getSequence() {
86             return sequence;
87         }
88
89         void incrementDelta() {
90             delta++;
91         }
92     }
93
94     /**
95      * Base class for representing logical state of this proxy. See individual instantiations and {@link SuccessorState}
96      * for details.
97      */
98     private static class State {
99         private final String string;
100
101         State(final String string) {
102             this.string = Preconditions.checkNotNull(string);
103         }
104
105         @Override
106         public final String toString() {
107             return string;
108         }
109     }
110
111     /**
112      * State class used when a successor has interfered. Contains coordinator latch, the successor and previous state.
113      * This is a temporary state introduced during reconnection process and is necessary for correct state hand-off
114      * between the old connection (potentially being accessed by the user) and the new connection (being cleaned up
115      * by the actor.
116      *
117      * <p>
118      * When a user operation encounters this state, it synchronizes on the it and wait until reconnection completes,
119      * at which point the request is routed to the successor transaction. This is a relatively heavy-weight solution
120      * to the problem of state transfer, but the user will observe it only if the race condition is hit.
121      */
122     private static class SuccessorState extends State {
123         private final CountDownLatch latch = new CountDownLatch(1);
124         private AbstractProxyTransaction successor;
125         private State prevState;
126
127         // SUCCESSOR + DONE
128         private boolean done;
129
130         SuccessorState() {
131             super("SUCCESSOR");
132         }
133
134         // Synchronize with succession process and return the successor
135         AbstractProxyTransaction await() {
136             try {
137                 latch.await();
138             } catch (InterruptedException e) {
139                 LOG.warn("Interrupted while waiting for latch of {}", successor);
140                 throw new RuntimeException(e);
141             }
142             return successor;
143         }
144
145         void finish() {
146             latch.countDown();
147         }
148
149         State getPrevState() {
150             return Verify.verifyNotNull(prevState, "Attempted to access previous state, which was not set");
151         }
152
153         void setPrevState(final State prevState) {
154             Verify.verify(this.prevState == null, "Attempted to set previous state to %s when we already have %s",
155                     prevState, this.prevState);
156             this.prevState = Preconditions.checkNotNull(prevState);
157             // We cannot have duplicate successor states, so this check is sufficient
158             this.done = DONE.equals(prevState);
159         }
160
161         // To be called from safe contexts, where successor is known to be completed
162         AbstractProxyTransaction getSuccessor() {
163             return Verify.verifyNotNull(successor);
164         }
165
166         void setSuccessor(final AbstractProxyTransaction successor) {
167             Verify.verify(this.successor == null, "Attempted to set successor to %s when we already have %s",
168                     successor, this.successor);
169             this.successor = Preconditions.checkNotNull(successor);
170         }
171
172         boolean isDone() {
173             return done;
174         }
175
176         void setDone() {
177             done = true;
178         }
179     }
180
181     private static final Logger LOG = LoggerFactory.getLogger(AbstractProxyTransaction.class);
182     private static final AtomicIntegerFieldUpdater<AbstractProxyTransaction> SEALED_UPDATER =
183             AtomicIntegerFieldUpdater.newUpdater(AbstractProxyTransaction.class, "sealed");
184     private static final AtomicReferenceFieldUpdater<AbstractProxyTransaction, State> STATE_UPDATER =
185             AtomicReferenceFieldUpdater.newUpdater(AbstractProxyTransaction.class, State.class, "state");
186
187     /**
188      * Transaction has been open and is being actively worked on.
189      */
190     private static final State OPEN = new State("OPEN");
191
192     /**
193      * Transaction has been sealed by the user, but it has not completed flushing to the backed, yet. This is
194      * a transition state, as we are waiting for the user to initiate commit procedures.
195      *
196      * <p>
197      * Since the reconnect mechanics relies on state replay for transactions, this state needs to be flushed into the
198      * queue to re-create state in successor transaction (which may be based on different messages as locality may have
199      * changed). Hence the transition to {@link #FLUSHED} state needs to be handled in a thread-safe manner.
200      */
201     private static final State SEALED = new State("SEALED");
202
203     /**
204      * Transaction state has been flushed into the queue, i.e. it is visible by the successor and potentially
205      * the backend. At this point the transaction does not hold any state besides successful requests, all other state
206      * is held either in the connection's queue or the successor object.
207      *
208      * <p>
209      * Transition to this state indicates we have all input from the user we need to initiate the correct commit
210      * protocol.
211      */
212     private static final State FLUSHED = new State("FLUSHED");
213
214     /**
215      * Transaction state has been completely resolved, we have received confirmation of the transaction fate from
216      * the backend. The only remaining task left to do is finishing up the state cleanup, which is done via purge
217      * request. We need to hang on to the transaction until that is done, as we have to make sure backend completes
218      * purging its state -- otherwise we could have a leak on the backend.
219      */
220     private static final State DONE = new State("DONE");
221
222     // Touched from client actor thread only
223     private final Deque<Object> successfulRequests = new ArrayDeque<>();
224     private final ProxyHistory parent;
225
226     // Accessed from user thread only, which may not access this object concurrently
227     private long sequence;
228
229     /*
230      * Atomic state-keeping is required to synchronize the process of propagating completed transaction state towards
231      * the backend -- which may include a successor.
232      *
233      * Successor, unlike {@link AbstractProxyTransaction#seal()} is triggered from the client actor thread, which means
234      * the successor placement needs to be atomic with regard to the application thread.
235      *
236      * In the common case, the application thread performs performs the seal operations and then "immediately" sends
237      * the corresponding message. The uncommon case is when the seal and send operations race with a connect completion
238      * or timeout, when a successor is injected.
239      *
240      * This leaves the problem of needing to completely transferring state just after all queued messages are replayed
241      * after a successor was injected, so that it can be properly sealed if we are racing. Further complication comes
242      * from lock ordering, where the successor injection works with a locked queue and locks proxy objects -- leading
243      * to a potential AB-BA deadlock in case of a naive implementation.
244      *
245      * For tracking user-visible state we use a single volatile int, which is flipped atomically from 0 to 1 exactly
246      * once in {@link AbstractProxyTransaction#seal()}. That keeps common operations fast, as they need to perform
247      * only a single volatile read to assert state correctness.
248      *
249      * For synchronizing client actor (successor-injecting) and user (commit-driving) thread, we keep a separate state
250      * variable. It uses pre-allocated objects for fast paths (i.e. no successor present) and a per-transition object
251      * for slow paths (when successor is injected/present).
252      */
253     private volatile int sealed;
254     private volatile State state;
255
256     AbstractProxyTransaction(final ProxyHistory parent, final boolean isDone) {
257         this.parent = Preconditions.checkNotNull(parent);
258         if (isDone) {
259             state = DONE;
260             // DONE implies previous seal operation completed
261             sealed = 1;
262         } else {
263             state = OPEN;
264         }
265     }
266
267     final void executeInActor(final Runnable command) {
268         parent.context().executeInActor(behavior -> {
269             command.run();
270             return behavior;
271         });
272     }
273
274     final ActorRef localActor() {
275         return parent.localActor();
276     }
277
278     final void incrementSequence(final long delta) {
279         sequence += delta;
280         LOG.debug("Transaction {} incremented sequence to {}", this, sequence);
281     }
282
283     final long nextSequence() {
284         final long ret = sequence++;
285         LOG.debug("Transaction {} allocated sequence {}", this, ret);
286         return ret;
287     }
288
289     final void delete(final YangInstanceIdentifier path) {
290         checkReadWrite();
291         checkNotSealed();
292         doDelete(path);
293     }
294
295     final void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
296         checkReadWrite();
297         checkNotSealed();
298         doMerge(path, data);
299     }
300
301     final void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
302         checkReadWrite();
303         checkNotSealed();
304         doWrite(path, data);
305     }
306
307     final CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
308         checkNotSealed();
309         return doExists(path);
310     }
311
312     final CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
313         checkNotSealed();
314         return doRead(path);
315     }
316
317     final void enqueueRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
318             final long enqueuedTicks) {
319         LOG.debug("Transaction proxy {} enqueing request {} callback {}", this, request, callback);
320         parent.enqueueRequest(request, callback, enqueuedTicks);
321     }
322
323     final void sendRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
324         LOG.debug("Transaction proxy {} sending request {} callback {}", this, request, callback);
325         parent.sendRequest(request, callback);
326     }
327
328     /**
329      * Seal this transaction before it is either committed or aborted. This method should only be invoked from
330      * application thread.
331      */
332     final void seal() {
333         // Transition user-visible state first
334         final boolean success = markSealed();
335         Preconditions.checkState(success, "Proxy %s was already sealed", getIdentifier());
336
337         if (!sealAndSend(Optional.absent())) {
338             sealSuccessor();
339         }
340     }
341
342     /**
343      * Internal seal propagation method, invoked when we have raced with reconnection thread. Note that there may have
344      * been multiple reconnects, so we have to make sure the action is propagate through all intermediate instances.
345      */
346     private void sealSuccessor() {
347         // Slow path: wait for the successor to complete
348         final AbstractProxyTransaction successor = awaitSuccessor();
349
350         // At this point the successor has completed transition and is possibly visible by the user thread, which is
351         // still stuck here. The successor has not seen final part of our state, nor the fact it is sealed.
352         // Propagate state and seal the successor.
353         flushState(successor);
354         successor.predecessorSealed();
355     }
356
357     private void predecessorSealed() {
358         if (markSealed() && !sealAndSend(Optional.absent())) {
359             sealSuccessor();
360         }
361     }
362
363     void sealOnly() {
364         parent.onTransactionSealed(this);
365         final boolean success = STATE_UPDATER.compareAndSet(this, OPEN, SEALED);
366         Verify.verify(success, "Attempted to replay seal on %s", this);
367     }
368
369     /**
370      * Seal this transaction and potentially send it out towards the backend. If this method reports false, the caller
371      * needs to deal with propagating the seal operation towards the successor.
372      *
373      * @param enqueuedTicks Enqueue ticks when this is invoked from replay path.
374      * @return True if seal operation was successful, false if this proxy has a successor.
375      */
376     boolean sealAndSend(final Optional<Long> enqueuedTicks) {
377         parent.onTransactionSealed(this);
378
379         // Transition internal state to sealed and detect presence of a successor
380         return STATE_UPDATER.compareAndSet(this, OPEN, SEALED);
381     }
382
383     /**
384      * Mark this proxy as having been sealed.
385      *
386      * @return True if this call has transitioned to sealed state.
387      */
388     final boolean markSealed() {
389         return SEALED_UPDATER.compareAndSet(this, 0, 1);
390     }
391
392     private void checkNotSealed() {
393         Preconditions.checkState(sealed == 0, "Transaction %s has already been sealed", getIdentifier());
394     }
395
396     private void checkSealed() {
397         Preconditions.checkState(sealed != 0, "Transaction %s has not been sealed yet", getIdentifier());
398     }
399
400     private SuccessorState getSuccessorState() {
401         final State local = state;
402         Verify.verify(local instanceof SuccessorState, "State %s has unexpected class", local);
403         return (SuccessorState) local;
404     }
405
406     private void checkReadWrite() {
407         if (isSnapshotOnly()) {
408             throw new UnsupportedOperationException("Transaction " + getIdentifier() + " is a read-only snapshot");
409         }
410     }
411
412     final void recordSuccessfulRequest(final @Nonnull TransactionRequest<?> req) {
413         successfulRequests.add(Verify.verifyNotNull(req));
414     }
415
416     final void recordFinishedRequest(final Response<?, ?> response) {
417         final Object last = successfulRequests.peekLast();
418         if (last instanceof IncrementSequence) {
419             ((IncrementSequence) last).incrementDelta();
420         } else {
421             successfulRequests.addLast(new IncrementSequence(response.getSequence()));
422         }
423     }
424
425     /**
426      * Abort this transaction. This is invoked only for read-only transactions and will result in an explicit message
427      * being sent to the backend.
428      */
429     final void abort() {
430         checkNotSealed();
431         parent.abortTransaction(this);
432
433         sendRequest(abortRequest(), resp -> {
434             LOG.debug("Transaction {} abort completed with {}", getIdentifier(), resp);
435             enqueuePurge();
436         });
437     }
438
439     final void abort(final VotingFuture<Void> ret) {
440         checkSealed();
441
442         sendDoAbort(t -> {
443             if (t instanceof TransactionAbortSuccess) {
444                 ret.voteYes();
445             } else if (t instanceof RequestFailure) {
446                 ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
447             } else {
448                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
449             }
450
451             // This is a terminal request, hence we do not need to record it
452             LOG.debug("Transaction {} abort completed", this);
453             enqueuePurge();
454         });
455     }
456
457     final void enqueueAbort(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
458         checkNotSealed();
459         parent.abortTransaction(this);
460
461         enqueueRequest(abortRequest(), resp -> {
462             LOG.debug("Transaction {} abort completed with {}", getIdentifier(), resp);
463             // Purge will be sent by the predecessor's callback
464             if (callback != null) {
465                 callback.accept(resp);
466             }
467         }, enqueuedTicks);
468     }
469
470     final void enqueueDoAbort(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
471         enqueueRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), callback,
472             enqueuedTicks);
473     }
474
475     final void sendDoAbort(final Consumer<Response<?, ?>> callback) {
476         sendRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), callback);
477     }
478
479     /**
480      * Commit this transaction, possibly in a coordinated fashion.
481      *
482      * @param coordinated True if this transaction should be coordinated across multiple participants.
483      * @return Future completion
484      */
485     final ListenableFuture<Boolean> directCommit() {
486         checkReadWrite();
487         checkSealed();
488
489         // Precludes startReconnect() from interfering with the fast path
490         synchronized (this) {
491             if (STATE_UPDATER.compareAndSet(this, SEALED, FLUSHED)) {
492                 final SettableFuture<Boolean> ret = SettableFuture.create();
493                 sendRequest(Verify.verifyNotNull(commitRequest(false)), t -> {
494                     if (t instanceof TransactionCommitSuccess) {
495                         ret.set(Boolean.TRUE);
496                     } else if (t instanceof RequestFailure) {
497                         final Throwable cause = ((RequestFailure<?, ?>) t).getCause().unwrap();
498                         if (cause instanceof ClosedTransactionException) {
499                             // This is okay, as it indicates the transaction has been completed. It can happen
500                             // when we lose connectivity with the backend after it has received the request.
501                             ret.set(Boolean.TRUE);
502                         } else {
503                             ret.setException(cause);
504                         }
505                     } else {
506                         ret.setException(new IllegalStateException("Unhandled response " + t.getClass()));
507                     }
508
509                     // This is a terminal request, hence we do not need to record it
510                     LOG.debug("Transaction {} directCommit completed", this);
511                     enqueuePurge();
512                 });
513
514                 return ret;
515             }
516         }
517
518         // We have had some interference with successor injection, wait for it to complete and defer to the successor.
519         return awaitSuccessor().directCommit();
520     }
521
522     final void canCommit(final VotingFuture<?> ret) {
523         checkReadWrite();
524         checkSealed();
525
526         // Precludes startReconnect() from interfering with the fast path
527         synchronized (this) {
528             if (STATE_UPDATER.compareAndSet(this, SEALED, FLUSHED)) {
529                 final TransactionRequest<?> req = Verify.verifyNotNull(commitRequest(true));
530
531                 sendRequest(req, t -> {
532                     if (t instanceof TransactionCanCommitSuccess) {
533                         ret.voteYes();
534                     } else if (t instanceof RequestFailure) {
535                         ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
536                     } else {
537                         ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
538                     }
539
540                     recordSuccessfulRequest(req);
541                     LOG.debug("Transaction {} canCommit completed", this);
542                 });
543
544                 return;
545             }
546         }
547
548         // We have had some interference with successor injection, wait for it to complete and defer to the successor.
549         awaitSuccessor().canCommit(ret);
550     }
551
552     private AbstractProxyTransaction awaitSuccessor() {
553         return getSuccessorState().await();
554     }
555
556     final void preCommit(final VotingFuture<?> ret) {
557         checkReadWrite();
558         checkSealed();
559
560         final TransactionRequest<?> req = new TransactionPreCommitRequest(getIdentifier(), nextSequence(),
561             localActor());
562         sendRequest(req, t -> {
563             if (t instanceof TransactionPreCommitSuccess) {
564                 ret.voteYes();
565             } else if (t instanceof RequestFailure) {
566                 ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
567             } else {
568                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
569             }
570
571             onPreCommitComplete(req);
572         });
573     }
574
575     private void onPreCommitComplete(final TransactionRequest<?> req) {
576         /*
577          * The backend has agreed that the transaction has entered PRE_COMMIT phase, meaning it will be committed
578          * to storage after the timeout completes.
579          *
580          * All state has been replicated to the backend, hence we do not need to keep it around. Retain only
581          * the precommit request, so we know which request to use for resync.
582          */
583         LOG.debug("Transaction {} preCommit completed, clearing successfulRequests", this);
584         successfulRequests.clear();
585
586         // TODO: this works, but can contain some useless state (like batched operations). Create an empty
587         //       equivalent of this request and store that.
588         recordSuccessfulRequest(req);
589     }
590
591     final void doCommit(final VotingFuture<?> ret) {
592         checkReadWrite();
593         checkSealed();
594
595         sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), t -> {
596             if (t instanceof TransactionCommitSuccess) {
597                 ret.voteYes();
598             } else if (t instanceof RequestFailure) {
599                 ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
600             } else {
601                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
602             }
603
604             LOG.debug("Transaction {} doCommit completed", this);
605
606             // Needed for ProxyHistory$Local data tree rebase points.
607             parent.completeTransaction(this);
608
609             enqueuePurge();
610         });
611     }
612
613     private void enqueuePurge() {
614         enqueuePurge(null);
615     }
616
617     final void enqueuePurge(final Consumer<Response<?, ?>> callback) {
618         // Purge request are dispatched internally, hence should not wait
619         enqueuePurge(callback, parent.currentTime());
620     }
621
622     final void enqueuePurge(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
623         LOG.debug("{}: initiating purge", this);
624
625         final State prev = state;
626         if (prev instanceof SuccessorState) {
627             ((SuccessorState) prev).setDone();
628         } else {
629             final boolean success = STATE_UPDATER.compareAndSet(this, prev, DONE);
630             if (!success) {
631                 LOG.warn("{}: moved from state {} while we were purging it", this, prev);
632             }
633         }
634
635         successfulRequests.clear();
636
637         enqueueRequest(new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor()), resp -> {
638             LOG.debug("{}: purge completed", this);
639             parent.purgeTransaction(this);
640
641             if (callback != null) {
642                 callback.accept(resp);
643             }
644         }, enqueuedTicks);
645     }
646
647     // Called with the connection unlocked
648     final synchronized void startReconnect() {
649         // At this point canCommit/directCommit are blocked, we assert a new successor state, retrieving the previous
650         // state. This method is called with the queue still unlocked.
651         final SuccessorState nextState = new SuccessorState();
652         final State prevState = STATE_UPDATER.getAndSet(this, nextState);
653
654         LOG.debug("Start reconnect of proxy {} previous state {}", this, prevState);
655         Verify.verify(!(prevState instanceof SuccessorState), "Proxy %s duplicate reconnect attempt after %s", this,
656             prevState);
657
658         // We have asserted a slow-path state, seal(), canCommit(), directCommit() are forced to slow paths, which will
659         // wait until we unblock nextState's latch before accessing state. Now we record prevState for later use and we
660         // are done.
661         nextState.setPrevState(prevState);
662     }
663
664     // Called with the connection locked
665     final void replayMessages(final ProxyHistory successorHistory, final Iterable<ConnectionEntry> enqueuedEntries) {
666         final SuccessorState local = getSuccessorState();
667         final State prevState = local.getPrevState();
668
669         final AbstractProxyTransaction successor = successorHistory.createTransactionProxy(getIdentifier(),
670             isSnapshotOnly(), local.isDone());
671         LOG.debug("{} created successor {}", this, successor);
672         local.setSuccessor(successor);
673
674         // Replay successful requests first
675         if (!successfulRequests.isEmpty()) {
676             // We need to find a good timestamp to use for successful requests, as we do not want to time them out
677             // nor create timing inconsistencies in the queue -- requests are expected to be ordered by their enqueue
678             // time. We will pick the time of the first entry available. If there is none, we will just use current
679             // time, as all other requests will get enqueued afterwards.
680             final ConnectionEntry firstInQueue = Iterables.getFirst(enqueuedEntries, null);
681             final long now = firstInQueue != null ? firstInQueue.getEnqueuedTicks() : parent.currentTime();
682
683             for (Object obj : successfulRequests) {
684                 if (obj instanceof TransactionRequest) {
685                     LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
686                     successor.doReplayRequest((TransactionRequest<?>) obj, resp -> { }, now);
687                 } else {
688                     Verify.verify(obj instanceof IncrementSequence);
689                     final IncrementSequence increment = (IncrementSequence) obj;
690                     successor.doReplayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
691                         increment.getSequence(), localActor(), isSnapshotOnly(), increment.getDelta()), resp -> { },
692                         now);
693                     LOG.debug("Incrementing sequence {} to successor {}", obj, successor);
694                 }
695             }
696             LOG.debug("{} replayed {} successful requests", getIdentifier(), successfulRequests.size());
697             successfulRequests.clear();
698         }
699
700         // Now replay whatever is in the connection
701         final Iterator<ConnectionEntry> it = enqueuedEntries.iterator();
702         while (it.hasNext()) {
703             final ConnectionEntry e = it.next();
704             final Request<?, ?> req = e.getRequest();
705
706             if (getIdentifier().equals(req.getTarget())) {
707                 Verify.verify(req instanceof TransactionRequest, "Unhandled request %s", req);
708                 LOG.debug("Replaying queued request {} to successor {}", req, successor);
709                 successor.doReplayRequest((TransactionRequest<?>) req, e.getCallback(), e.getEnqueuedTicks());
710                 it.remove();
711             }
712         }
713
714         /*
715          * Check the state at which we have started the reconnect attempt. State transitions triggered while we were
716          * reconnecting have been forced to slow paths, which will be unlocked once we unblock the state latch
717          * at the end of this method.
718          */
719         if (SEALED.equals(prevState)) {
720             LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
721             flushState(successor);
722             if (successor.markSealed()) {
723                 successor.sealAndSend(Optional.of(parent.currentTime()));
724             }
725         }
726     }
727
728     /**
729      * Invoked from {@link #replayMessages(AbstractProxyTransaction, Iterable)} to have successor adopt an in-flight
730      * request.
731      *
732      * <p>
733      * Note: this method is invoked by the predecessor on the successor.
734      *
735      * @param request Request which needs to be forwarded
736      * @param callback Callback to be invoked once the request completes
737      * @param enqueuedTicks ticker-based time stamp when the request was enqueued
738      */
739     private void doReplayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
740             final long enqueuedTicks) {
741         if (request instanceof AbstractLocalTransactionRequest) {
742             handleReplayedLocalRequest((AbstractLocalTransactionRequest<?>) request, callback, enqueuedTicks);
743         } else {
744             handleReplayedRemoteRequest(request, callback, enqueuedTicks);
745         }
746     }
747
748     // Called with the connection locked
749     final void finishReconnect() {
750         final SuccessorState local = getSuccessorState();
751         LOG.debug("Finishing reconnect of proxy {}", this);
752
753         // All done, release the latch, unblocking seal() and canCommit() slow paths
754         local.finish();
755     }
756
757     /**
758      * Invoked from a retired connection for requests which have been in-flight and need to be re-adjusted
759      * and forwarded to the successor connection.
760      *
761      * @param request Request to be forwarded
762      * @param callback Original callback
763      */
764     final void forwardRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
765         forwardToSuccessor(getSuccessorState().getSuccessor(), request, callback);
766     }
767
768     final void forwardToSuccessor(final AbstractProxyTransaction successor, final TransactionRequest<?> request,
769             final Consumer<Response<?, ?>> callback) {
770         if (successor instanceof LocalProxyTransaction) {
771             forwardToLocal((LocalProxyTransaction)successor, request, callback);
772         } else if (successor instanceof RemoteProxyTransaction) {
773             forwardToRemote((RemoteProxyTransaction)successor, request, callback);
774         } else {
775             throw new IllegalStateException("Unhandled successor " + successor);
776         }
777     }
778
779     final void replayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
780             final long enqueuedTicks) {
781         getSuccessorState().getSuccessor().doReplayRequest(request, callback, enqueuedTicks);
782     }
783
784     abstract boolean isSnapshotOnly();
785
786     abstract void doDelete(YangInstanceIdentifier path);
787
788     abstract void doMerge(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
789
790     abstract void doWrite(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
791
792     abstract CheckedFuture<Boolean, ReadFailedException> doExists(YangInstanceIdentifier path);
793
794     abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(YangInstanceIdentifier path);
795
796     @GuardedBy("this")
797     abstract void flushState(AbstractProxyTransaction successor);
798
799     abstract TransactionRequest<?> abortRequest();
800
801     abstract TransactionRequest<?> commitRequest(boolean coordinated);
802
803     /**
804      * Replay a request originating in this proxy to a successor remote proxy.
805      */
806     abstract void forwardToRemote(RemoteProxyTransaction successor, TransactionRequest<?> request,
807             Consumer<Response<?, ?>> callback);
808
809     /**
810      * Replay a request originating in this proxy to a successor local proxy.
811      */
812     abstract void forwardToLocal(LocalProxyTransaction successor, TransactionRequest<?> request,
813             Consumer<Response<?, ?>> callback);
814
815     /**
816      * Invoked from {@link LocalProxyTransaction} when it replays its successful requests to its successor.
817      *
818      * <p>
819      * Note: this method is invoked by the predecessor on the successor.
820      *
821      * @param request Request which needs to be forwarded
822      * @param callback Callback to be invoked once the request completes
823      * @param enqueuedTicks Time stamp to use for enqueue time
824      */
825     abstract void handleReplayedLocalRequest(AbstractLocalTransactionRequest<?> request,
826             @Nullable Consumer<Response<?, ?>> callback, long enqueuedTicks);
827
828     /**
829      * Invoked from {@link RemoteProxyTransaction} when it replays its successful requests to its successor.
830      *
831      * <p>
832      * Note: this method is invoked by the predecessor on the successor.
833      *
834      * @param request Request which needs to be forwarded
835      * @param callback Callback to be invoked once the request completes
836      * @param enqueuedTicks Time stamp to use for enqueue time
837      */
838     abstract void handleReplayedRemoteRequest(TransactionRequest<?> request,
839             @Nullable Consumer<Response<?, ?>> callback, long enqueuedTicks);
840
841     @Override
842     public final String toString() {
843         return MoreObjects.toStringHelper(this).add("identifier", getIdentifier()).add("state", state).toString();
844     }
845 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.