*/
-public class TransactionProxy implements DOMStoreReadWriteTransaction {
+public class TransactionProxy extends AbstractDOMStoreTransaction implements DOMStoreReadWriteTransaction {
public static enum TransactionType {
READ_ONLY,
WRITE_ONLY,
- READ_WRITE
+ READ_WRITE;
+
+ // Cache all values
+ private static final TransactionType[] VALUES = values();
+
+ public static TransactionType fromInt(final int type) {
+ try {
+ return VALUES[type];
+ } catch (IndexOutOfBoundsException e) {
+ throw new IllegalArgumentException("In TransactionType enum value " + type, e);
+ }
+ }
+ }
+
+ private static enum TransactionState {
+ OPEN,
+ READY,
+ CLOSED,
}
static final Mapper SAME_FAILURE_TRANSFORMER =
@@ -88,72 +107,6 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
private static final FiniteDuration CREATE_TX_TRY_INTERVAL =
FiniteDuration.create(1, TimeUnit.SECONDS);
- /**
- * Used to enqueue the PhantomReferences for read-only TransactionProxy instances. The
- * FinalizableReferenceQueue is safe to use statically in an OSGi environment as it uses some
- * trickery to clean up its internal thread when the bundle is unloaded.
- */
- private static final FinalizableReferenceQueue phantomReferenceQueue =
- new FinalizableReferenceQueue();
-
- /**
- * This stores the TransactionProxyCleanupPhantomReference instances statically, This is
- * necessary because PhantomReferences need a hard reference so they're not garbage collected.
- * Once finalized, the TransactionProxyCleanupPhantomReference removes itself from this map
- * and thus becomes eligible for garbage collection.
- */
- private static final Map phantomReferenceCache =
- new ConcurrentHashMap<>();
-
- /**
- * A PhantomReference that closes remote transactions for a TransactionProxy when it's
- * garbage collected. This is used for read-only transactions as they're not explicitly closed
- * by clients. So the only way to detect that a transaction is no longer in use and it's safe
- * to clean up is when it's garbage collected. It's inexact as to when an instance will be GC'ed
- * but TransactionProxy instances should generally be short-lived enough to avoid being moved
- * to the old generation space and thus should be cleaned up in a timely manner as the GC
- * runs on the young generation (eden, swap1...) space much more frequently.
- */
- private static class TransactionProxyCleanupPhantomReference
- extends FinalizablePhantomReference {
-
- private final List remoteTransactionActors;
- private final AtomicBoolean remoteTransactionActorsMB;
- private final ActorContext actorContext;
- private final TransactionIdentifier identifier;
-
- protected TransactionProxyCleanupPhantomReference(TransactionProxy referent) {
- super(referent, phantomReferenceQueue);
-
- // Note we need to cache the relevant fields from the TransactionProxy as we can't
- // have a hard reference to the TransactionProxy instance itself.
-
- remoteTransactionActors = referent.remoteTransactionActors;
- remoteTransactionActorsMB = referent.remoteTransactionActorsMB;
- actorContext = referent.actorContext;
- identifier = referent.identifier;
- }
-
- @Override
- public void finalizeReferent() {
- LOG.trace("Cleaning up {} Tx actors for TransactionProxy {}",
- remoteTransactionActors.size(), identifier);
-
- phantomReferenceCache.remove(this);
-
- // Access the memory barrier volatile to ensure all previous updates to the
- // remoteTransactionActors list are visible to this thread.
-
- if(remoteTransactionActorsMB.get()) {
- for(ActorSelection actor : remoteTransactionActors) {
- LOG.trace("Sending CloseTransaction to {}", actor);
- actorContext.sendOperationAsync(actor, CloseTransaction.INSTANCE.toSerializable());
- }
- }
- }
- }
-
/**
* Stores the remote Tx actors for each requested data store path to be used by the
* PhantomReference to close the remote Tx's. This is only used for read-only Tx's. The
@@ -161,8 +114,8 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
* remoteTransactionActors list so they will be visible to the thread accessing the
* PhantomReference.
*/
- private List remoteTransactionActors;
- private AtomicBoolean remoteTransactionActorsMB;
+ List remoteTransactionActors;
+ volatile AtomicBoolean remoteTransactionActorsMB;
/**
* Stores the create transaction results per shard.
@@ -170,20 +123,21 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
private final Map txFutureCallbackMap = new HashMap<>();
private final TransactionType transactionType;
- private final ActorContext actorContext;
- private final TransactionIdentifier identifier;
+ final ActorContext actorContext;
private final String transactionChainId;
private final SchemaContext schemaContext;
- private boolean inReadyState;
- private final Semaphore operationLimiter;
- private final OperationCompleter operationCompleter;
+ private TransactionState state = TransactionState.OPEN;
+
+ private volatile boolean initialized;
+ private Semaphore operationLimiter;
+ private OperationCompleter operationCompleter;
public TransactionProxy(ActorContext actorContext, TransactionType transactionType) {
this(actorContext, transactionType, "");
}
- public TransactionProxy(ActorContext actorContext, TransactionType transactionType,
- String transactionChainId) {
+ public TransactionProxy(ActorContext actorContext, TransactionType transactionType, String transactionChainId) {
+ super(createIdentifier(actorContext));
this.actorContext = Preconditions.checkNotNull(actorContext,
"actorContext should not be null");
this.transactionType = Preconditions.checkNotNull(transactionType,
@@ -192,32 +146,16 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
"schemaContext should not be null");
this.transactionChainId = transactionChainId;
+ LOG.debug("Created txn {} of type {} on chain {}", getIdentifier(), transactionType, transactionChainId);
+ }
+
+ private static TransactionIdentifier createIdentifier(ActorContext actorContext) {
String memberName = actorContext.getCurrentMemberName();
- if(memberName == null){
+ if (memberName == null) {
memberName = "UNKNOWN-MEMBER";
}
- this.identifier = TransactionIdentifier.builder().memberName(memberName).counter(
- counter.getAndIncrement()).build();
-
- if(transactionType == TransactionType.READ_ONLY) {
- // Read-only Tx's aren't explicitly closed by the client so we create a PhantomReference
- // to close the remote Tx's when this instance is no longer in use and is garbage
- // collected.
-
- remoteTransactionActors = Lists.newArrayList();
- remoteTransactionActorsMB = new AtomicBoolean();
-
- TransactionProxyCleanupPhantomReference cleanup =
- new TransactionProxyCleanupPhantomReference(this);
- phantomReferenceCache.put(cleanup, cleanup);
- }
-
- // Note : Currently mailbox-capacity comes from akka.conf and not from the config-subsystem
- this.operationLimiter = new Semaphore(actorContext.getTransactionOutstandingOperationLimit());
- this.operationCompleter = new OperationCompleter(operationLimiter);
-
- LOG.debug("Created txn {} of type {} on chain {}", identifier, transactionType, transactionChainId);
+ return new TransactionIdentifier(memberName, counter.getAndIncrement());
}
@VisibleForTesting
@@ -225,8 +163,8 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
List> recordedOperationFutures = Lists.newArrayList();
for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
TransactionContext transactionContext = txFutureCallback.getTransactionContext();
- if(transactionContext != null) {
- recordedOperationFutures.addAll(transactionContext.getRecordedOperationFutures());
+ if (transactionContext != null) {
+ transactionContext.copyRecordedOperationFutures(recordedOperationFutures);
}
}
@@ -245,36 +183,81 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
return false;
}
+ private boolean isRootPath(YangInstanceIdentifier path){
+ return !path.getPathArguments().iterator().hasNext();
+ }
+
@Override
public CheckedFuture>, ReadFailedException> read(final YangInstanceIdentifier path) {
Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
"Read operation on write-only transaction is not allowed");
- LOG.debug("Tx {} read {}", identifier, path);
-
- throttleOperation();
+ LOG.debug("Tx {} read {}", getIdentifier(), path);
final SettableFuture>> proxyFuture = SettableFuture.create();
- TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
- txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
- @Override
- public void invoke(TransactionContext transactionContext) {
- transactionContext.readData(path, proxyFuture);
- }
- });
+ if(isRootPath(path)){
+ readAllData(path, proxyFuture);
+ } else {
+ throttleOperation();
+
+ TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
+ @Override
+ public void invoke(TransactionContext transactionContext) {
+ transactionContext.readData(path, proxyFuture);
+ }
+ });
+
+ }
return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
}
+ private void readAllData(final YangInstanceIdentifier path,
+ final SettableFuture>> proxyFuture) {
+ Set allShardNames = actorContext.getConfiguration().getAllShardNames();
+ List>>> futures = new ArrayList<>(allShardNames.size());
+
+ for(String shardName : allShardNames){
+ final SettableFuture>> subProxyFuture = SettableFuture.create();
+
+ throttleOperation();
+
+ TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(shardName);
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
+ @Override
+ public void invoke(TransactionContext transactionContext) {
+ transactionContext.readData(path, subProxyFuture);
+ }
+ });
+
+ futures.add(subProxyFuture);
+ }
+
+ final ListenableFuture>>> future = Futures.allAsList(futures);
+
+ future.addListener(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ proxyFuture.set(NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.builder().build(),
+ future.get(), actorContext.getSchemaContext()));
+ } catch (InterruptedException | ExecutionException e) {
+ proxyFuture.setException(e);
+ }
+ }
+ }, actorContext.getActorSystem().dispatcher());
+ }
+
@Override
public CheckedFuture exists(final YangInstanceIdentifier path) {
Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
"Exists operation on write-only transaction is not allowed");
- LOG.debug("Tx {} exists {}", identifier, path);
+ LOG.debug("Tx {} exists {}", getIdentifier(), path);
throttleOperation();
@@ -294,7 +277,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
private void checkModificationState() {
Preconditions.checkState(transactionType != TransactionType.READ_ONLY,
"Modification operation on read-only transaction is not allowed");
- Preconditions.checkState(!inReadyState,
+ Preconditions.checkState(state == TransactionState.OPEN,
"Transaction is sealed - further modifications are not allowed");
}
@@ -303,6 +286,16 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
}
private void throttleOperation(int acquirePermits) {
+ if(!initialized) {
+ // Note : Currently mailbox-capacity comes from akka.conf and not from the config-subsystem
+ operationLimiter = new Semaphore(actorContext.getTransactionOutstandingOperationLimit());
+ operationCompleter = new OperationCompleter(operationLimiter);
+
+ // Make sure we write this last because it's volatile and will also publish the non-volatile writes
+ // above as well so they'll be visible to other threads.
+ initialized = true;
+ }
+
try {
if(!operationLimiter.tryAcquire(acquirePermits,
actorContext.getDatastoreContext().getOperationTimeoutInSeconds(), TimeUnit.SECONDS)){
@@ -317,13 +310,12 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
}
}
-
@Override
public void write(final YangInstanceIdentifier path, final NormalizedNode, ?> data) {
checkModificationState();
- LOG.debug("Tx {} write {}", identifier, path);
+ LOG.debug("Tx {} write {}", getIdentifier(), path);
throttleOperation();
@@ -341,7 +333,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
checkModificationState();
- LOG.debug("Tx {} merge {}", identifier, path);
+ LOG.debug("Tx {} merge {}", getIdentifier(), path);
throttleOperation();
@@ -359,7 +351,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
checkModificationState();
- LOG.debug("Tx {} delete {}", identifier, path);
+ LOG.debug("Tx {} delete {}", getIdentifier(), path);
throttleOperation();
@@ -372,23 +364,37 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
});
}
- @Override
- public DOMStoreThreePhaseCommitCohort ready() {
-
- checkModificationState();
+ private boolean seal(final TransactionState newState) {
+ if (state == TransactionState.OPEN) {
+ state = newState;
+ return true;
+ } else {
+ return false;
+ }
+ }
- throttleOperation(txFutureCallbackMap.size());
+ @Override
+ public AbstractThreePhaseCommitCohort ready() {
+ Preconditions.checkState(transactionType != TransactionType.READ_ONLY,
+ "Read-only transactions cannot be readied");
- inReadyState = true;
+ final boolean success = seal(TransactionState.READY);
+ Preconditions.checkState(success, "Transaction %s is %s, it cannot be readied", getIdentifier(), state);
- LOG.debug("Tx {} Readying {} transactions for commit", identifier,
+ LOG.debug("Tx {} Readying {} transactions for commit", getIdentifier(),
txFutureCallbackMap.size());
- List> cohortFutures = Lists.newArrayList();
+ if (txFutureCallbackMap.isEmpty()) {
+ TransactionRateLimitingCallback.adjustRateLimitForUnusedTransaction(actorContext);
+ return NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
+ }
+
+ throttleOperation(txFutureCallbackMap.size());
+ List> cohortFutures = new ArrayList<>(txFutureCallbackMap.size());
for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
- LOG.debug("Tx {} Readying transaction for shard {} chain {}", identifier,
+ LOG.debug("Tx {} Readying transaction for shard {} chain {}", getIdentifier(),
txFutureCallback.getShardName(), transactionChainId);
final TransactionContext transactionContext = txFutureCallback.getTransactionContext();
@@ -410,39 +416,22 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
cohortFutures.add(future);
}
- onTransactionReady(cohortFutures);
-
return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
- identifier.toString());
- }
-
- /**
- * Method for derived classes to be notified when the transaction has been readied.
- *
- * @param cohortFutures the cohort Futures for each shard transaction.
- */
- protected void onTransactionReady(List> cohortFutures) {
- }
-
- /**
- * Method called to send a CreateTransaction message to a shard.
- *
- * @param shard the shard actor to send to
- * @param serializedCreateMessage the serialized message to send
- * @return the response Future
- */
- protected Future