import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.CheckedFuture;
-import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.controller.sal.core.spi.data.AbstractDOMStoreTransaction;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
-import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
WRITE_ONLY,
READ_WRITE;
- public static TransactionType fromInt(int type) {
- if(type == WRITE_ONLY.ordinal()) {
- return WRITE_ONLY;
- } else if(type == READ_WRITE.ordinal()) {
- return READ_WRITE;
- } else if(type == READ_ONLY.ordinal()) {
- return READ_ONLY;
- } else {
- throw new IllegalArgumentException("In TransactionType enum value" + type);
+ // 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<Throwable, Throwable> SAME_FAILURE_TRANSFORMER =
new Mapper<Throwable, Throwable>() {
@Override
private final ActorContext actorContext;
private final String transactionChainId;
private final SchemaContext schemaContext;
- private boolean inReadyState;
+ private TransactionState state = TransactionState.OPEN;
private volatile boolean initialized;
private Semaphore operationLimiter;
List<Future<Object>> 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);
}
}
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");
}
}
}
-
@Override
public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
});
}
- @Override
- public DOMStoreThreePhaseCommitCohort ready() {
+ private boolean seal(final TransactionState newState) {
+ if (state == TransactionState.OPEN) {
+ state = newState;
+ return true;
+ } else {
+ return false;
+ }
+ }
- checkModificationState();
+ @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", getIdentifier(),
txFutureCallbackMap.size());
- if(txFutureCallbackMap.size() == 0) {
- onTransactionReady(Collections.<Future<ActorSelection>>emptyList());
+ if (txFutureCallbackMap.isEmpty()) {
TransactionRateLimitingCallback.adjustRateLimitForUnusedTransaction(actorContext);
return NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
}
throttleOperation(txFutureCallbackMap.size());
- List<Future<ActorSelection>> cohortFutures = Lists.newArrayList();
-
+ List<Future<ActorSelection>> cohortFutures = new ArrayList<>(txFutureCallbackMap.size());
for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
LOG.debug("Tx {} Readying transaction for shard {} chain {}", getIdentifier(),
cohortFutures.add(future);
}
- onTransactionReady(cohortFutures);
-
return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
getIdentifier().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<Future<ActorSelection>> cohortFutures) {
- }
-
@Override
public void close() {
+ if (!seal(TransactionState.CLOSED)) {
+ if (state == TransactionState.CLOSED) {
+ // Idempotent no-op as per AutoCloseable recommendation
+ return;
+ }
+
+ throw new IllegalStateException(String.format("Transaction %s is ready, it cannot be closed",
+ getIdentifier()));
+ }
+
for (TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
@Override
return actorContext;
}
- /**
- * Interfaces for transaction operations to be invoked later.
- */
- private static interface TransactionOperation {
- void invoke(TransactionContext transactionContext);
- }
-
/**
* Implements a Future OnComplete callback for a CreateTransaction message. This class handles
* retries, up to a limit, if the shard doesn't have a leader yet. This is done by scheduling a
}
}
}
-
- private static class NoOpDOMStoreThreePhaseCommitCohort implements DOMStoreThreePhaseCommitCohort {
- static NoOpDOMStoreThreePhaseCommitCohort INSTANCE = new NoOpDOMStoreThreePhaseCommitCohort();
-
- private static final ListenableFuture<Void> IMMEDIATE_VOID_SUCCESS =
- com.google.common.util.concurrent.Futures.immediateFuture(null);
- private static final ListenableFuture<Boolean> IMMEDIATE_BOOLEAN_SUCCESS =
- com.google.common.util.concurrent.Futures.immediateFuture(Boolean.TRUE);
-
- private NoOpDOMStoreThreePhaseCommitCohort() {
- }
-
- @Override
- public ListenableFuture<Boolean> canCommit() {
- return IMMEDIATE_BOOLEAN_SUCCESS;
- }
-
- @Override
- public ListenableFuture<Void> preCommit() {
- return IMMEDIATE_VOID_SUCCESS;
- }
-
- @Override
- public ListenableFuture<Void> abort() {
- return IMMEDIATE_VOID_SUCCESS;
- }
-
- @Override
- public ListenableFuture<Void> commit() {
- return IMMEDIATE_VOID_SUCCESS;
- }
- }
}
Code Review / controller.git / blobdiff