import static com.google.common.base.Preconditions.checkState;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
-import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.ListeningExecutorService;
-import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
-import javax.annotation.concurrent.GuardedBy;
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
import org.opendaylight.controller.md.sal.common.api.data.OptimisticLockFailedException;
public class InMemoryDOMDataStore extends TransactionReadyPrototype implements DOMStore, Identifiable<String>, SchemaContextListener, AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataStore.class);
private static final ListenableFuture<Void> SUCCESSFUL_FUTURE = Futures.immediateFuture(null);
+ private static final ListenableFuture<Boolean> CAN_COMMIT_FUTURE = Futures.immediateFuture(Boolean.TRUE);
private static final Invoker<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> DCL_NOTIFICATION_MGR_INVOKER =
new Invoker<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent>() {
private final QueuedNotificationManager<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> dataChangeListenerNotificationManager;
private final ExecutorService dataChangeListenerExecutor;
- private final ListeningExecutorService commitExecutor;
private final boolean debugTransactions;
private final String name;
private volatile AutoCloseable closeable;
- public InMemoryDOMDataStore(final String name, final ListeningExecutorService commitExecutor,
- final ExecutorService dataChangeListenerExecutor) {
- this(name, commitExecutor, dataChangeListenerExecutor,
- InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_LISTENER_QUEUE_SIZE, false);
+ public InMemoryDOMDataStore(final String name, final ExecutorService dataChangeListenerExecutor) {
+ this(name, dataChangeListenerExecutor, InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_LISTENER_QUEUE_SIZE, false);
}
- public InMemoryDOMDataStore(final String name, final ListeningExecutorService commitExecutor,
- final ExecutorService dataChangeListenerExecutor, final int maxDataChangeListenerQueueSize,
- final boolean debugTransactions) {
+ public InMemoryDOMDataStore(final String name, final ExecutorService dataChangeListenerExecutor,
+ final int maxDataChangeListenerQueueSize, final boolean debugTransactions) {
this.name = Preconditions.checkNotNull(name);
- this.commitExecutor = Preconditions.checkNotNull(commitExecutor);
this.dataChangeListenerExecutor = Preconditions.checkNotNull(dataChangeListenerExecutor);
this.debugTransactions = debugTransactions;
return dataChangeListenerNotificationManager;
}
- public ExecutorService getDomStoreExecutor() {
- return commitExecutor;
- }
-
@Override
public final String getIdentifier() {
return name;
@Override
public DOMStoreTransactionChain createTransactionChain() {
- return new DOMStoreTransactionChainImpl();
+ return new DOMStoreTransactionChainImpl(this);
}
@Override
@Override
public void close() {
- ExecutorServiceUtil.tryGracefulShutdown(commitExecutor, 30, TimeUnit.SECONDS);
ExecutorServiceUtil.tryGracefulShutdown(dataChangeListenerExecutor, 30, TimeUnit.SECONDS);
if(closeable != null) {
}
}
- boolean getDebugTransactions() {
+ public final boolean getDebugTransactions() {
return debugTransactions;
}
+ final DataTreeSnapshot takeSnapshot() {
+ return dataTree.takeSnapshot();
+ }
+
@Override
public <L extends AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>> ListenerRegistration<L> registerChangeListener(
final YangInstanceIdentifier path, final L listener, final DataChangeScope scope) {
return new ThreePhaseCommitImpl(tx, tree);
}
- private Object nextIdentifier() {
+ Object nextIdentifier() {
return name + "-" + txCounter.getAndIncrement();
}
- private class DOMStoreTransactionChainImpl extends TransactionReadyPrototype implements DOMStoreTransactionChain {
- @GuardedBy("this")
- private SnapshotBackedWriteTransaction allocatedTransaction;
- @GuardedBy("this")
- private DataTreeSnapshot readySnapshot;
- @GuardedBy("this")
- private boolean chainFailed = false;
-
- @GuardedBy("this")
- private void checkFailed() {
- Preconditions.checkState(!chainFailed, "Transaction chain is failed.");
- }
-
- @GuardedBy("this")
- private DataTreeSnapshot getSnapshot() {
- checkFailed();
-
- if (allocatedTransaction != null) {
- Preconditions.checkState(readySnapshot != null, "Previous transaction %s is not ready yet", allocatedTransaction.getIdentifier());
- return readySnapshot;
- } else {
- return dataTree.takeSnapshot();
- }
- }
-
- @GuardedBy("this")
- private <T extends SnapshotBackedWriteTransaction> T recordTransaction(final T transaction) {
- allocatedTransaction = transaction;
- readySnapshot = null;
- return transaction;
- }
-
- @Override
- public synchronized DOMStoreReadTransaction newReadOnlyTransaction() {
- final DataTreeSnapshot snapshot = getSnapshot();
- return new SnapshotBackedReadTransaction(nextIdentifier(), getDebugTransactions(), snapshot);
- }
-
- @Override
- public synchronized DOMStoreReadWriteTransaction newReadWriteTransaction() {
- final DataTreeSnapshot snapshot = getSnapshot();
- return recordTransaction(new SnapshotBackedReadWriteTransaction(nextIdentifier(),
- getDebugTransactions(), snapshot, this));
- }
-
- @Override
- public synchronized DOMStoreWriteTransaction newWriteOnlyTransaction() {
- final DataTreeSnapshot snapshot = getSnapshot();
- return recordTransaction(new SnapshotBackedWriteTransaction(nextIdentifier(),
- getDebugTransactions(), snapshot, this));
- }
-
- @Override
- protected synchronized void transactionAborted(final SnapshotBackedWriteTransaction tx) {
- if (tx.equals(allocatedTransaction)) {
- Preconditions.checkState(readySnapshot == null, "Unexpected abort of transaction %s with ready snapshot %s", tx, readySnapshot);
- allocatedTransaction = null;
- }
- }
-
- @Override
- protected synchronized DOMStoreThreePhaseCommitCohort transactionReady(final SnapshotBackedWriteTransaction tx, final DataTreeModification tree) {
- Preconditions.checkState(tx.equals(allocatedTransaction), "Mis-ordered ready transaction %s last allocated was %s", tx, allocatedTransaction);
- if (readySnapshot != null) {
- // The snapshot should have been cleared
- LOG.warn("Uncleared snapshot {} encountered, overwritten with transaction {} snapshot {}", readySnapshot, tx, tree);
- }
-
- final DOMStoreThreePhaseCommitCohort cohort = InMemoryDOMDataStore.this.transactionReady(tx, tree);
- readySnapshot = tree;
- return new ChainedTransactionCommitImpl(tx, cohort, this);
- }
-
- @Override
- public void close() {
- // FIXME: this call doesn't look right here - listeningExecutor is shared and owned
- // by the outer class.
- //listeningExecutor.shutdownNow();
- }
-
- protected synchronized void onTransactionFailed(final SnapshotBackedWriteTransaction transaction,
- final Throwable t) {
- chainFailed = true;
- }
-
- public synchronized void onTransactionCommited(final SnapshotBackedWriteTransaction transaction) {
- // If the committed transaction was the one we allocated last,
- // we clear it and the ready snapshot, so the next transaction
- // allocated refers to the data tree directly.
- if (transaction.equals(allocatedTransaction)) {
- if (readySnapshot == null) {
- LOG.warn("Transaction {} committed while no ready snapshot present", transaction);
- }
-
- allocatedTransaction = null;
- readySnapshot = null;
- }
- }
- }
-
- private static class ChainedTransactionCommitImpl implements DOMStoreThreePhaseCommitCohort {
- private final SnapshotBackedWriteTransaction transaction;
- private final DOMStoreThreePhaseCommitCohort delegate;
- private final DOMStoreTransactionChainImpl txChain;
-
- protected ChainedTransactionCommitImpl(final SnapshotBackedWriteTransaction transaction,
- final DOMStoreThreePhaseCommitCohort delegate, final DOMStoreTransactionChainImpl txChain) {
- this.transaction = transaction;
- this.delegate = delegate;
- this.txChain = txChain;
- }
-
- @Override
- public ListenableFuture<Boolean> canCommit() {
- return delegate.canCommit();
- }
-
- @Override
- public ListenableFuture<Void> preCommit() {
- return delegate.preCommit();
- }
-
- @Override
- public ListenableFuture<Void> abort() {
- return delegate.abort();
- }
-
- @Override
- public ListenableFuture<Void> commit() {
- ListenableFuture<Void> commitFuture = delegate.commit();
- Futures.addCallback(commitFuture, new FutureCallback<Void>() {
- @Override
- public void onFailure(final Throwable t) {
- txChain.onTransactionFailed(transaction, t);
- }
-
- @Override
- public void onSuccess(final Void result) {
- txChain.onTransactionCommited(transaction);
- }
- });
- return commitFuture;
- }
- }
-
- private class ThreePhaseCommitImpl implements DOMStoreThreePhaseCommitCohort {
+ private final class ThreePhaseCommitImpl implements DOMStoreThreePhaseCommitCohort {
private final SnapshotBackedWriteTransaction transaction;
private final DataTreeModification modification;
@Override
public ListenableFuture<Boolean> canCommit() {
- return commitExecutor.submit(new Callable<Boolean>() {
- @Override
- public Boolean call() throws TransactionCommitFailedException {
- try {
- dataTree.validate(modification);
- LOG.debug("Store Transaction: {} can be committed", transaction.getIdentifier());
- return true;
- } catch (ConflictingModificationAppliedException e) {
- LOG.warn("Store Tx: {} Conflicting modification for {}.", transaction.getIdentifier(),
- e.getPath());
- transaction.warnDebugContext(LOG);
- throw new OptimisticLockFailedException("Optimistic lock failed.",e);
- } catch (DataValidationFailedException e) {
- LOG.warn("Store Tx: {} Data Precondition failed for {}.", transaction.getIdentifier(),
- e.getPath(), e);
- transaction.warnDebugContext(LOG);
- throw new TransactionCommitFailedException("Data did not pass validation.",e);
- }
- }
- });
+ try {
+ dataTree.validate(modification);
+ LOG.debug("Store Transaction: {} can be committed", transaction.getIdentifier());
+ return CAN_COMMIT_FUTURE;
+ } catch (ConflictingModificationAppliedException e) {
+ LOG.warn("Store Tx: {} Conflicting modification for {}.", transaction.getIdentifier(),
+ e.getPath());
+ transaction.warnDebugContext(LOG);
+ return Futures.immediateFailedFuture(new OptimisticLockFailedException("Optimistic lock failed.", e));
+ } catch (DataValidationFailedException e) {
+ LOG.warn("Store Tx: {} Data Precondition failed for {}.", transaction.getIdentifier(),
+ e.getPath(), e);
+ transaction.warnDebugContext(LOG);
+
+ // For debugging purposes, allow dumping of the modification. Coupled with the above
+ // precondition log, it should allow us to understand what went on.
+ LOG.trace("Store Tx: {} modifications: {} tree: {}", modification, dataTree);
+
+ return Futures.immediateFailedFuture(new TransactionCommitFailedException("Data did not pass validation.", e));
+ } catch (Exception e) {
+ LOG.warn("Unexpected failure in validation phase", e);
+ return Futures.immediateFailedFuture(e);
+ }
}
@Override
public ListenableFuture<Void> preCommit() {
- return commitExecutor.submit(new Callable<Void>() {
- @Override
- public Void call() {
- candidate = dataTree.prepare(modification);
- listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree);
- return null;
- }
- });
+ try {
+ candidate = dataTree.prepare(modification);
+ listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree);
+ return SUCCESSFUL_FUTURE;
+ } catch (Exception e) {
+ LOG.warn("Unexpected failure in pre-commit phase", e);
+ return Futures.immediateFailedFuture(e);
+ }
}
@Override
Code Review / controller.git / blobdiff