opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendTransaction.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.datastore;
   9
  10 import com.google.common.base.Preconditions;
  11 import com.google.common.base.Verify;
  12 import java.util.ArrayDeque;
  13 import java.util.Iterator;
  14 import java.util.Queue;
  15 import javax.annotation.Nullable;
  16 import javax.annotation.concurrent.NotThreadSafe;
  17 import org.opendaylight.controller.cluster.access.commands.OutOfOrderRequestException;
  18 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
  19 import org.opendaylight.controller.cluster.access.commands.TransactionSuccess;
  20 import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope;
  21 import org.opendaylight.controller.cluster.access.concepts.RequestException;
  22 import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException;
  23 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  24 import org.opendaylight.yangtools.concepts.Identifiable;
  25
  26 /**
  27  * Frontend common transaction state as observed by the shard leader.
  28  *
  29  * @author Robert Varga
  30  */
  31 @NotThreadSafe
  32 abstract class FrontendTransaction implements Identifiable<TransactionIdentifier> {
  33     private final AbstractFrontendHistory history;
  34     private final TransactionIdentifier id;
  35
  36     /**
  37      * It is possible that after we process a request and send a response that response gets lost and the client
  38      * initiates a retry. Since subsequent requests can mutate transaction state we need to retain the response until
  39      * it is acknowledged by the client.
  40      */
  41     private final Queue<Object> replayQueue = new ArrayDeque<>();
  42     private long firstReplaySequence;
  43     private Long lastPurgedSequence;
  44     private long expectedSequence;
  45
  46     FrontendTransaction(final AbstractFrontendHistory history, final TransactionIdentifier id) {
  47         this.history = Preconditions.checkNotNull(history);
  48         this.id = Preconditions.checkNotNull(id);
  49     }
  50
  51     @Override
  52     public final TransactionIdentifier getIdentifier() {
  53         return id;
  54     }
  55
  56     final AbstractFrontendHistory history() {
  57         return history;
  58     }
  59
  60     final java.util.Optional<TransactionSuccess<?>> replaySequence(final long sequence) throws RequestException {
  61         // Fast path check: if the requested sequence is the next request, bail early
  62         if (expectedSequence == sequence) {
  63             return java.util.Optional.empty();
  64         }
  65
  66         // Check sequencing: we do not need to bother with future requests
  67         if (Long.compareUnsigned(expectedSequence, sequence) < 0) {
  68             throw new OutOfOrderRequestException(expectedSequence);
  69         }
  70
  71         // Sanity check: if we have purged sequences, this has to be newer
  72         if (lastPurgedSequence != null && Long.compareUnsigned(lastPurgedSequence.longValue(), sequence) >= 0) {
  73             // Client has sent a request sequence, which has already been purged. This is a hard error, which should
  74             // never occur. Throwing an IllegalArgumentException will cause it to be wrapped in a
  75             // RuntimeRequestException (which is not retriable) and report it back to the client.
  76             throw new IllegalArgumentException(String.format("Invalid purged sequence %s (last purged is %s)",
  77                 sequence, lastPurgedSequence));
  78         }
  79
  80         // At this point we have established that the requested sequence lies in the open interval
  81         // (lastPurgedSequence, expectedSequence). That does not actually mean we have a response, as the commit
  82         // machinery is asynchronous, hence a reply may be in the works and not available.
  83
  84         long replaySequence = firstReplaySequence;
  85         final Iterator<?> it = replayQueue.iterator();
  86         while (it.hasNext()) {
  87             final Object replay = it.next();
  88             if (replaySequence == sequence) {
  89                 if (replay instanceof RequestException) {
  90                     throw (RequestException) replay;
  91                 }
  92
  93                 Verify.verify(replay instanceof TransactionSuccess);
  94                 return java.util.Optional.of((TransactionSuccess<?>) replay);
  95             }
  96
  97             replaySequence++;
  98         }
  99
 100         // Not found
 101         return java.util.Optional.empty();
 102     }
 103
 104     final void purgeSequencesUpTo(final long sequence) {
 105         // FIXME: implement this
 106
 107         lastPurgedSequence = sequence;
 108     }
 109
 110     // Sequence has already been checked
 111     abstract @Nullable TransactionSuccess<?> handleRequest(final TransactionRequest<?> request,
 112             final RequestEnvelope envelope, final long now) throws RequestException;
 113
 114     private void recordResponse(final long sequence, final Object response) {
 115         if (replayQueue.isEmpty()) {
 116             firstReplaySequence = sequence;
 117         }
 118         replayQueue.add(response);
 119         expectedSequence++;
 120     }
 121
 122     final <T extends TransactionSuccess<?>> T recordSuccess(final long sequence, final T success) {
 123         recordResponse(sequence, success);
 124         return success;
 125     }
 126
 127     private long executionTime(final long startTime) {
 128         return history.readTime() - startTime;
 129     }
 130
 131     final void recordAndSendSuccess(final RequestEnvelope envelope, final long startTime,
 132             final TransactionSuccess<?> success) {
 133         recordResponse(success.getSequence(), success);
 134         envelope.sendSuccess(success, executionTime(startTime));
 135     }
 136
 137     final void recordAndSendFailure(final RequestEnvelope envelope, final long startTime,
 138             final RuntimeRequestException failure) {
 139         recordResponse(envelope.getMessage().getSequence(), failure);
 140         envelope.sendFailure(failure, executionTime(startTime));
 141     }
 142 }