X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-remoterpc-connector%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fremote%2Frpc%2Fregistry%2FRpcRegistryTest.java;h=5b7b7e4fdc61dfaaa2377078853b691b48b07394;hp=e0d145dbe179b4af06109f1df28151b066b50cd8;hb=c083d5c6ef874e38eebca402105e740f3614d6f7;hpb=87eeb0d62755bf5d6bcfd07d40dd8e0ab86c155e diff --git a/opendaylight/md-sal/sal-remoterpc-connector/src/test/java/org/opendaylight/controller/remote/rpc/registry/RpcRegistryTest.java b/opendaylight/md-sal/sal-remoterpc-connector/src/test/java/org/opendaylight/controller/remote/rpc/registry/RpcRegistryTest.java index e0d145dbe1..5b7b7e4fdc 100644 --- a/opendaylight/md-sal/sal-remoterpc-connector/src/test/java/org/opendaylight/controller/remote/rpc/registry/RpcRegistryTest.java +++ b/opendaylight/md-sal/sal-remoterpc-connector/src/test/java/org/opendaylight/controller/remote/rpc/registry/RpcRegistryTest.java @@ -1,38 +1,41 @@ package org.opendaylight.controller.remote.rpc.registry; -import akka.actor.ActorPath; import akka.actor.ActorRef; -import akka.actor.ActorSelection; import akka.actor.ActorSystem; -import akka.actor.ChildActorPath; +import akka.actor.Address; import akka.actor.Props; -import akka.pattern.Patterns; +import akka.japi.Pair; import akka.testkit.JavaTestKit; -import akka.util.Timeout; -import com.google.common.base.Predicate; +import com.google.common.util.concurrent.Uninterruptibles; import java.net.URI; import java.net.URISyntaxException; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; -import javax.annotation.Nullable; import org.junit.After; import org.junit.AfterClass; +import org.junit.Assert; import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig; import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl; import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.AddOrUpdateRoutes; +import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.FindRouters; +import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.FindRoutersReply; import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.RemoveRoutes; import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.SetLocalRouter; -import org.opendaylight.controller.remote.rpc.registry.gossip.Messages; +import org.opendaylight.controller.remote.rpc.registry.gossip.Bucket; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBuckets; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBucketsReply; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersions; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersionsReply; import org.opendaylight.controller.sal.connector.api.RpcRouter; -import org.opendaylight.controller.utils.ConditionalProbe; +import org.opendaylight.controller.sal.connector.api.RpcRouter.RouteIdentifier; import org.opendaylight.yangtools.yang.common.QName; -import scala.concurrent.Await; -import scala.concurrent.Future; import scala.concurrent.duration.Duration; import scala.concurrent.duration.FiniteDuration; @@ -46,6 +49,8 @@ public class RpcRegistryTest { private ActorRef registry2; private ActorRef registry3; + private int routeIdCounter = 1; + @BeforeClass public static void staticSetup() throws InterruptedException { RemoteRpcProviderConfig config1 = new RemoteRpcProviderConfig.Builder("memberA").build(); @@ -97,27 +102,30 @@ public class RpcRegistryTest { final JavaTestKit mockBroker = new JavaTestKit(node1); - final ActorPath bucketStorePath = new ChildActorPath(registry1.path(), "store"); + Address nodeAddress = node1.provider().getDefaultAddress(); // Add rpc on node 1 registry1.tell(new SetLocalRouter(mockBroker.getRef()), mockBroker.getRef()); - // install probe - final JavaTestKit probe1 = createProbeForMessage(node1, bucketStorePath, - Messages.BucketStoreMessages.UpdateBucket.class); + List> addedRouteIds = createRouteIds(); - registry1.tell(getAddRouteMessage(), mockBroker.getRef()); + registry1.tell(new AddOrUpdateRoutes(addedRouteIds), mockBroker.getRef()); // Bucket store should get an update bucket message. Updated bucket contains added rpc. - probe1.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS), - Messages.BucketStoreMessages.UpdateBucket.class); + + Map buckets = retrieveBuckets(registry1, mockBroker, nodeAddress); + verifyBucket(buckets.get(nodeAddress), addedRouteIds); + + Map versions = retrieveVersions(registry1, mockBroker); + Assert.assertEquals("Version for bucket " + nodeAddress, buckets.get(nodeAddress).getVersion(), + versions.get(nodeAddress)); // Now remove rpc - registry1.tell(getRemoveRouteMessage(), mockBroker.getRef()); + registry1.tell(new RemoveRoutes(addedRouteIds), mockBroker.getRef()); // Bucket store should get an update bucket message. Rpc is removed in the updated bucket - probe1.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS), - Messages.BucketStoreMessages.UpdateBucket.class); + + verifyEmptyBucket(mockBroker, registry1, nodeAddress); System.out.println("testAddRemoveRpcOnSameNode ending"); @@ -136,30 +144,52 @@ public class RpcRegistryTest { System.out.println("testRpcAddRemoveInCluster starting"); final JavaTestKit mockBroker1 = new JavaTestKit(node1); + final JavaTestKit mockBroker2 = new JavaTestKit(node2); + + List> addedRouteIds = createRouteIds(); - // install probe on node2's bucket store - final ActorPath bucketStorePath = new ChildActorPath(registry2.path(), "store"); - final JavaTestKit probe2 = createProbeForMessage(node2, bucketStorePath, - Messages.BucketStoreMessages.UpdateRemoteBuckets.class); + Address node1Address = node1.provider().getDefaultAddress(); // Add rpc on node 1 registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef()); - registry1.tell(getAddRouteMessage(), mockBroker1.getRef()); + registry1.tell(new AddOrUpdateRoutes(addedRouteIds), mockBroker1.getRef()); // Bucket store on node2 should get a message to update its local copy of remote buckets - probe2.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS), - Messages.BucketStoreMessages.UpdateRemoteBuckets.class); + + Map buckets = retrieveBuckets(registry2, mockBroker2, node1Address); + verifyBucket(buckets.get(node1Address), addedRouteIds); // Now remove - registry1.tell(getRemoveRouteMessage(), mockBroker1.getRef()); + registry1.tell(new RemoveRoutes(addedRouteIds), mockBroker1.getRef()); - // Bucket store on node2 should get a message to update its local copy of remote buckets - probe2.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS), - Messages.BucketStoreMessages.UpdateRemoteBuckets.class); + // Bucket store on node2 should get a message to update its local copy of remote buckets. + // Wait for the bucket for node1 to be empty. + + verifyEmptyBucket(mockBroker2, registry2, node1Address); System.out.println("testRpcAddRemoveInCluster ending"); } + private void verifyEmptyBucket(JavaTestKit testKit, ActorRef registry, Address address) + throws AssertionError { + Map buckets; + int nTries = 0; + while(true) { + buckets = retrieveBuckets(registry1, testKit, address); + + try { + verifyBucket(buckets.get(address), Collections.>emptyList()); + break; + } catch (AssertionError e) { + if(++nTries >= 50) { + throw e; + } + } + + Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS); + } + } + /** * Three node cluster. Register rpc on 2 nodes. Ensure 3rd gets updated. * @@ -174,76 +204,142 @@ public class RpcRegistryTest { registry3.tell(new SetLocalRouter(mockBroker3.getRef()), mockBroker3.getRef()); - // install probe on node 3 - final ActorPath bucketStorePath = new ChildActorPath(registry3.path(), "store"); - final JavaTestKit probe3 = createProbeForMessage(node3, bucketStorePath, - Messages.BucketStoreMessages.UpdateRemoteBuckets.class); - // Add rpc on node 1 + List> addedRouteIds1 = createRouteIds(); registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef()); - registry1.tell(getAddRouteMessage(), mockBroker1.getRef()); - - probe3.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS), - Messages.BucketStoreMessages.UpdateRemoteBuckets.class); + registry1.tell(new AddOrUpdateRoutes(addedRouteIds1), mockBroker1.getRef()); - // Add same rpc on node 2 + // Add rpc on node 2 + List> addedRouteIds2 = createRouteIds(); registry2.tell(new SetLocalRouter(mockBroker2.getRef()), mockBroker2.getRef()); - registry2.tell(getAddRouteMessage(), mockBroker2.getRef()); + registry2.tell(new AddOrUpdateRoutes(addedRouteIds2), mockBroker2.getRef()); + + Address node1Address = node1.provider().getDefaultAddress(); + Address node2Address = node2.provider().getDefaultAddress(); + + Map buckets = retrieveBuckets(registry3, mockBroker3, node1Address, + node2Address); - probe3.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS), - Messages.BucketStoreMessages.UpdateRemoteBuckets.class); + verifyBucket(buckets.get(node1Address), addedRouteIds1); + verifyBucket(buckets.get(node2Address), addedRouteIds2); + + Map versions = retrieveVersions(registry3, mockBroker3); + Assert.assertEquals("Version for bucket " + node1Address, buckets.get(node1Address).getVersion(), + versions.get(node1Address)); + Assert.assertEquals("Version for bucket " + node2Address, buckets.get(node2Address).getVersion(), + versions.get(node2Address)); + + RouteIdentifier routeID = addedRouteIds1.get(0); + registry3.tell(new FindRouters(routeID), mockBroker3.getRef()); + + FindRoutersReply reply = mockBroker3.expectMsgClass(Duration.create(3, TimeUnit.SECONDS), + FindRoutersReply.class); + + List> respList = reply.getRouterWithUpdateTime(); + Assert.assertEquals("getRouterWithUpdateTime size", 1, respList.size()); + + respList.get(0).first().tell("hello", ActorRef.noSender()); + mockBroker1.expectMsgEquals(Duration.create(3, TimeUnit.SECONDS), "hello"); } - private JavaTestKit createProbeForMessage(ActorSystem node, ActorPath subjectPath, final Class clazz) - throws Exception { - final JavaTestKit probe = new JavaTestKit(node); - - ConditionalProbe conditionalProbe = new ConditionalProbe(probe.getRef(), new Predicate() { - @Override - public boolean apply(@Nullable Object input) { - if (input != null) { - return clazz.equals(input.getClass()); - } else { - return false; - } + private Map retrieveVersions(ActorRef bucketStore, JavaTestKit testKit) { + bucketStore.tell(new GetBucketVersions(), testKit.getRef()); + GetBucketVersionsReply reply = testKit.expectMsgClass(Duration.create(3, TimeUnit.SECONDS), + GetBucketVersionsReply.class); + return reply.getVersions(); + } + + private void verifyBucket(Bucket bucket, List> expRouteIds) { + RoutingTable table = bucket.getData(); + Assert.assertNotNull("Bucket RoutingTable is null", table); + for(RouteIdentifier r: expRouteIds) { + if(!table.contains(r)) { + Assert.fail("RoutingTable does not contain " + r + ". Actual: " + table); } - }); + } - FiniteDuration duration = Duration.create(3, TimeUnit.SECONDS); - Timeout timeout = new Timeout(duration); - int maxTries = 30; - int i = 0; - while(true) { - ActorSelection subject = node.actorSelection(subjectPath); - Future future = Patterns.ask(subject, conditionalProbe, timeout); + Assert.assertEquals("RoutingTable size", expRouteIds.size(), table.size()); + } - try { - Await.ready(future, duration); - break; - } catch (TimeoutException | InterruptedException e) { - if(++i > maxTries) { - throw e; + private Map retrieveBuckets(ActorRef bucketStore, JavaTestKit testKit, + Address... addresses) { + int nTries = 0; + while(true) { + bucketStore.tell(new GetAllBuckets(), testKit.getRef()); + GetAllBucketsReply reply = testKit.expectMsgClass(Duration.create(3, TimeUnit.SECONDS), + GetAllBucketsReply.class); + + Map buckets = reply.getBuckets(); + boolean foundAll = true; + for(Address addr: addresses) { + Bucket bucket = buckets.get(addr); + if(bucket == null) { + foundAll = false; + break; } } - } - return probe; + if(foundAll) { + return buckets; + } - } + if(++nTries >= 50) { + Assert.fail("Missing expected buckets for addresses: " + Arrays.toString(addresses) + + ", Actual: " + buckets); + } - private AddOrUpdateRoutes getAddRouteMessage() throws URISyntaxException { - return new AddOrUpdateRoutes(createRouteIds()); + Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS); + } } - private RemoveRoutes getRemoveRouteMessage() throws URISyntaxException { - return new RemoveRoutes(createRouteIds()); + @SuppressWarnings("unchecked") + @Test + public void testAddRoutesConcurrency() throws Exception { + final JavaTestKit testKit = new JavaTestKit(node1); + + registry1.tell(new SetLocalRouter(testKit.getRef()), ActorRef.noSender()); + + final int nRoutes = 500; + final RouteIdentifier[] added = new RouteIdentifier[nRoutes]; + for(int i = 0; i < nRoutes; i++) { + final RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, + new QName(new URI("/mockrpc"), "type" + i), null); + added[i] = routeId; + + //Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); + registry1.tell(new AddOrUpdateRoutes(Arrays.>asList(routeId)), + ActorRef.noSender()); + } + + GetAllBuckets getAllBuckets = new GetAllBuckets(); + FiniteDuration duration = Duration.create(3, TimeUnit.SECONDS); + int nTries = 0; + while(true) { + registry1.tell(getAllBuckets, testKit.getRef()); + GetAllBucketsReply reply = testKit.expectMsgClass(duration, GetAllBucketsReply.class); + + Bucket localBucket = reply.getBuckets().values().iterator().next(); + RoutingTable table = localBucket.getData(); + if(table != null && table.size() == nRoutes) { + for(RouteIdentifier r: added) { + Assert.assertEquals("RoutingTable contains " + r, true, table.contains(r)); + } + + break; + } + + if(++nTries >= 50) { + Assert.fail("Expected # routes: " + nRoutes + ", Actual: " + table.size()); + } + + Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS); + } } private List> createRouteIds() throws URISyntaxException { - QName type = new QName(new URI("/mockrpc"), "mockrpc"); + QName type = new QName(new URI("/mockrpc"), "mockrpc" + routeIdCounter++); List> routeIds = new ArrayList<>(); routeIds.add(new RouteIdentifierImpl(null, type, null)); return routeIds; } - }