*/
package org.opendaylight.controller.md.sal.dom.store.impl;
+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 com.google.common.util.concurrent.MoreExecutors;
+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;
import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
import org.opendaylight.controller.md.sal.dom.store.impl.SnapshotBackedWriteTransaction.TransactionReadyPrototype;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.ConflictingModificationAppliedException;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerTree;
-import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
import org.opendaylight.controller.sal.core.spi.data.DOMStore;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
import org.opendaylight.yangtools.concepts.Identifiable;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
+import org.opendaylight.yangtools.util.ExecutorServiceUtil;
+import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager;
+import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager.Invoker;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.ConflictingModificationAppliedException;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
+import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.concurrent.GuardedBy;
-import java.util.Collections;
-import java.util.concurrent.Callable;
-import java.util.concurrent.atomic.AtomicLong;
-
-import static com.google.common.base.Preconditions.checkState;
-
/**
* In-memory DOM Data Store
*
public class InMemoryDOMDataStore implements DOMStore, Identifiable<String>, SchemaContextListener,
TransactionReadyPrototype,AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataStore.class);
+ private static final ListenableFuture<Void> SUCCESSFUL_FUTURE = Futures.immediateFuture(null);
+
+ private static final Invoker<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> DCL_NOTIFICATION_MGR_INVOKER =
+ new Invoker<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent>() {
+ @Override
+ public void invokeListener(final DataChangeListenerRegistration<?> listener,
+ final DOMImmutableDataChangeEvent notification ) {
+ final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> inst = listener.getInstance();
+ if (inst != null) {
+ inst.onDataChanged(notification);
+ }
+ }
+ };
+
private final DataTree dataTree = InMemoryDataTreeFactory.getInstance().create();
private final ListenerTree listenerTree = ListenerTree.create();
private final AtomicLong txCounter = new AtomicLong(0);
- private final ListeningExecutorService executor;
+ private final ListeningExecutorService listeningExecutor;
+
+ private final QueuedNotificationManager<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> dataChangeListenerNotificationManager;
+ private final ExecutorService dataChangeListenerExecutor;
+ private final ExecutorService domStoreExecutor;
+ private final boolean debugTransactions;
private final String name;
- public InMemoryDOMDataStore(final String name, final ListeningExecutorService executor) {
+ private volatile AutoCloseable closeable;
+
+ public InMemoryDOMDataStore(final String name, final ExecutorService domStoreExecutor,
+ final ExecutorService dataChangeListenerExecutor) {
+ this(name, domStoreExecutor, dataChangeListenerExecutor,
+ InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_LISTENER_QUEUE_SIZE, false);
+ }
+
+ public InMemoryDOMDataStore(final String name, final ExecutorService domStoreExecutor,
+ final ExecutorService dataChangeListenerExecutor, final int maxDataChangeListenerQueueSize,
+ final boolean debugTransactions) {
this.name = Preconditions.checkNotNull(name);
- this.executor = Preconditions.checkNotNull(executor);
+ this.domStoreExecutor = Preconditions.checkNotNull(domStoreExecutor);
+ this.listeningExecutor = MoreExecutors.listeningDecorator(this.domStoreExecutor);
+ this.dataChangeListenerExecutor = Preconditions.checkNotNull(dataChangeListenerExecutor);
+ this.debugTransactions = debugTransactions;
+
+ dataChangeListenerNotificationManager =
+ new QueuedNotificationManager<>(this.dataChangeListenerExecutor,
+ DCL_NOTIFICATION_MGR_INVOKER, maxDataChangeListenerQueueSize,
+ "DataChangeListenerQueueMgr");
+ }
+
+ public void setCloseable(AutoCloseable closeable) {
+ this.closeable = closeable;
+ }
+
+ public QueuedNotificationManager<?, ?> getDataChangeListenerNotificationManager() {
+ return dataChangeListenerNotificationManager;
+ }
+
+ public ExecutorService getDomStoreExecutor() {
+ return domStoreExecutor;
}
@Override
@Override
public DOMStoreReadTransaction newReadOnlyTransaction() {
- return new SnapshotBackedReadTransaction(nextIdentifier(), dataTree.takeSnapshot());
+ return new SnapshotBackedReadTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot());
}
@Override
public DOMStoreReadWriteTransaction newReadWriteTransaction() {
- return new SnapshotBackedReadWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this);
+ return new SnapshotBackedReadWriteTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot(), this);
}
@Override
public DOMStoreWriteTransaction newWriteOnlyTransaction() {
- return new SnapshotBackedWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this);
+ return new SnapshotBackedWriteTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot(), this);
}
@Override
}
@Override
- public void close(){
- executor.shutdownNow();
+ public void close() {
+ ExecutorServiceUtil.tryGracefulShutdown(listeningExecutor, 30, TimeUnit.SECONDS);
+ ExecutorServiceUtil.tryGracefulShutdown(dataChangeListenerExecutor, 30, TimeUnit.SECONDS);
+
+ if(closeable != null) {
+ try {
+ closeable.close();
+ } catch(Exception e) {
+ LOG.debug("Error closing instance", e);
+ }
+ }
+ }
+
+ boolean getDebugTransactions() {
+ return debugTransactions;
}
+
@Override
- public <L extends AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>> ListenerRegistration<L> registerChangeListener(
- final InstanceIdentifier path, final L listener, final DataChangeScope scope) {
+ public <L extends AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>> ListenerRegistration<L> registerChangeListener(
+ final YangInstanceIdentifier path, final L listener, final DataChangeScope scope) {
/*
* Make sure commit is not occurring right now. Listener has to be
.setAfter(data) //
.addCreated(path, data) //
.build();
- executor.submit(new ChangeListenerNotifyTask(Collections.singletonList(reg), event));
+
+ dataChangeListenerNotificationManager.submitNotification(reg, event);
}
}
}
@Override
- public synchronized DOMStoreThreePhaseCommitCohort ready(final SnapshotBackedWriteTransaction writeTx) {
+ public DOMStoreThreePhaseCommitCohort ready(final SnapshotBackedWriteTransaction writeTx) {
LOG.debug("Tx: {} is submitted. Modifications: {}", writeTx.getIdentifier(), writeTx.getMutatedView());
return new ThreePhaseCommitImpl(writeTx);
}
} else {
snapshot = dataTree.takeSnapshot();
}
- return new SnapshotBackedReadTransaction(nextIdentifier(), snapshot);
+ return new SnapshotBackedReadTransaction(nextIdentifier(), getDebugTransactions(), snapshot);
}
@Override
snapshot = dataTree.takeSnapshot();
}
final SnapshotBackedReadWriteTransaction ret = new SnapshotBackedReadWriteTransaction(nextIdentifier(),
- snapshot, this);
+ getDebugTransactions(), snapshot, this);
latestOutstandingTx = ret;
return ret;
}
} else {
snapshot = dataTree.takeSnapshot();
}
- final SnapshotBackedWriteTransaction ret = new SnapshotBackedWriteTransaction(nextIdentifier(), snapshot,
- this);
+ final SnapshotBackedWriteTransaction ret = new SnapshotBackedWriteTransaction(nextIdentifier(),
+ getDebugTransactions(), snapshot, this);
latestOutstandingTx = ret;
return ret;
}
@Override
public void close() {
- executor.shutdownNow();
-
+ // 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,
}
public synchronized void onTransactionCommited(final SnapshotBackedWriteTransaction transaction) {
- // If commited transaction is latestOutstandingTx we clear
+ // If committed transaction is latestOutstandingTx we clear
// latestOutstandingTx
// field in order to base new transactions on Datastore Data Tree
// directly.
@Override
public ListenableFuture<Boolean> canCommit() {
- return executor.submit(new Callable<Boolean>() {
+ return listeningExecutor.submit(new Callable<Boolean>() {
@Override
public Boolean call() throws TransactionCommitFailedException {
try {
} 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);
}
}
@Override
public ListenableFuture<Void> preCommit() {
- return executor.submit(new Callable<Void>() {
+ return listeningExecutor.submit(new Callable<Void>() {
@Override
public Void call() {
candidate = dataTree.prepare(modification);
@Override
public ListenableFuture<Void> abort() {
candidate = null;
- return Futures.immediateFuture(null);
+ return SUCCESSFUL_FUTURE;
}
@Override
*/
synchronized (this) {
dataTree.commit(candidate);
-
- for (ChangeListenerNotifyTask task : listenerResolver.call()) {
- LOG.trace("Scheduling invocation of listeners: {}", task);
- executor.submit(task);
- }
+ listenerResolver.resolve(dataChangeListenerNotificationManager);
}
- return Futures.immediateFuture(null);
+ return SUCCESSFUL_FUTURE;
}
}
}
Code Review / controller.git / blobdiff