opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionChainProxyTest.java

   1 /*
   2  * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.datastore;
  10
  11 import static org.junit.Assert.assertEquals;
  12 import static org.junit.Assert.fail;
  13 import static org.mockito.Matchers.any;
  14 import static org.mockito.Matchers.eq;
  15 import static org.mockito.Matchers.isA;
  16 import static org.mockito.Mockito.doReturn;
  17 import static org.mockito.Mockito.never;
  18 import static org.mockito.Mockito.timeout;
  19 import static org.mockito.Mockito.times;
  20 import static org.mockito.Mockito.verify;
  21 import static org.opendaylight.controller.cluster.datastore.TransactionType.READ_WRITE;
  22 import static org.opendaylight.controller.cluster.datastore.TransactionType.WRITE_ONLY;
  23
  24 import akka.actor.ActorRef;
  25 import akka.util.Timeout;
  26 import java.util.concurrent.CountDownLatch;
  27 import java.util.concurrent.TimeUnit;
  28 import java.util.concurrent.atomic.AtomicReference;
  29 import java.util.function.Function;
  30 import org.junit.Assert;
  31 import org.junit.Test;
  32 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
  33 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  34 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
  35 import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
  36 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  37 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
  38 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
  39 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransaction;
  40 import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
  41 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  42 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  43 import scala.concurrent.Promise;
  44
  45 public class TransactionChainProxyTest extends AbstractTransactionProxyTest {
  46     private LocalHistoryIdentifier historyId;
  47
  48     @Override
  49     public void setUp() {
  50         super.setUp();
  51         historyId = MockIdentifiers.historyIdentifier(TransactionChainProxyTest.class, memberName);
  52     }
  53
  54     @SuppressWarnings("resource")
  55     @Test
  56     public void testNewReadOnlyTransaction() {
  57
  58         DOMStoreTransaction dst = new TransactionChainProxy(mockComponentFactory, historyId).newReadOnlyTransaction();
  59         Assert.assertTrue(dst instanceof DOMStoreReadTransaction);
  60
  61     }
  62
  63     @SuppressWarnings("resource")
  64     @Test
  65     public void testNewReadWriteTransaction() {
  66         DOMStoreTransaction dst = new TransactionChainProxy(mockComponentFactory, historyId).newReadWriteTransaction();
  67         Assert.assertTrue(dst instanceof DOMStoreReadWriteTransaction);
  68
  69     }
  70
  71     @SuppressWarnings("resource")
  72     @Test
  73     public void testNewWriteOnlyTransaction() {
  74         DOMStoreTransaction dst = new TransactionChainProxy(mockComponentFactory, historyId).newWriteOnlyTransaction();
  75         Assert.assertTrue(dst instanceof DOMStoreWriteTransaction);
  76
  77     }
  78
  79     @SuppressWarnings("unchecked")
  80     @Test
  81     public void testClose() {
  82         new TransactionChainProxy(mockComponentFactory, historyId).close();
  83
  84         verify(mockActorContext, times(1)).broadcast(any(Function.class), any(Class.class));
  85     }
  86
  87     @Test
  88     public void testRateLimitingUsedInReadWriteTxCreation() {
  89         try (TransactionChainProxy txChainProxy = new TransactionChainProxy(mockComponentFactory, historyId)) {
  90
  91             txChainProxy.newReadWriteTransaction();
  92
  93             verify(mockActorContext, times(1)).acquireTxCreationPermit();
  94         }
  95     }
  96
  97     @Test
  98     public void testRateLimitingUsedInWriteOnlyTxCreation() {
  99         try (TransactionChainProxy txChainProxy = new TransactionChainProxy(mockComponentFactory, historyId)) {
 100
 101             txChainProxy.newWriteOnlyTransaction();
 102
 103             verify(mockActorContext, times(1)).acquireTxCreationPermit();
 104         }
 105     }
 106
 107     @Test
 108     public void testRateLimitingNotUsedInReadOnlyTxCreation() {
 109         try (TransactionChainProxy txChainProxy = new TransactionChainProxy(mockComponentFactory, historyId)) {
 110
 111             txChainProxy.newReadOnlyTransaction();
 112
 113             verify(mockActorContext, times(0)).acquireTxCreationPermit();
 114         }
 115     }
 116
 117     /**
 118      * Tests 2 successive chained write-only transactions and verifies the second transaction isn't
 119      * initiated until the first one completes its read future.
 120      */
 121     @Test
 122     @SuppressWarnings("checkstyle:IllegalCatch")
 123     public void testChainedWriteOnlyTransactions() throws Exception {
 124         dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
 125
 126         try (TransactionChainProxy txChainProxy = new TransactionChainProxy(mockComponentFactory, historyId)) {
 127
 128             ActorRef txActorRef1 = setupActorContextWithoutInitialCreateTransaction(getSystem());
 129
 130             Promise<Object> batchedReplyPromise1 = akka.dispatch.Futures.promise();
 131             doReturn(batchedReplyPromise1.future()).when(mockActorContext).executeOperationAsync(
 132                     eq(actorSelection(txActorRef1)), isA(BatchedModifications.class), any(Timeout.class));
 133
 134             DOMStoreWriteTransaction writeTx1 = txChainProxy.newWriteOnlyTransaction();
 135
 136             NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 137             writeTx1.write(TestModel.TEST_PATH, writeNode1);
 138
 139             writeTx1.ready();
 140
 141             verify(mockActorContext, times(1)).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 142
 143             verifyOneBatchedModification(txActorRef1, new WriteModification(TestModel.TEST_PATH, writeNode1), true);
 144
 145             ActorRef txActorRef2 = setupActorContextWithoutInitialCreateTransaction(getSystem());
 146
 147             expectBatchedModifications(txActorRef2, 1);
 148
 149             final NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
 150
 151             final DOMStoreWriteTransaction writeTx2 = txChainProxy.newWriteOnlyTransaction();
 152
 153             final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 154             final CountDownLatch write2Complete = new CountDownLatch(1);
 155             new Thread(() -> {
 156                 try {
 157                     writeTx2.write(TestModel.OUTER_LIST_PATH, writeNode2);
 158                 } catch (Exception e) {
 159                     caughtEx.set(e);
 160                 } finally {
 161                     write2Complete.countDown();
 162                 }
 163             }).start();
 164
 165             assertEquals("Tx 2 write should've completed", true, write2Complete.await(5, TimeUnit.SECONDS));
 166
 167             if (caughtEx.get() != null) {
 168                 throw caughtEx.get();
 169             }
 170
 171             try {
 172                 verify(mockActorContext, times(1)).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 173             } catch (AssertionError e) {
 174                 fail("Tx 2 should not have initiated until the Tx 1's ready future completed");
 175             }
 176
 177             batchedReplyPromise1.success(readyTxReply(txActorRef1.path().toString()).value().get().get());
 178
 179             // Tx 2 should've proceeded to find the primary shard.
 180             verify(mockActorContext, timeout(5000).times(2)).findPrimaryShardAsync(
 181                     eq(DefaultShardStrategy.DEFAULT_SHARD));
 182         }
 183     }
 184
 185     /**
 186      * Tests 2 successive chained read-write transactions and verifies the second transaction isn't
 187      * initiated until the first one completes its read future.
 188      */
 189     @Test
 190     @SuppressWarnings("checkstyle:IllegalCatch")
 191     public void testChainedReadWriteTransactions() throws Exception {
 192         try (TransactionChainProxy txChainProxy = new TransactionChainProxy(mockComponentFactory, historyId)) {
 193
 194             ActorRef txActorRef1 = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 195
 196             expectBatchedModifications(txActorRef1, 1);
 197
 198             Promise<Object> readyReplyPromise1 = akka.dispatch.Futures.promise();
 199             doReturn(readyReplyPromise1.future()).when(mockActorContext).executeOperationAsync(
 200                     eq(actorSelection(txActorRef1)), isA(BatchedModifications.class), any(Timeout.class));
 201
 202             DOMStoreWriteTransaction writeTx1 = txChainProxy.newReadWriteTransaction();
 203
 204             NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 205             writeTx1.write(TestModel.TEST_PATH, writeNode1);
 206
 207             writeTx1.ready();
 208
 209             verifyOneBatchedModification(txActorRef1, new WriteModification(TestModel.TEST_PATH, writeNode1), true);
 210
 211             String tx2MemberName = "mock-member";
 212             ActorRef shardActorRef2 = setupActorContextWithoutInitialCreateTransaction(getSystem());
 213             ActorRef txActorRef2 = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE,
 214                     DataStoreVersions.CURRENT_VERSION, tx2MemberName, shardActorRef2);
 215
 216             expectBatchedModifications(txActorRef2, 1);
 217
 218             final NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
 219
 220             final DOMStoreWriteTransaction writeTx2 = txChainProxy.newReadWriteTransaction();
 221
 222             final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 223             final CountDownLatch write2Complete = new CountDownLatch(1);
 224             new Thread(() -> {
 225                 try {
 226                     writeTx2.write(TestModel.OUTER_LIST_PATH, writeNode2);
 227                 } catch (Exception e) {
 228                     caughtEx.set(e);
 229                 } finally {
 230                     write2Complete.countDown();
 231                 }
 232             }).start();
 233
 234             assertEquals("Tx 2 write should've completed", true, write2Complete.await(5, TimeUnit.SECONDS));
 235
 236             if (caughtEx.get() != null) {
 237                 throw caughtEx.get();
 238             }
 239
 240             try {
 241                 verify(mockActorContext, never()).executeOperationAsync(
 242                         eq(getSystem().actorSelection(shardActorRef2.path())),
 243                         eqCreateTransaction(tx2MemberName, READ_WRITE));
 244             } catch (AssertionError e) {
 245                 fail("Tx 2 should not have initiated until the Tx 1's ready future completed");
 246             }
 247
 248             readyReplyPromise1.success(readyTxReply(txActorRef1.path().toString()).value().get().get());
 249
 250             verify(mockActorContext, timeout(5000)).executeOperationAsync(
 251                     eq(getSystem().actorSelection(shardActorRef2.path())),
 252                     eqCreateTransaction(tx2MemberName, READ_WRITE), any(Timeout.class));
 253         }
 254     }
 255
 256     @Test(expected = IllegalStateException.class)
 257     public void testChainedWriteTransactionsWithPreviousTxNotReady() {
 258         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 259
 260         expectBatchedModifications(actorRef, 1);
 261
 262         try (TransactionChainProxy txChainProxy = new TransactionChainProxy(mockComponentFactory, historyId)) {
 263
 264             DOMStoreWriteTransaction writeTx1 = txChainProxy.newWriteOnlyTransaction();
 265
 266             NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 267             writeTx1.write(TestModel.TEST_PATH, writeNode1);
 268
 269             txChainProxy.newWriteOnlyTransaction();
 270         }
 271     }
 272 }