opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardDataTree.java

   1 /*
   2  * Copyright (c) 2015 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 akka.actor.ActorRef;
  11 import akka.util.Timeout;
  12 import com.google.common.annotations.VisibleForTesting;
  13 import com.google.common.base.Optional;
  14 import com.google.common.base.Preconditions;
  15 import com.google.common.base.Stopwatch;
  16 import com.google.common.base.Verify;
  17 import com.google.common.collect.ImmutableList;
  18 import com.google.common.collect.ImmutableMap;
  19 import com.google.common.collect.ImmutableMap.Builder;
  20 import com.google.common.primitives.UnsignedLong;
  21 import java.io.IOException;
  22 import java.util.AbstractMap.SimpleEntry;
  23 import java.util.ArrayDeque;
  24 import java.util.ArrayList;
  25 import java.util.Collection;
  26 import java.util.HashMap;
  27 import java.util.Iterator;
  28 import java.util.Map;
  29 import java.util.Map.Entry;
  30 import java.util.Queue;
  31 import java.util.concurrent.ExecutionException;
  32 import java.util.concurrent.TimeUnit;
  33 import java.util.concurrent.TimeoutException;
  34 import java.util.function.UnaryOperator;
  35 import javax.annotation.Nonnull;
  36 import javax.annotation.concurrent.NotThreadSafe;
  37 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
  38 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  39 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActorRegistry.CohortRegistryCommand;
  40 import org.opendaylight.controller.cluster.datastore.ShardDataTreeCohort.State;
  41 import org.opendaylight.controller.cluster.datastore.persisted.CommitTransactionPayload;
  42 import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
  43 import org.opendaylight.controller.cluster.datastore.persisted.ShardDataTreeSnapshot;
  44 import org.opendaylight.controller.cluster.datastore.persisted.ShardDataTreeSnapshotMetadata;
  45 import org.opendaylight.controller.cluster.datastore.utils.PruningDataTreeModification;
  46 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  47 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  48 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
  49 import org.opendaylight.controller.md.sal.common.api.data.OptimisticLockFailedException;
  50 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  51 import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
  52 import org.opendaylight.controller.md.sal.dom.store.impl.DataChangeListenerRegistration;
  53 import org.opendaylight.yangtools.concepts.Identifier;
  54 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  55 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  56 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  57 import org.opendaylight.yangtools.yang.data.api.schema.tree.ConflictingModificationAppliedException;
  58 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  59 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateTip;
  60 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
  61 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  62 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
  63 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
  64 import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
  65 import org.opendaylight.yangtools.yang.data.api.schema.tree.TipProducingDataTree;
  66 import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
  67 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
  68 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  69 import org.slf4j.Logger;
  70 import org.slf4j.LoggerFactory;
  71 import scala.concurrent.duration.Duration;
  72
  73 /**
  74  * Internal shard state, similar to a DOMStore, but optimized for use in the actor system,
  75  * e.g. it does not expose public interfaces and assumes it is only ever called from a
  76  * single thread.
  77  *
  78  * This class is not part of the API contract and is subject to change at any time.
  79  */
  80 @NotThreadSafe
  81 public class ShardDataTree extends ShardDataTreeTransactionParent {
  82     private static final class CommitEntry {
  83         final SimpleShardDataTreeCohort cohort;
  84         long lastAccess;
  85
  86         CommitEntry(final SimpleShardDataTreeCohort cohort, final long now) {
  87             this.cohort = Preconditions.checkNotNull(cohort);
  88             lastAccess = now;
  89         }
  90     }
  91
  92     private static final Timeout COMMIT_STEP_TIMEOUT = new Timeout(Duration.create(5, TimeUnit.SECONDS));
  93     private static final Logger LOG = LoggerFactory.getLogger(ShardDataTree.class);
  94
  95     private final Map<LocalHistoryIdentifier, ShardDataTreeTransactionChain> transactionChains = new HashMap<>();
  96     private final DataTreeCohortActorRegistry cohortRegistry = new DataTreeCohortActorRegistry();
  97     private final Queue<CommitEntry> pendingTransactions = new ArrayDeque<>();
  98     private final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher;
  99     private final ShardDataChangeListenerPublisher dataChangeListenerPublisher;
 100     private final Collection<ShardDataTreeMetadata<?>> metadata;
 101     private final TipProducingDataTree dataTree;
 102     private final String logContext;
 103     private final Shard shard;
 104     private Runnable runOnPendingTransactionsComplete;
 105
 106     private SchemaContext schemaContext;
 107
 108     public ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TipProducingDataTree dataTree,
 109             final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher,
 110             final ShardDataChangeListenerPublisher dataChangeListenerPublisher, final String logContext,
 111             final ShardDataTreeMetadata<?>... metadata) {
 112         this.dataTree = Preconditions.checkNotNull(dataTree);
 113         updateSchemaContext(schemaContext);
 114
 115         this.shard = Preconditions.checkNotNull(shard);
 116         this.treeChangeListenerPublisher = Preconditions.checkNotNull(treeChangeListenerPublisher);
 117         this.dataChangeListenerPublisher = Preconditions.checkNotNull(dataChangeListenerPublisher);
 118         this.logContext = Preconditions.checkNotNull(logContext);
 119         this.metadata = ImmutableList.copyOf(metadata);
 120     }
 121
 122     public ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TreeType treeType,
 123             final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher,
 124             final ShardDataChangeListenerPublisher dataChangeListenerPublisher, final String logContext) {
 125         this(shard, schemaContext, InMemoryDataTreeFactory.getInstance().create(treeType),
 126                 treeChangeListenerPublisher, dataChangeListenerPublisher, logContext);
 127     }
 128
 129     @VisibleForTesting
 130     public ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TreeType treeType) {
 131         this(shard, schemaContext, treeType, new DefaultShardDataTreeChangeListenerPublisher(),
 132                 new DefaultShardDataChangeListenerPublisher(), "");
 133     }
 134
 135     String logContext() {
 136         return logContext;
 137     }
 138
 139     public TipProducingDataTree getDataTree() {
 140         return dataTree;
 141     }
 142
 143     SchemaContext getSchemaContext() {
 144         return schemaContext;
 145     }
 146
 147     void updateSchemaContext(final SchemaContext schemaContext) {
 148         dataTree.setSchemaContext(schemaContext);
 149         this.schemaContext = Preconditions.checkNotNull(schemaContext);
 150     }
 151
 152     /**
 153      * Take a snapshot of current state for later recovery.
 154      *
 155      * @return A state snapshot
 156      */
 157     @Nonnull ShardDataTreeSnapshot takeStateSnapshot() {
 158         final NormalizedNode<?, ?> rootNode = dataTree.takeSnapshot().readNode(YangInstanceIdentifier.EMPTY).get();
 159         final Builder<Class<? extends ShardDataTreeSnapshotMetadata<?>>, ShardDataTreeSnapshotMetadata<?>> metaBuilder =
 160                 ImmutableMap.builder();
 161
 162         for (ShardDataTreeMetadata<?> m : metadata) {
 163             final ShardDataTreeSnapshotMetadata<?> meta = m.toStapshot();
 164             if (meta != null) {
 165                 metaBuilder.put(meta.getType(), meta);
 166             }
 167         }
 168
 169         return new MetadataShardDataTreeSnapshot(rootNode, metaBuilder.build());
 170     }
 171
 172     private void applySnapshot(final @Nonnull ShardDataTreeSnapshot snapshot,
 173             final UnaryOperator<DataTreeModification> wrapper) throws DataValidationFailedException {
 174         final Stopwatch elapsed = Stopwatch.createStarted();
 175
 176         if (!pendingTransactions.isEmpty()) {
 177             LOG.warn("{}: applying state snapshot with pending transactions", logContext);
 178         }
 179
 180         final Map<Class<? extends ShardDataTreeSnapshotMetadata<?>>, ShardDataTreeSnapshotMetadata<?>> snapshotMeta;
 181         if (snapshot instanceof MetadataShardDataTreeSnapshot) {
 182             snapshotMeta = ((MetadataShardDataTreeSnapshot) snapshot).getMetadata();
 183         } else {
 184             snapshotMeta = ImmutableMap.of();
 185         }
 186
 187         for (ShardDataTreeMetadata<?> m : metadata) {
 188             final ShardDataTreeSnapshotMetadata<?> s = snapshotMeta.get(m.getSupportedType());
 189             if (s != null) {
 190                 m.applySnapshot(s);
 191             } else {
 192                 m.reset();
 193             }
 194         }
 195
 196         final DataTreeModification mod = wrapper.apply(dataTree.takeSnapshot().newModification());
 197         // delete everything first
 198         mod.delete(YangInstanceIdentifier.EMPTY);
 199
 200         final java.util.Optional<NormalizedNode<?, ?>> maybeNode = snapshot.getRootNode();
 201         if (maybeNode.isPresent()) {
 202             // Add everything from the remote node back
 203             mod.write(YangInstanceIdentifier.EMPTY, maybeNode.get());
 204         }
 205         mod.ready();
 206
 207         final DataTreeModification unwrapped = unwrap(mod);
 208         dataTree.validate(unwrapped);
 209         dataTree.commit(dataTree.prepare(unwrapped));
 210         LOG.debug("{}: state snapshot applied in %s", logContext, elapsed);
 211     }
 212
 213     private PruningDataTreeModification wrapWithPruning(final DataTreeModification delegate) {
 214         return new PruningDataTreeModification(delegate, dataTree, schemaContext);
 215     }
 216
 217     private static DataTreeModification unwrap(final DataTreeModification modification) {
 218         if (modification instanceof PruningDataTreeModification) {
 219             return ((PruningDataTreeModification)modification).delegate();
 220         }
 221         return modification;
 222     }
 223
 224     /**
 225      * Apply a snapshot coming from recovery. This method does not assume the SchemaContexts match and performs data
 226      * pruning in an attempt to adjust the state to our current SchemaContext.
 227      *
 228      * @param snapshot Snapshot that needs to be applied
 229      * @throws DataValidationFailedException when the snapshot fails to apply
 230      */
 231     void applyRecoverySnapshot(final @Nonnull ShardDataTreeSnapshot snapshot) throws DataValidationFailedException {
 232         applySnapshot(snapshot, this::wrapWithPruning);
 233     }
 234
 235
 236     /**
 237      * Apply a snapshot coming from the leader. This method assumes the leader and follower SchemaContexts match and
 238      * does not perform any pruning.
 239      *
 240      * @param snapshot Snapshot that needs to be applied
 241      * @throws DataValidationFailedException when the snapshot fails to apply
 242      */
 243     void applySnapshot(final @Nonnull ShardDataTreeSnapshot snapshot) throws DataValidationFailedException {
 244         applySnapshot(snapshot, UnaryOperator.identity());
 245     }
 246
 247     private void applyRecoveryCandidate(final DataTreeCandidate candidate) throws DataValidationFailedException {
 248         final PruningDataTreeModification mod = wrapWithPruning(dataTree.takeSnapshot().newModification());
 249         DataTreeCandidates.applyToModification(mod, candidate);
 250         mod.ready();
 251
 252         final DataTreeModification unwrapped = mod.delegate();
 253         LOG.trace("{}: Applying recovery modification {}", logContext, unwrapped);
 254
 255         dataTree.validate(unwrapped);
 256         dataTree.commit(dataTree.prepare(unwrapped));
 257     }
 258
 259     /**
 260      * Apply a payload coming from recovery. This method does not assume the SchemaContexts match and performs data
 261      * pruning in an attempt to adjust the state to our current SchemaContext.
 262      *
 263      * @param payload Payload
 264      * @throws IOException when the snapshot fails to deserialize
 265      * @throws DataValidationFailedException when the snapshot fails to apply
 266      */
 267     void applyRecoveryPayload(final @Nonnull Payload payload) throws IOException, DataValidationFailedException {
 268         if (payload instanceof CommitTransactionPayload) {
 269             final Entry<TransactionIdentifier, DataTreeCandidate> e = ((CommitTransactionPayload) payload).getCandidate();
 270             applyRecoveryCandidate(e.getValue());
 271             allMetadataCommittedTransaction(e.getKey());
 272         } else if (payload instanceof DataTreeCandidatePayload) {
 273             applyRecoveryCandidate(((DataTreeCandidatePayload) payload).getCandidate());
 274         } else {
 275             LOG.warn("{}: ignoring unhandled payload {}", logContext, payload);
 276         }
 277     }
 278
 279     private void applyReplicatedCandidate(final Identifier identifier, final DataTreeCandidate foreign)
 280             throws DataValidationFailedException {
 281         LOG.debug("{}: Applying foreign transaction {}", logContext, identifier);
 282
 283         final DataTreeModification mod = dataTree.takeSnapshot().newModification();
 284         DataTreeCandidates.applyToModification(mod, foreign);
 285         mod.ready();
 286
 287         LOG.trace("{}: Applying foreign modification {}", logContext, mod);
 288         dataTree.validate(mod);
 289         final DataTreeCandidate candidate = dataTree.prepare(mod);
 290         dataTree.commit(candidate);
 291
 292         notifyListeners(candidate);
 293     }
 294
 295     /**
 296      * Apply a payload coming from the leader, which could actually be us. This method assumes the leader and follower
 297      * SchemaContexts match and does not perform any pruning.
 298      *
 299      * @param identifier Payload identifier as returned from RaftActor
 300      * @param payload Payload
 301      * @throws IOException when the snapshot fails to deserialize
 302      * @throws DataValidationFailedException when the snapshot fails to apply
 303      */
 304     void applyReplicatedPayload(final Identifier identifier, final Payload payload) throws IOException,
 305             DataValidationFailedException {
 306         /*
 307          * This is a bit more involved than it needs to be due to to the fact we do not want to be touching the payload
 308          * if we are the leader and it has originated with us.
 309          *
 310          * The identifier will only ever be non-null when we were the leader which achieved consensus. Unfortunately,
 311          * though, this may not be the case anymore, as we are being called some time afterwards and we may not be
 312          * acting in that capacity anymore.
 313          *
 314          * In any case, we know that this is an entry coming from replication, hence we can be sure we will not observe
 315          * pre-Boron state -- which limits the number of options here.
 316          */
 317         if (payload instanceof CommitTransactionPayload) {
 318             if (identifier == null) {
 319                 final Entry<TransactionIdentifier, DataTreeCandidate> e = ((CommitTransactionPayload) payload).getCandidate();
 320                 applyReplicatedCandidate(e.getKey(), e.getValue());
 321                 allMetadataCommittedTransaction(e.getKey());
 322             } else {
 323                 Verify.verify(identifier instanceof TransactionIdentifier);
 324                 payloadReplicationComplete((TransactionIdentifier) identifier);
 325             }
 326         } else {
 327             LOG.warn("{}: ignoring unhandled identifier {} payload {}", logContext, identifier, payload);
 328         }
 329     }
 330
 331     private void payloadReplicationComplete(final TransactionIdentifier txId) {
 332         final CommitEntry current = pendingTransactions.peek();
 333         if (current == null) {
 334             LOG.warn("{}: No outstanding transactions, ignoring consensus on transaction {}", logContext, txId);
 335             return;
 336         }
 337
 338         if (!current.cohort.getIdentifier().equals(txId)) {
 339             LOG.warn("{}: Head of queue is {}, ignoring consensus on transaction {}", logContext,
 340                 current.cohort.getIdentifier(), txId);
 341             return;
 342         }
 343
 344         finishCommit(current.cohort);
 345     }
 346
 347     private void allMetadataCommittedTransaction(final TransactionIdentifier txId) {
 348         for (ShardDataTreeMetadata<?> m : metadata) {
 349             m.transactionCommitted(txId);
 350         }
 351     }
 352
 353     private ShardDataTreeTransactionChain ensureTransactionChain(final LocalHistoryIdentifier localHistoryIdentifier) {
 354         ShardDataTreeTransactionChain chain = transactionChains.get(localHistoryIdentifier);
 355         if (chain == null) {
 356             chain = new ShardDataTreeTransactionChain(localHistoryIdentifier, this);
 357             transactionChains.put(localHistoryIdentifier, chain);
 358         }
 359
 360         return chain;
 361     }
 362
 363     ReadOnlyShardDataTreeTransaction newReadOnlyTransaction(final TransactionIdentifier txId) {
 364         if (txId.getHistoryId().getHistoryId() == 0) {
 365             return new ReadOnlyShardDataTreeTransaction(txId, dataTree.takeSnapshot());
 366         }
 367
 368         return ensureTransactionChain(txId.getHistoryId()).newReadOnlyTransaction(txId);
 369     }
 370
 371     ReadWriteShardDataTreeTransaction newReadWriteTransaction(final TransactionIdentifier txId) {
 372         if (txId.getHistoryId().getHistoryId() == 0) {
 373             return new ReadWriteShardDataTreeTransaction(ShardDataTree.this, txId, dataTree.takeSnapshot()
 374                     .newModification());
 375         }
 376
 377         return ensureTransactionChain(txId.getHistoryId()).newReadWriteTransaction(txId);
 378     }
 379
 380     public void notifyListeners(final DataTreeCandidate candidate) {
 381         treeChangeListenerPublisher.publishChanges(candidate, logContext);
 382         dataChangeListenerPublisher.publishChanges(candidate, logContext);
 383     }
 384
 385     void notifyOfInitialData(final DataChangeListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier,
 386             NormalizedNode<?, ?>>> listenerReg, final Optional<DataTreeCandidate> currentState) {
 387         if (currentState.isPresent()) {
 388             ShardDataChangeListenerPublisher localPublisher = dataChangeListenerPublisher.newInstance();
 389             localPublisher.registerDataChangeListener(listenerReg.getPath(), listenerReg.getInstance(),
 390                     listenerReg.getScope());
 391             localPublisher.publishChanges(currentState.get(), logContext);
 392         }
 393     }
 394
 395     void notifyOfInitialData(final YangInstanceIdentifier path, final DOMDataTreeChangeListener listener,
 396             final Optional<DataTreeCandidate> currentState) {
 397         if (currentState.isPresent()) {
 398             ShardDataTreeChangeListenerPublisher localPublisher = treeChangeListenerPublisher.newInstance();
 399             localPublisher.registerTreeChangeListener(path, listener);
 400             localPublisher.publishChanges(currentState.get(), logContext);
 401         }
 402     }
 403
 404     void closeAllTransactionChains() {
 405         for (ShardDataTreeTransactionChain chain : transactionChains.values()) {
 406             chain.close();
 407         }
 408
 409         transactionChains.clear();
 410     }
 411
 412     void closeTransactionChain(final LocalHistoryIdentifier transactionChainId) {
 413         final ShardDataTreeTransactionChain chain = transactionChains.remove(transactionChainId);
 414         if (chain != null) {
 415             chain.close();
 416         } else {
 417             LOG.debug("{}: Closing non-existent transaction chain {}", logContext, transactionChainId);
 418         }
 419     }
 420
 421     Entry<DataChangeListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>,
 422             Optional<DataTreeCandidate>> registerChangeListener(final YangInstanceIdentifier path,
 423                     final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener,
 424                     final DataChangeScope scope) {
 425         final DataChangeListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>> reg =
 426                 dataChangeListenerPublisher.registerDataChangeListener(path, listener, scope);
 427
 428         return new SimpleEntry<>(reg, readCurrentData());
 429     }
 430
 431     private Optional<DataTreeCandidate> readCurrentData() {
 432         final Optional<NormalizedNode<?, ?>> currentState = dataTree.takeSnapshot().readNode(YangInstanceIdentifier.EMPTY);
 433         return currentState.isPresent() ? Optional.of(DataTreeCandidates.fromNormalizedNode(
 434             YangInstanceIdentifier.EMPTY, currentState.get())) : Optional.<DataTreeCandidate>absent();
 435     }
 436
 437     public Entry<ListenerRegistration<DOMDataTreeChangeListener>, Optional<DataTreeCandidate>> registerTreeChangeListener(
 438             final YangInstanceIdentifier path, final DOMDataTreeChangeListener listener) {
 439         final ListenerRegistration<DOMDataTreeChangeListener> reg = treeChangeListenerPublisher.registerTreeChangeListener(
 440                 path, listener);
 441
 442         return new SimpleEntry<>(reg, readCurrentData());
 443     }
 444
 445     int getQueueSize() {
 446         return pendingTransactions.size();
 447     }
 448
 449     @Override
 450     void abortTransaction(final AbstractShardDataTreeTransaction<?> transaction) {
 451         // Intentional no-op
 452     }
 453
 454     @Override
 455     ShardDataTreeCohort finishTransaction(final ReadWriteShardDataTreeTransaction transaction) {
 456         final DataTreeModification snapshot = transaction.getSnapshot();
 457         snapshot.ready();
 458
 459         return createReadyCohort(transaction.getId(), snapshot);
 460     }
 461
 462     public Optional<NormalizedNode<?, ?>> readNode(final YangInstanceIdentifier path) {
 463         return dataTree.takeSnapshot().readNode(path);
 464     }
 465
 466     public DataTreeSnapshot takeSnapshot() {
 467         return dataTree.takeSnapshot();
 468     }
 469
 470     public DataTreeModification newModification() {
 471         return dataTree.takeSnapshot().newModification();
 472     }
 473
 474     /**
 475      * @deprecated This method violates DataTree containment and will be removed.
 476      */
 477     @VisibleForTesting
 478     @Deprecated
 479     public DataTreeCandidate commit(final DataTreeModification modification) throws DataValidationFailedException {
 480         modification.ready();
 481         dataTree.validate(modification);
 482         DataTreeCandidate candidate = dataTree.prepare(modification);
 483         dataTree.commit(candidate);
 484         return candidate;
 485     }
 486
 487     public Collection<ShardDataTreeCohort> getAndClearPendingTransactions() {
 488         Collection<ShardDataTreeCohort> ret = new ArrayList<>(pendingTransactions.size());
 489         for(CommitEntry entry: pendingTransactions) {
 490             ret.add(entry.cohort);
 491         }
 492
 493         pendingTransactions.clear();
 494         return ret;
 495     }
 496
 497     private void processNextTransaction() {
 498         while (!pendingTransactions.isEmpty()) {
 499             final CommitEntry entry = pendingTransactions.peek();
 500             final SimpleShardDataTreeCohort cohort = entry.cohort;
 501             final DataTreeModification modification = cohort.getDataTreeModification();
 502
 503             if(cohort.getState() != State.CAN_COMMIT_PENDING) {
 504                 break;
 505             }
 506
 507             LOG.debug("{}: Validating transaction {}", logContext, cohort.getIdentifier());
 508             Exception cause;
 509             try {
 510                 dataTree.validate(modification);
 511                 LOG.debug("{}: Transaction {} validated", logContext, cohort.getIdentifier());
 512                 cohort.successfulCanCommit();
 513                 entry.lastAccess = shard.ticker().read();
 514                 return;
 515             } catch (ConflictingModificationAppliedException e) {
 516                 LOG.warn("{}: Store Tx {}: Conflicting modification for path {}.", logContext, cohort.getIdentifier(),
 517                     e.getPath());
 518                 cause = new OptimisticLockFailedException("Optimistic lock failed.", e);
 519             } catch (DataValidationFailedException e) {
 520                 LOG.warn("{}: Store Tx {}: Data validation failed for path {}.", logContext, cohort.getIdentifier(),
 521                     e.getPath(), e);
 522
 523                 // For debugging purposes, allow dumping of the modification. Coupled with the above
 524                 // precondition log, it should allow us to understand what went on.
 525                 LOG.debug("{}: Store Tx {}: modifications: {} tree: {}", cohort.getIdentifier(), modification, dataTree);
 526                 cause = new TransactionCommitFailedException("Data did not pass validation.", e);
 527             } catch (Exception e) {
 528                 LOG.warn("{}: Unexpected failure in validation phase", logContext, e);
 529                 cause = e;
 530             }
 531
 532             // Failure path: propagate the failure, remove the transaction from the queue and loop to the next one
 533             pendingTransactions.poll().cohort.failedCanCommit(cause);
 534         }
 535
 536         maybeRunOperationOnPendingTransactionsComplete();
 537     }
 538
 539     void startCanCommit(final SimpleShardDataTreeCohort cohort) {
 540         final SimpleShardDataTreeCohort current = pendingTransactions.peek().cohort;
 541         if (!cohort.equals(current)) {
 542             LOG.debug("{}: Transaction {} scheduled for canCommit step", logContext, cohort.getIdentifier());
 543             return;
 544         }
 545
 546         processNextTransaction();
 547     }
 548
 549     private void failPreCommit(final Exception cause) {
 550         shard.getShardMBean().incrementFailedTransactionsCount();
 551         pendingTransactions.poll().cohort.failedPreCommit(cause);
 552         processNextTransaction();
 553     }
 554
 555     void startPreCommit(final SimpleShardDataTreeCohort cohort) {
 556         final CommitEntry entry = pendingTransactions.peek();
 557         Preconditions.checkState(entry != null, "Attempted to pre-commit of %s when no transactions pending", cohort);
 558
 559         final SimpleShardDataTreeCohort current = entry.cohort;
 560         Verify.verify(cohort.equals(current), "Attempted to pre-commit %s while %s is pending", cohort, current);
 561         final DataTreeCandidateTip candidate;
 562         try {
 563             candidate = dataTree.prepare(cohort.getDataTreeModification());
 564         } catch (Exception e) {
 565             failPreCommit(e);
 566             return;
 567         }
 568
 569         try {
 570             cohort.userPreCommit(candidate);
 571         } catch (ExecutionException | TimeoutException e) {
 572             failPreCommit(e);
 573             return;
 574         }
 575
 576         entry.lastAccess = shard.ticker().read();
 577         cohort.successfulPreCommit(candidate);
 578     }
 579
 580     private void failCommit(final Exception cause) {
 581         shard.getShardMBean().incrementFailedTransactionsCount();
 582         pendingTransactions.poll().cohort.failedCommit(cause);
 583         processNextTransaction();
 584     }
 585
 586     private void finishCommit(final SimpleShardDataTreeCohort cohort) {
 587         final TransactionIdentifier txId = cohort.getIdentifier();
 588         final DataTreeCandidate candidate = cohort.getCandidate();
 589
 590         LOG.debug("{}: Resuming commit of transaction {}", logContext, txId);
 591
 592         try {
 593             dataTree.commit(candidate);
 594         } catch (Exception e) {
 595             LOG.error("{}: Failed to commit transaction {}", logContext, txId, e);
 596             failCommit(e);
 597             return;
 598         }
 599
 600         shard.getShardMBean().incrementCommittedTransactionCount();
 601         shard.getShardMBean().setLastCommittedTransactionTime(System.currentTimeMillis());
 602
 603         // FIXME: propagate journal index
 604         pendingTransactions.poll().cohort.successfulCommit(UnsignedLong.ZERO);
 605
 606         LOG.trace("{}: Transaction {} committed, proceeding to notify", logContext, txId);
 607         notifyListeners(candidate);
 608
 609         processNextTransaction();
 610     }
 611
 612     void startCommit(final SimpleShardDataTreeCohort cohort, final DataTreeCandidate candidate) {
 613         final CommitEntry entry = pendingTransactions.peek();
 614         Preconditions.checkState(entry != null, "Attempted to start commit of %s when no transactions pending", cohort);
 615
 616         final SimpleShardDataTreeCohort current = entry.cohort;
 617         Verify.verify(cohort.equals(current), "Attempted to commit %s while %s is pending", cohort, current);
 618
 619         if (shard.canSkipPayload() || candidate.getRootNode().getModificationType() == ModificationType.UNMODIFIED) {
 620             LOG.debug("{}: No replication required, proceeding to finish commit", logContext);
 621             finishCommit(cohort);
 622             return;
 623         }
 624
 625         final TransactionIdentifier txId = cohort.getIdentifier();
 626         final Payload payload;
 627         try {
 628             payload = CommitTransactionPayload.create(txId, candidate);
 629         } catch (IOException e) {
 630             LOG.error("{}: Failed to encode transaction {} candidate {}", logContext, txId, candidate, e);
 631             pendingTransactions.poll().cohort.failedCommit(e);
 632             return;
 633         }
 634
 635         // Once completed, we will continue via payloadReplicationComplete
 636         entry.lastAccess = shard.ticker().read();
 637         shard.persistPayload(txId, payload);
 638         LOG.debug("{}: Transaction {} submitted to persistence", logContext, txId);
 639     }
 640
 641     void processCohortRegistryCommand(final ActorRef sender, final CohortRegistryCommand message) {
 642         cohortRegistry.process(sender, message);
 643     }
 644
 645     ShardDataTreeCohort createReadyCohort(final TransactionIdentifier txId,
 646             final DataTreeModification modification) {
 647         SimpleShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(this, modification, txId,
 648                 cohortRegistry.createCohort(schemaContext, txId, COMMIT_STEP_TIMEOUT));
 649         pendingTransactions.add(new CommitEntry(cohort, shard.ticker().read()));
 650         return cohort;
 651     }
 652
 653     void checkForExpiredTransactions(final long transactionCommitTimeoutMillis) {
 654         final long timeout = TimeUnit.MILLISECONDS.toNanos(transactionCommitTimeoutMillis);
 655         final long now = shard.ticker().read();
 656         final CommitEntry currentTx = pendingTransactions.peek();
 657         if (currentTx != null && currentTx.lastAccess + timeout < now) {
 658             LOG.warn("{}: Current transaction {} has timed out after {} ms in state {}", logContext,
 659                     currentTx.cohort.getIdentifier(), transactionCommitTimeoutMillis, currentTx.cohort.getState());
 660             boolean processNext = true;
 661             switch (currentTx.cohort.getState()) {
 662                 case CAN_COMMIT_PENDING:
 663                     pendingTransactions.poll().cohort.failedCanCommit(new TimeoutException());
 664                     break;
 665                 case CAN_COMMIT_COMPLETE:
 666                     pendingTransactions.poll().cohort.reportFailure(new TimeoutException());
 667                     break;
 668                 case PRE_COMMIT_PENDING:
 669                     pendingTransactions.poll().cohort.failedPreCommit(new TimeoutException());
 670                     break;
 671                 case PRE_COMMIT_COMPLETE:
 672                     // FIXME: this is a legacy behavior problem. Three-phase commit protocol specifies that after we
 673                     //        are ready we should commit the transaction, not abort it. Our current software stack does
 674                     //        not allow us to do that consistently, because we persist at the time of commit, hence
 675                     //        we can end up in a state where we have pre-committed a transaction, then a leader failover
 676                     //        occurred ... the new leader does not see the pre-committed transaction and does not have
 677                     //        a running timer. To fix this we really need two persistence events.
 678                     //
 679                     //        The first one, done at pre-commit time will hold the transaction payload. When consensus
 680                     //        is reached, we exit the pre-commit phase and start the pre-commit timer. Followers do not
 681                     //        apply the state in this event.
 682                     //
 683                     //        The second one, done at commit (or abort) time holds only the transaction identifier and
 684                     //        signals to followers that the state should (or should not) be applied.
 685                     //
 686                     //        In order to make the pre-commit timer working across failovers, though, we need
 687                     //        a per-shard cluster-wide monotonic time, so a follower becoming the leader can accurately
 688                     //        restart the timer.
 689                     pendingTransactions.poll().cohort.reportFailure(new TimeoutException());
 690                     break;
 691                 case COMMIT_PENDING:
 692                     LOG.warn("{}: Transaction {} is still committing, cannot abort", logContext,
 693                         currentTx.cohort.getIdentifier());
 694                     currentTx.lastAccess = now;
 695                     processNext = false;
 696                     return;
 697                 case ABORTED:
 698                 case COMMITTED:
 699                 case FAILED:
 700                 case READY:
 701                 default:
 702                     pendingTransactions.poll();
 703             }
 704
 705             if (processNext) {
 706                 processNextTransaction();
 707             }
 708         }
 709     }
 710
 711     void startAbort(final SimpleShardDataTreeCohort cohort) {
 712         final Iterator<CommitEntry> it = pendingTransactions.iterator();
 713         if (!it.hasNext()) {
 714             LOG.debug("{}: no open transaction while attempting to abort {}", logContext, cohort.getIdentifier());
 715             return;
 716         }
 717
 718         // First entry is special, as it may already be committing
 719         final CommitEntry first = it.next();
 720         if (cohort.equals(first.cohort)) {
 721             if (cohort.getState() != State.COMMIT_PENDING) {
 722                 LOG.debug("{}: aborted head of queue {} in state {}", logContext, cohort.getIdentifier(),
 723                     cohort.getIdentifier());
 724                 pendingTransactions.poll();
 725                 processNextTransaction();
 726             } else {
 727                 LOG.warn("{}: transaction {} is committing, skipping abort", logContext, cohort.getIdentifier());
 728             }
 729
 730             return;
 731         }
 732
 733         while (it.hasNext()) {
 734             final CommitEntry e = it.next();
 735             if (cohort.equals(e.cohort)) {
 736                 LOG.debug("{}: aborting queued transaction {}", logContext, cohort.getIdentifier());
 737                 it.remove();
 738                 return;
 739             }
 740         }
 741
 742         LOG.debug("{}: aborted transaction {} not found in the queue", logContext, cohort.getIdentifier());
 743     }
 744
 745     void setRunOnPendingTransactionsComplete(final Runnable operation) {
 746         runOnPendingTransactionsComplete = operation;
 747         maybeRunOperationOnPendingTransactionsComplete();
 748     }
 749
 750     private void maybeRunOperationOnPendingTransactionsComplete() {
 751       if (runOnPendingTransactionsComplete != null && pendingTransactions.isEmpty()) {
 752           LOG.debug("{}: Pending transactions complete - running operation {}", logContext,
 753                   runOnPendingTransactionsComplete);
 754
 755           runOnPendingTransactionsComplete.run();
 756           runOnPendingTransactionsComplete = null;
 757       }
 758   }
 759 }