2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.cluster.datastore;
11 import akka.actor.ActorSelection;
12 import akka.dispatch.Mapper;
13 import akka.dispatch.OnComplete;
14 import com.google.common.annotations.VisibleForTesting;
15 import com.google.common.base.FinalizablePhantomReference;
16 import com.google.common.base.FinalizableReferenceQueue;
17 import com.google.common.base.Optional;
18 import com.google.common.base.Preconditions;
19 import com.google.common.collect.Lists;
20 import com.google.common.util.concurrent.CheckedFuture;
21 import com.google.common.util.concurrent.ListenableFuture;
22 import com.google.common.util.concurrent.SettableFuture;
23 import java.util.ArrayList;
24 import java.util.Collection;
25 import java.util.Collections;
26 import java.util.HashMap;
27 import java.util.List;
29 import java.util.concurrent.ConcurrentHashMap;
30 import java.util.concurrent.Semaphore;
31 import java.util.concurrent.TimeUnit;
32 import java.util.concurrent.atomic.AtomicBoolean;
33 import java.util.concurrent.atomic.AtomicLong;
34 import javax.annotation.concurrent.GuardedBy;
35 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
36 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
37 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
38 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
39 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
40 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
41 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
42 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
43 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
44 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
45 import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture;
46 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
47 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
48 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
49 import org.slf4j.Logger;
50 import org.slf4j.LoggerFactory;
51 import scala.concurrent.Future;
52 import scala.concurrent.Promise;
53 import scala.concurrent.duration.FiniteDuration;
56 * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
58 * Creating a transaction on the consumer side will create one instance of a transaction proxy. If during
59 * the transaction reads and writes are done on data that belongs to different shards then a separate transaction will
60 * be created on each of those shards by the TransactionProxy
63 * The TransactionProxy does not make any guarantees about atomicity or order in which the transactions on the various
64 * shards will be executed.
67 public class TransactionProxy implements DOMStoreReadWriteTransaction {
69 public static enum TransactionType {
75 static final Mapper<Throwable, Throwable> SAME_FAILURE_TRANSFORMER =
76 new Mapper<Throwable, Throwable>() {
78 public Throwable apply(Throwable failure) {
83 private static final AtomicLong counter = new AtomicLong();
85 private static final Logger LOG = LoggerFactory.getLogger(TransactionProxy.class);
88 * Time interval in between transaction create retries.
90 private static final FiniteDuration CREATE_TX_TRY_INTERVAL =
91 FiniteDuration.create(1, TimeUnit.SECONDS);
94 * Used to enqueue the PhantomReferences for read-only TransactionProxy instances. The
95 * FinalizableReferenceQueue is safe to use statically in an OSGi environment as it uses some
96 * trickery to clean up its internal thread when the bundle is unloaded.
98 private static final FinalizableReferenceQueue phantomReferenceQueue =
99 new FinalizableReferenceQueue();
102 * This stores the TransactionProxyCleanupPhantomReference instances statically, This is
103 * necessary because PhantomReferences need a hard reference so they're not garbage collected.
104 * Once finalized, the TransactionProxyCleanupPhantomReference removes itself from this map
105 * and thus becomes eligible for garbage collection.
107 private static final Map<TransactionProxyCleanupPhantomReference,
108 TransactionProxyCleanupPhantomReference> phantomReferenceCache =
109 new ConcurrentHashMap<>();
112 * A PhantomReference that closes remote transactions for a TransactionProxy when it's
113 * garbage collected. This is used for read-only transactions as they're not explicitly closed
114 * by clients. So the only way to detect that a transaction is no longer in use and it's safe
115 * to clean up is when it's garbage collected. It's inexact as to when an instance will be GC'ed
116 * but TransactionProxy instances should generally be short-lived enough to avoid being moved
117 * to the old generation space and thus should be cleaned up in a timely manner as the GC
118 * runs on the young generation (eden, swap1...) space much more frequently.
120 private static class TransactionProxyCleanupPhantomReference
121 extends FinalizablePhantomReference<TransactionProxy> {
123 private final List<ActorSelection> remoteTransactionActors;
124 private final AtomicBoolean remoteTransactionActorsMB;
125 private final ActorContext actorContext;
126 private final TransactionIdentifier identifier;
128 protected TransactionProxyCleanupPhantomReference(TransactionProxy referent) {
129 super(referent, phantomReferenceQueue);
131 // Note we need to cache the relevant fields from the TransactionProxy as we can't
132 // have a hard reference to the TransactionProxy instance itself.
134 remoteTransactionActors = referent.remoteTransactionActors;
135 remoteTransactionActorsMB = referent.remoteTransactionActorsMB;
136 actorContext = referent.actorContext;
137 identifier = referent.identifier;
141 public void finalizeReferent() {
142 LOG.trace("Cleaning up {} Tx actors for TransactionProxy {}",
143 remoteTransactionActors.size(), identifier);
145 phantomReferenceCache.remove(this);
147 // Access the memory barrier volatile to ensure all previous updates to the
148 // remoteTransactionActors list are visible to this thread.
150 if(remoteTransactionActorsMB.get()) {
151 for(ActorSelection actor : remoteTransactionActors) {
152 LOG.trace("Sending CloseTransaction to {}", actor);
153 actorContext.sendOperationAsync(actor, CloseTransaction.INSTANCE.toSerializable());
160 * Stores the remote Tx actors for each requested data store path to be used by the
161 * PhantomReference to close the remote Tx's. This is only used for read-only Tx's. The
162 * remoteTransactionActorsMB volatile serves as a memory barrier to publish updates to the
163 * remoteTransactionActors list so they will be visible to the thread accessing the
166 private List<ActorSelection> remoteTransactionActors;
167 private volatile AtomicBoolean remoteTransactionActorsMB;
170 * Stores the create transaction results per shard.
172 private final Map<String, TransactionFutureCallback> txFutureCallbackMap = new HashMap<>();
174 private final TransactionType transactionType;
175 private final ActorContext actorContext;
176 private final TransactionIdentifier identifier;
177 private final String transactionChainId;
178 private final SchemaContext schemaContext;
179 private boolean inReadyState;
181 private volatile boolean initialized;
182 private Semaphore operationLimiter;
183 private OperationCompleter operationCompleter;
185 public TransactionProxy(ActorContext actorContext, TransactionType transactionType) {
186 this(actorContext, transactionType, "");
189 public TransactionProxy(ActorContext actorContext, TransactionType transactionType,
190 String transactionChainId) {
191 this.actorContext = Preconditions.checkNotNull(actorContext,
192 "actorContext should not be null");
193 this.transactionType = Preconditions.checkNotNull(transactionType,
194 "transactionType should not be null");
195 this.schemaContext = Preconditions.checkNotNull(actorContext.getSchemaContext(),
196 "schemaContext should not be null");
197 this.transactionChainId = transactionChainId;
199 String memberName = actorContext.getCurrentMemberName();
200 if(memberName == null){
201 memberName = "UNKNOWN-MEMBER";
204 this.identifier = new TransactionIdentifier(memberName, counter.getAndIncrement());
206 LOG.debug("Created txn {} of type {} on chain {}", identifier, transactionType, transactionChainId);
210 List<Future<Object>> getRecordedOperationFutures() {
211 List<Future<Object>> recordedOperationFutures = Lists.newArrayList();
212 for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
213 TransactionContext transactionContext = txFutureCallback.getTransactionContext();
214 if(transactionContext != null) {
215 recordedOperationFutures.addAll(transactionContext.getRecordedOperationFutures());
219 return recordedOperationFutures;
223 boolean hasTransactionContext() {
224 for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
225 TransactionContext transactionContext = txFutureCallback.getTransactionContext();
226 if(transactionContext != null) {
235 public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
237 Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
238 "Read operation on write-only transaction is not allowed");
240 LOG.debug("Tx {} read {}", identifier, path);
244 final SettableFuture<Optional<NormalizedNode<?, ?>>> proxyFuture = SettableFuture.create();
246 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
247 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
249 public void invoke(TransactionContext transactionContext) {
250 transactionContext.readData(path, proxyFuture);
254 return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
258 public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
260 Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
261 "Exists operation on write-only transaction is not allowed");
263 LOG.debug("Tx {} exists {}", identifier, path);
267 final SettableFuture<Boolean> proxyFuture = SettableFuture.create();
269 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
270 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
272 public void invoke(TransactionContext transactionContext) {
273 transactionContext.dataExists(path, proxyFuture);
277 return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
280 private void checkModificationState() {
281 Preconditions.checkState(transactionType != TransactionType.READ_ONLY,
282 "Modification operation on read-only transaction is not allowed");
283 Preconditions.checkState(!inReadyState,
284 "Transaction is sealed - further modifications are not allowed");
287 private void throttleOperation() {
288 throttleOperation(1);
291 private void throttleOperation(int acquirePermits) {
293 // Note : Currently mailbox-capacity comes from akka.conf and not from the config-subsystem
294 operationLimiter = new Semaphore(actorContext.getTransactionOutstandingOperationLimit());
295 operationCompleter = new OperationCompleter(operationLimiter);
297 // Make sure we write this last because it's volatile and will also publish the non-volatile writes
298 // above as well so they'll be visible to other threads.
303 if(!operationLimiter.tryAcquire(acquirePermits,
304 actorContext.getDatastoreContext().getOperationTimeoutInSeconds(), TimeUnit.SECONDS)){
305 LOG.warn("Failed to acquire operation permit for transaction {}", getIdentifier());
307 } catch (InterruptedException e) {
308 if(LOG.isDebugEnabled()) {
309 LOG.debug("Interrupted when trying to acquire operation permit for transaction " + getIdentifier().toString(), e);
311 LOG.warn("Interrupted when trying to acquire operation permit for transaction {}", getIdentifier());
318 public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
320 checkModificationState();
322 LOG.debug("Tx {} write {}", identifier, path);
326 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
327 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
329 public void invoke(TransactionContext transactionContext) {
330 transactionContext.writeData(path, data);
336 public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
338 checkModificationState();
340 LOG.debug("Tx {} merge {}", identifier, path);
344 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
345 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
347 public void invoke(TransactionContext transactionContext) {
348 transactionContext.mergeData(path, data);
354 public void delete(final YangInstanceIdentifier path) {
356 checkModificationState();
358 LOG.debug("Tx {} delete {}", identifier, path);
362 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
363 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
365 public void invoke(TransactionContext transactionContext) {
366 transactionContext.deleteData(path);
372 public DOMStoreThreePhaseCommitCohort ready() {
374 checkModificationState();
378 LOG.debug("Tx {} Readying {} transactions for commit", identifier,
379 txFutureCallbackMap.size());
381 if(txFutureCallbackMap.size() == 0) {
382 onTransactionReady(Collections.<Future<ActorSelection>>emptyList());
383 return NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
386 throttleOperation(txFutureCallbackMap.size());
388 List<Future<ActorSelection>> cohortFutures = Lists.newArrayList();
390 for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
392 LOG.debug("Tx {} Readying transaction for shard {} chain {}", identifier,
393 txFutureCallback.getShardName(), transactionChainId);
395 final TransactionContext transactionContext = txFutureCallback.getTransactionContext();
396 final Future<ActorSelection> future;
397 if (transactionContext != null) {
398 // avoid the creation of a promise and a TransactionOperation
399 future = transactionContext.readyTransaction();
401 final Promise<ActorSelection> promise = akka.dispatch.Futures.promise();
402 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
404 public void invoke(TransactionContext transactionContext) {
405 promise.completeWith(transactionContext.readyTransaction());
408 future = promise.future();
411 cohortFutures.add(future);
414 onTransactionReady(cohortFutures);
416 return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
417 identifier.toString());
421 * Method for derived classes to be notified when the transaction has been readied.
423 * @param cohortFutures the cohort Futures for each shard transaction.
425 protected void onTransactionReady(List<Future<ActorSelection>> cohortFutures) {
429 * Method called to send a CreateTransaction message to a shard.
431 * @param shard the shard actor to send to
432 * @param serializedCreateMessage the serialized message to send
433 * @return the response Future
435 protected Future<Object> sendCreateTransaction(ActorSelection shard,
436 Object serializedCreateMessage) {
437 return actorContext.executeOperationAsync(shard, serializedCreateMessage);
441 public Object getIdentifier() {
442 return this.identifier;
446 public void close() {
447 for (TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
448 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
450 public void invoke(TransactionContext transactionContext) {
451 transactionContext.closeTransaction();
456 txFutureCallbackMap.clear();
458 if(remoteTransactionActorsMB != null) {
459 remoteTransactionActors.clear();
460 remoteTransactionActorsMB.set(true);
464 private String shardNameFromIdentifier(YangInstanceIdentifier path){
465 return ShardStrategyFactory.getStrategy(path).findShard(path);
468 private TransactionFutureCallback getOrCreateTxFutureCallback(YangInstanceIdentifier path) {
469 String shardName = shardNameFromIdentifier(path);
470 TransactionFutureCallback txFutureCallback = txFutureCallbackMap.get(shardName);
471 if(txFutureCallback == null) {
472 Future<ActorSelection> findPrimaryFuture = actorContext.findPrimaryShardAsync(shardName);
474 final TransactionFutureCallback newTxFutureCallback =
475 new TransactionFutureCallback(shardName);
477 txFutureCallback = newTxFutureCallback;
478 txFutureCallbackMap.put(shardName, txFutureCallback);
480 findPrimaryFuture.onComplete(new OnComplete<ActorSelection>() {
482 public void onComplete(Throwable failure, ActorSelection primaryShard) {
483 if(failure != null) {
484 newTxFutureCallback.onComplete(failure, null);
486 newTxFutureCallback.setPrimaryShard(primaryShard);
489 }, actorContext.getClientDispatcher());
492 return txFutureCallback;
495 public String getTransactionChainId() {
496 return transactionChainId;
499 protected ActorContext getActorContext() {
504 * Interfaces for transaction operations to be invoked later.
506 private static interface TransactionOperation {
507 void invoke(TransactionContext transactionContext);
511 * Implements a Future OnComplete callback for a CreateTransaction message. This class handles
512 * retries, up to a limit, if the shard doesn't have a leader yet. This is done by scheduling a
513 * retry task after a short delay.
515 * The end result from a completed CreateTransaction message is a TransactionContext that is
516 * used to perform transaction operations. Transaction operations that occur before the
517 * CreateTransaction completes are cache and executed once the CreateTransaction completes,
518 * successfully or not.
520 private class TransactionFutureCallback extends OnComplete<Object> {
523 * The list of transaction operations to execute once the CreateTransaction completes.
525 @GuardedBy("txOperationsOnComplete")
526 private final List<TransactionOperation> txOperationsOnComplete = Lists.newArrayList();
529 * The TransactionContext resulting from the CreateTransaction reply.
531 private volatile TransactionContext transactionContext;
534 * The target primary shard.
536 private volatile ActorSelection primaryShard;
538 private volatile int createTxTries = (int) (actorContext.getDatastoreContext().
539 getShardLeaderElectionTimeout().duration().toMillis() /
540 CREATE_TX_TRY_INTERVAL.toMillis());
542 private final String shardName;
544 TransactionFutureCallback(String shardName) {
545 this.shardName = shardName;
548 String getShardName() {
552 TransactionContext getTransactionContext() {
553 return transactionContext;
558 * Sets the target primary shard and initiates a CreateTransaction try.
560 void setPrimaryShard(ActorSelection primaryShard) {
561 LOG.debug("Tx {} Primary shard found - trying create transaction", identifier);
563 this.primaryShard = primaryShard;
564 tryCreateTransaction();
568 * Adds a TransactionOperation to be executed after the CreateTransaction completes.
570 void addTxOperationOnComplete(TransactionOperation operation) {
571 boolean invokeOperation = true;
572 synchronized(txOperationsOnComplete) {
573 if(transactionContext == null) {
574 LOG.debug("Tx {} Adding operation on complete {}", identifier);
576 invokeOperation = false;
577 txOperationsOnComplete.add(operation);
581 if(invokeOperation) {
582 operation.invoke(transactionContext);
586 void enqueueTransactionOperation(final TransactionOperation op) {
588 if (transactionContext != null) {
589 op.invoke(transactionContext);
591 // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
592 // callback to be executed after the Tx is created.
593 addTxOperationOnComplete(op);
598 * Performs a CreateTransaction try async.
600 private void tryCreateTransaction() {
601 Future<Object> createTxFuture = sendCreateTransaction(primaryShard,
602 new CreateTransaction(identifier.toString(),
603 TransactionProxy.this.transactionType.ordinal(),
604 getTransactionChainId()).toSerializable());
606 createTxFuture.onComplete(this, actorContext.getClientDispatcher());
610 public void onComplete(Throwable failure, Object response) {
611 if(failure instanceof NoShardLeaderException) {
612 // There's no leader for the shard yet - schedule and try again, unless we're out
613 // of retries. Note: createTxTries is volatile as it may be written by different
614 // threads however not concurrently, therefore decrementing it non-atomically here
616 if(--createTxTries > 0) {
617 LOG.debug("Tx {} Shard {} has no leader yet - scheduling create Tx retry",
618 identifier, shardName);
620 actorContext.getActorSystem().scheduler().scheduleOnce(CREATE_TX_TRY_INTERVAL,
624 tryCreateTransaction();
626 }, actorContext.getClientDispatcher());
631 // Mainly checking for state violation here to perform a volatile read of "initialized" to
632 // ensure updates to operationLimter et al are visible to this thread (ie we're doing
633 // "piggy-back" synchronization here).
634 Preconditions.checkState(initialized, "Tx was not propertly initialized.");
636 // Create the TransactionContext from the response or failure. Store the new
637 // TransactionContext locally until we've completed invoking the
638 // TransactionOperations. This avoids thread timing issues which could cause
639 // out-of-order TransactionOperations. Eg, on a modification operation, if the
640 // TransactionContext is non-null, then we directly call the TransactionContext.
641 // However, at the same time, the code may be executing the cached
642 // TransactionOperations. So to avoid thus timing, we don't publish the
643 // TransactionContext until after we've executed all cached TransactionOperations.
644 TransactionContext localTransactionContext;
645 if(failure != null) {
646 LOG.debug("Tx {} Creating NoOpTransaction because of error: {}", identifier,
647 failure.getMessage());
649 localTransactionContext = new NoOpTransactionContext(failure, identifier, operationLimiter);
650 } else if (response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
651 localTransactionContext = createValidTransactionContext(
652 CreateTransactionReply.fromSerializable(response));
654 IllegalArgumentException exception = new IllegalArgumentException(String.format(
655 "Invalid reply type %s for CreateTransaction", response.getClass()));
657 localTransactionContext = new NoOpTransactionContext(exception, identifier, operationLimiter);
660 executeTxOperatonsOnComplete(localTransactionContext);
663 private void executeTxOperatonsOnComplete(TransactionContext localTransactionContext) {
665 // Access to txOperationsOnComplete and transactionContext must be protected and atomic
666 // (ie synchronized) with respect to #addTxOperationOnComplete to handle timing
667 // issues and ensure no TransactionOperation is missed and that they are processed
668 // in the order they occurred.
670 // We'll make a local copy of the txOperationsOnComplete list to handle re-entrancy
671 // in case a TransactionOperation results in another transaction operation being
672 // queued (eg a put operation from a client read Future callback that is notified
674 Collection<TransactionOperation> operationsBatch = null;
675 synchronized(txOperationsOnComplete) {
676 if(txOperationsOnComplete.isEmpty()) {
677 // We're done invoking the TransactionOperations so we can now publish the
678 // TransactionContext.
679 transactionContext = localTransactionContext;
683 operationsBatch = new ArrayList<>(txOperationsOnComplete);
684 txOperationsOnComplete.clear();
687 // Invoke TransactionOperations outside the sync block to avoid unnecessary blocking.
688 // A slight down-side is that we need to re-acquire the lock below but this should
690 for(TransactionOperation oper: operationsBatch) {
691 oper.invoke(localTransactionContext);
696 private TransactionContext createValidTransactionContext(CreateTransactionReply reply) {
697 String transactionPath = reply.getTransactionPath();
699 LOG.debug("Tx {} Received {}", identifier, reply);
701 ActorSelection transactionActor = actorContext.actorSelection(transactionPath);
703 if (transactionType == TransactionType.READ_ONLY) {
704 // Read-only Tx's aren't explicitly closed by the client so we create a PhantomReference
705 // to close the remote Tx's when this instance is no longer in use and is garbage
708 if(remoteTransactionActorsMB == null) {
709 remoteTransactionActors = Lists.newArrayList();
710 remoteTransactionActorsMB = new AtomicBoolean();
712 TransactionProxyCleanupPhantomReference cleanup =
713 new TransactionProxyCleanupPhantomReference(TransactionProxy.this);
714 phantomReferenceCache.put(cleanup, cleanup);
717 // Add the actor to the remoteTransactionActors list for access by the
718 // cleanup PhantonReference.
719 remoteTransactionActors.add(transactionActor);
721 // Write to the memory barrier volatile to publish the above update to the
722 // remoteTransactionActors list for thread visibility.
723 remoteTransactionActorsMB.set(true);
726 // TxActor is always created where the leader of the shard is.
727 // Check if TxActor is created in the same node
728 boolean isTxActorLocal = actorContext.isPathLocal(transactionPath);
730 if(reply.getVersion() >= DataStoreVersions.LITHIUM_VERSION) {
731 return new TransactionContextImpl(transactionPath, transactionActor, identifier,
732 actorContext, schemaContext, isTxActorLocal, reply.getVersion(), operationCompleter);
734 return new LegacyTransactionContextImpl(transactionPath, transactionActor, identifier,
735 actorContext, schemaContext, isTxActorLocal, reply.getVersion(), operationCompleter);
740 private static class NoOpDOMStoreThreePhaseCommitCohort implements DOMStoreThreePhaseCommitCohort {
741 static NoOpDOMStoreThreePhaseCommitCohort INSTANCE = new NoOpDOMStoreThreePhaseCommitCohort();
743 private static final ListenableFuture<Void> IMMEDIATE_VOID_SUCCESS =
744 com.google.common.util.concurrent.Futures.immediateFuture(null);
745 private static final ListenableFuture<Boolean> IMMEDIATE_BOOLEAN_SUCCESS =
746 com.google.common.util.concurrent.Futures.immediateFuture(Boolean.TRUE);
748 private NoOpDOMStoreThreePhaseCommitCohort() {
752 public ListenableFuture<Boolean> canCommit() {
753 return IMMEDIATE_BOOLEAN_SUCCESS;
757 public ListenableFuture<Void> preCommit() {
758 return IMMEDIATE_VOID_SUCCESS;
762 public ListenableFuture<Void> abort() {
763 return IMMEDIATE_VOID_SUCCESS;
767 public ListenableFuture<Void> commit() {
768 return IMMEDIATE_VOID_SUCCESS;