X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FAbstractFrontendHistory.java;h=f6dee7ce7b32705a333fbf7f5cc23f5eb410a5ef;hp=d7e3c6533f901b443becfefdf3302ac6cb74d3cc;hb=5e7cf2452ef634dc934a3ea5a2dd95059fbab68c;hpb=b66d5a3c59525a1c7885c3d653d9657a99f4103d

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractFrontendHistory.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractFrontendHistory.java
index d7e3c6533f..f6dee7ce7b 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractFrontendHistory.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractFrontendHistory.java
@@ -7,15 +7,24 @@
  */
 package org.opendaylight.controller.cluster.datastore;
 
+import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
-import com.google.common.base.Ticker;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Range;
+import com.google.common.collect.RangeSet;
+import com.google.common.primitives.UnsignedLong;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import javax.annotation.Nullable;
 import org.opendaylight.controller.cluster.access.commands.AbstractReadTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.ClosedTransactionException;
 import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.DeadTransactionException;
+import org.opendaylight.controller.cluster.access.commands.LocalHistorySuccess;
 import org.opendaylight.controller.cluster.access.commands.OutOfOrderRequestException;
+import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
+import org.opendaylight.controller.cluster.access.commands.TransactionPurgeResponse;
 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionSuccess;
 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
@@ -38,12 +47,22 @@ abstract class AbstractFrontendHistory implements Identifiable<LocalHistoryIdent
     private static final OutOfOrderRequestException UNSEQUENCED_START = new OutOfOrderRequestException(0);
 
     private final Map<TransactionIdentifier, FrontendTransaction> transactions = new HashMap<>();
+    private final RangeSet<UnsignedLong> purgedTransactions;
     private final String persistenceId;
-    private final Ticker ticker;
+    private final ShardDataTree tree;
 
-    AbstractFrontendHistory(final String persistenceId, final Ticker ticker) {
+    /**
+     * Transactions closed by the previous leader. Boolean indicates whether the transaction was committed (true) or
+     * aborted (false). We only ever shrink these.
+     */
+    private Map<UnsignedLong, Boolean> closedTransactions;
+
+    AbstractFrontendHistory(final String persistenceId, final ShardDataTree tree,
+        final Map<UnsignedLong, Boolean> closedTransactions, final RangeSet<UnsignedLong> purgedTransactions) {
         this.persistenceId = Preconditions.checkNotNull(persistenceId);
-        this.ticker = Preconditions.checkNotNull(ticker);
+        this.tree = Preconditions.checkNotNull(tree);
+        this.closedTransactions = Preconditions.checkNotNull(closedTransactions);
+        this.purgedTransactions = Preconditions.checkNotNull(purgedTransactions);
     }
 
     final String persistenceId() {
@@ -51,15 +70,68 @@ abstract class AbstractFrontendHistory implements Identifiable<LocalHistoryIdent
     }
 
     final long readTime() {
-        return ticker.read();
+        return tree.ticker().read();
     }
 
     final @Nullable TransactionSuccess<?> handleTransactionRequest(final TransactionRequest<?> request,
             final RequestEnvelope envelope, final long now) throws RequestException {
+        final TransactionIdentifier id = request.getTarget();
+        final UnsignedLong ul = UnsignedLong.fromLongBits(id.getTransactionId());
 
-        // FIXME: handle purging of transactions
+        if (request instanceof TransactionPurgeRequest) {
+            if (purgedTransactions.contains(ul)) {
+                // Retransmitted purge request: nothing to do
+                LOG.debug("{}: transaction {} already purged", persistenceId, id);
+                return new TransactionPurgeResponse(id, request.getSequence());
+            }
+
+            // We perform two lookups instead of a straight remove, because once the map becomes empty we switch it
+            // to an ImmutableMap, which does not allow remove().
+            if (closedTransactions.containsKey(ul)) {
+                tree.purgeTransaction(id, () -> {
+                    closedTransactions.remove(ul);
+                    if (closedTransactions.isEmpty()) {
+                        closedTransactions = ImmutableMap.of();
+                    }
+
+                    purgedTransactions.add(Range.singleton(ul));
+                    LOG.debug("{}: finished purging inherited transaction {}", persistenceId(), id);
+                    envelope.sendSuccess(new TransactionPurgeResponse(id, request.getSequence()), readTime() - now);
+                });
+                return null;
+            }
+
+            final FrontendTransaction tx = transactions.get(id);
+            if (tx == null) {
+                // This should never happen because the purge callback removes the transaction and puts it into
+                // purged transactions in one go. If it does, we warn about the situation and
+                LOG.warn("{}: transaction {} not tracked in {}, but not present in active transactions", persistenceId,
+                    id, purgedTransactions);
+                purgedTransactions.add(Range.singleton(ul));
+                return new TransactionPurgeResponse(id, request.getSequence());
+            }
+
+            tree.purgeTransaction(id, () -> {
+                purgedTransactions.add(Range.singleton(ul));
+                transactions.remove(id);
+                LOG.debug("{}: finished purging transaction {}", persistenceId(), id);
+                envelope.sendSuccess(new TransactionPurgeResponse(id, request.getSequence()), readTime() - now);
+            });
+            return null;
+        }
+
+        if (purgedTransactions.contains(ul)) {
+            LOG.warn("{}: Request {} is contained purged transactions {}", persistenceId, request, purgedTransactions);
+            throw new DeadTransactionException(purgedTransactions);
+        }
+        final Boolean closed = closedTransactions.get(ul);
+        if (closed != null) {
+            final boolean successful = closed.booleanValue();
+            LOG.debug("{}: Request {} refers to a {} transaction", persistenceId, request, successful ? "successful"
+                    : "failed");
+            throw new ClosedTransactionException(successful);
+        }
 
-        final TransactionIdentifier id = request.getTarget();
         FrontendTransaction tx = transactions.get(id);
         if (tx == null) {
             // The transaction does not exist and we are about to create it, check sequence number
@@ -82,6 +154,18 @@ abstract class AbstractFrontendHistory implements Identifiable<LocalHistoryIdent
         return tx.handleRequest(request, envelope, now);
     }
 
+    void destroy(final long sequence, final RequestEnvelope envelope, final long now) {
+        LOG.debug("{}: closing history {}", persistenceId(), getIdentifier());
+        tree.closeTransactionChain(getIdentifier(),
+            () -> envelope.sendSuccess(new LocalHistorySuccess(getIdentifier(), sequence), readTime() - now));
+    }
+
+    void purge(final long sequence, final RequestEnvelope envelope, final long now) {
+        LOG.debug("{}: purging history {}", persistenceId(), getIdentifier());
+        tree.purgeTransactionChain(getIdentifier(),
+            () -> envelope.sendSuccess(new LocalHistorySuccess(getIdentifier(), sequence), readTime() - now));
+    }
+
     private FrontendTransaction createTransaction(final TransactionRequest<?> request, final TransactionIdentifier id)
             throws RequestException {
         if (request instanceof CommitLocalTransactionRequest) {
@@ -106,5 +190,14 @@ abstract class AbstractFrontendHistory implements Identifiable<LocalHistoryIdent
     abstract FrontendTransaction createReadyTransaction(TransactionIdentifier id, DataTreeModification mod)
         throws RequestException;
 
+    abstract ShardDataTreeCohort createFailedCohort(TransactionIdentifier id, DataTreeModification mod,
+            Exception failure);
+
     abstract ShardDataTreeCohort createReadyCohort(TransactionIdentifier id, DataTreeModification mod);
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this).omitNullValues().add("identifier", getIdentifier())
+                .add("persistenceId", persistenceId).add("transactions", transactions).toString();
+    }
 }