@GuardedBy("this")
private final DOMDataTreePrefixTable<DOMDataTreeProducer> producers = DOMDataTreePrefixTable.create();
-
void removeShard(final ShardRegistration<?> reg) {
final DOMDataTreeIdentifier prefix = reg.getPrefix();
final ShardRegistration<?> parentReg;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import java.util.Collection;
+import com.google.common.util.concurrent.SettableFuture;
import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
+import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
import javax.annotation.Nonnull;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.NotThreadSafe;
Preconditions.checkState(openCursor == null, "Cannot submit transaction while there is a cursor open");
final Set<DOMDataTreeShardWriteTransaction> txns = ImmutableSet.copyOf(idToTransaction.values());
- for (final DOMDataTreeShardWriteTransaction tx : txns) {
- tx.ready();
- }
+ final ListenableFuture<List<Void>> listListenableFuture =
+ Futures.allAsList(txns.stream().map(tx -> {
+ tx.ready();
+ return tx.submit();
+ }).collect(Collectors.toList()));
+
+ final SettableFuture<Void> ret = SettableFuture.create();
+ Futures.addCallback(listListenableFuture, new FutureCallback<List<Void>>() {
+ @Override
+ public void onSuccess(final List<Void> result) {
+ ret.set(null);
+ }
+
+ @Override
+ public void onFailure(final Throwable t) {
+ ret.setException(t);
+ }
+ });
+
producer.transactionSubmitted(this);
- try {
- return Futures.immediateCheckedFuture(new SubmitCoordinationTask(identifier, txns).call());
- } catch (final TransactionCommitFailedException e) {
- return Futures.immediateFailedCheckedFuture(e);
- }
+ return Futures.makeChecked(ret, TransactionCommitFailedExceptionMapper.create("submit"));
}
synchronized void cursorClosed() {
path.removeLast();
}
}
-
- private static class SubmitCoordinationTask implements Callable<Void> {
-
- private static final Logger LOG = LoggerFactory.getLogger(SubmitCoordinationTask.class);
-
- private final String identifier;
- private final Collection<DOMDataTreeShardWriteTransaction> transactions;
-
- SubmitCoordinationTask(final String identifier,
- final Collection<DOMDataTreeShardWriteTransaction> transactions) {
- this.identifier = identifier;
- this.transactions = transactions;
- }
-
- @Override
- public Void call() throws TransactionCommitFailedException {
-
- try {
- LOG.debug("Producer {}, submit started", identifier);
- submitBlocking();
-
- return null;
- } catch (final TransactionCommitFailedException e) {
- LOG.warn("Failure while submitting transaction for producer {}", identifier, e);
- //FIXME abort here
- throw e;
- }
- }
-
- void submitBlocking() throws TransactionCommitFailedException {
- for (final ListenableFuture<?> commit : submitAll()) {
- try {
- commit.get();
- } catch (InterruptedException | ExecutionException e) {
- throw new TransactionCommitFailedException("Submit failed", e);
- }
- }
- }
-
- private ListenableFuture<?>[] submitAll() {
- final ListenableFuture<?>[] ops = new ListenableFuture<?>[transactions.size()];
- int i = 0;
- for (final DOMDataTreeShardWriteTransaction tx : transactions) {
- ops[i++] = tx.submit();
- }
- return ops;
- }
- }
}
public void setUp() throws Exception {
MockitoAnnotations.initMocks(this);
- rootShard = InMemoryDOMDataTreeShard.create(ROOT_ID, executor, 1, 1);
+ rootShard = InMemoryDOMDataTreeShard.create(ROOT_ID, executor, 1);
rootShard.onGlobalContextUpdated(schemaContext);
final ShardedDOMDataTree dataTree = new ShardedDOMDataTree();
final DOMDataTreeListener mockedDataTreeListener = Mockito.mock(DOMDataTreeListener.class);
doNothing().when(mockedDataTreeListener).onDataTreeChanged(anyCollection(), anyMap());
- final InMemoryDOMDataTreeShard innerShard = InMemoryDOMDataTreeShard.create(INNER_CONTAINER_ID, executor, 1, 1);
+ final InMemoryDOMDataTreeShard innerShard = InMemoryDOMDataTreeShard.create(INNER_CONTAINER_ID, executor, 1);
innerShard.onGlobalContextUpdated(schemaContext);
final DOMDataTreeProducer shardRegProducer = dataTreeService.createProducer(Collections.singletonList(INNER_CONTAINER_ID));
innerShardReg = dataTreeService.registerDataTreeShard(INNER_CONTAINER_ID, innerShard, shardRegProducer);
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import junit.framework.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
public void setUp() throws Exception {
MockitoAnnotations.initMocks(this);
- rootShard = InMemoryDOMDataTreeShard.create(ROOT_ID, executor, 1, 1);
+ rootShard = InMemoryDOMDataTreeShard.create(ROOT_ID, executor, 1);
rootShard.onGlobalContextUpdated(schemaContext);
final ShardedDOMDataTree dataTree = new ShardedDOMDataTree();
public void testShardRegistrationClose() throws Exception {
rootShardReg.close();
- final InMemoryDOMDataTreeShard newRootShard = InMemoryDOMDataTreeShard.create(ROOT_ID, executor, 1, 1);
+ final InMemoryDOMDataTreeShard newRootShard = InMemoryDOMDataTreeShard.create(ROOT_ID, executor, 1);
newRootShard.onGlobalContextUpdated(schemaContext);
final DOMDataTreeProducer shardRegProducer = dataTreeService.createProducer(Collections.singletonList(ROOT_ID));
- ListenerRegistration<InMemoryDOMDataTreeShard> newRootShardReg =
+ final ListenerRegistration<InMemoryDOMDataTreeShard> newRootShardReg =
dataTreeService.registerDataTreeShard(ROOT_ID, rootShard, shardRegProducer);
shardRegProducer.close();
- final InMemoryDOMDataTreeShard innerShard = InMemoryDOMDataTreeShard.create(INNER_CONTAINER_ID, executor, 1, 1);
+ final InMemoryDOMDataTreeShard innerShard = InMemoryDOMDataTreeShard.create(INNER_CONTAINER_ID, executor, 1);
innerShard.onGlobalContextUpdated(schemaContext);
final DOMDataTreeProducer shardRegProducer2 = dataTreeService.createProducer(Collections.singletonList(INNER_CONTAINER_ID));
ListenerRegistration<InMemoryDOMDataTreeShard> innerShardReg = dataTreeService.registerDataTreeShard(INNER_CONTAINER_ID, innerShard, shardRegProducer2);
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import javax.annotation.Nonnull;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
import org.opendaylight.mdsal.dom.api.DOMDataTreeShard;
import org.opendaylight.mdsal.dom.spi.DOMDataTreePrefixTable;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.util.concurrent.SpecialExecutors;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.tree.CursorAwareDataTreeSnapshot;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
private final InMemoryDOMDataTreeShardChangePublisher shardChangePublisher;
private final ListeningExecutorService executor;
- private SchemaContext schemaContext;
-
private InMemoryDOMDataTreeShard(final DOMDataTreeIdentifier prefix, final ExecutorService dataTreeChangeExecutor,
- final int maxDataChangeListenerQueueSize, final int maxCommitQueueSize) {
+ final int maxDataChangeListenerQueueSize) {
this.prefix = Preconditions.checkNotNull(prefix);
final TreeType treeType = treeTypeFor(prefix.getDatastoreType());
this.dataTree = InMemoryDataTreeFactory.getInstance().create(treeType, prefix.getRootIdentifier());
this.shardChangePublisher = new InMemoryDOMDataTreeShardChangePublisher(dataTreeChangeExecutor, maxDataChangeListenerQueueSize, dataTree, prefix.getRootIdentifier(), childShards);
- this.executor = MoreExecutors.listeningDecorator(SpecialExecutors.newBoundedSingleThreadExecutor(maxCommitQueueSize, "Shard-executor[" + prefix + "]"));
+ this.executor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
}
public static InMemoryDOMDataTreeShard create(final DOMDataTreeIdentifier id, final ExecutorService dataTreeChangeExecutor,
- final int maxDataChangeListenerQueueSize, final int maxCommitQueueSize) {
- return new InMemoryDOMDataTreeShard(id, dataTreeChangeExecutor, maxDataChangeListenerQueueSize, maxCommitQueueSize);
+ final int maxDataChangeListenerQueueSize) {
+ return new InMemoryDOMDataTreeShard(id, dataTreeChangeExecutor, maxDataChangeListenerQueueSize);
}
@Override
public void onGlobalContextUpdated(final SchemaContext context) {
dataTree.setSchemaContext(context);
- schemaContext = context;
}
@Override
public void basicTest() throws Exception {
final InMemoryDOMDataTreeShard inMemoryDOMDataTreeShard =
InMemoryDOMDataTreeShard.create(DOM_DATA_TREE_IDENTIFIER,
- MoreExecutors.newDirectExecutorService(), 1, 1);
+ MoreExecutors.newDirectExecutorService(), 1);
final DOMDataTreeIdentifier domDataTreeIdentifier =
new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
final InMemoryDOMDataTreeShard inMemoryDOMDataTreeShard =
InMemoryDOMDataTreeShard.create(domDataTreeIdentifier,
- MoreExecutors.newDirectExecutorService(), 1 ,1);
+ MoreExecutors.newDirectExecutorService(), 1);
final InmemoryDOMDataTreeShardWriteTransaction inmemoryDOMDataTreeShardWriteTransaction =
mock(InmemoryDOMDataTreeShardWriteTransaction.class);
Code Review / mdsal.git / commitdiff