geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From r..@apache.org
Subject [08/51] [partial] incubator-geode git commit: WAN and CQ code drop under the Pivotal SGA
Date Wed, 20 Jan 2016 02:22:14 GMT
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelWANPropagationClientServerDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelWANPropagationClientServerDUnitTest.java
b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelWANPropagationClientServerDUnitTest.java
new file mode 100644
index 0000000..228cdad
--- /dev/null
+++ b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelWANPropagationClientServerDUnitTest.java
@@ -0,0 +1,109 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+/**
+ * 
+ */
+package com.gemstone.gemfire.internal.cache.wan.parallel;
+
+import com.gemstone.gemfire.internal.cache.wan.WANTestBase;
+
+import dunit.AsyncInvocation;
+
+/**
+ * @author skumar
+ * 
+ */
+public class ParallelWANPropagationClientServerDUnitTest extends WANTestBase {
+  private static final long serialVersionUID = 1L;
+
+  public ParallelWANPropagationClientServerDUnitTest(String name) {
+    super(name);
+  }
+
+  public void setUp() throws Exception {
+    super.setUp();
+  }
+
+  /**
+   * Normal happy scenario test case.
+   * 
+   * @throws Exception
+   */
+  public void testParallelPropagationWithClientServer() throws Exception {
+    Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+        "createFirstLocatorWithDSId", new Object[] { 1 });
+    Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+        "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+    vm2.invoke(WANTestBase.class, "createReceiverAndServer", new Object[] { nyPort });
+    vm3.invoke(WANTestBase.class, "createReceiverAndServer", new Object[] { nyPort });
+    vm2.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", null, 1, 100, isOffHeap() });
+    vm3.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", null, 1, 100, isOffHeap() });
+
+    vm4.invoke(WANTestBase.class, "createClientWithLocator", new Object[] {
+        nyPort, "localhost", testName + "_PR" });
+    vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_PR",
+      100 });
+
+    vm5.invoke(WANTestBase.class, "createServer", new Object[] { lnPort });
+    vm6.invoke(WANTestBase.class, "createServer", new Object[] { lnPort });
+    vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2, true,
+         100, 10, false, false, null, true });
+    vm6.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2, true,
+         100, 10, false, false, null, true });
+    vm5.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", "ln", 1, 100, isOffHeap() });
+    vm6.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", "ln", 1, 100, isOffHeap() });
+
+    vm7.invoke(WANTestBase.class, "createClientWithLocator", new Object[] {
+      lnPort, "localhost", testName + "_PR" });
+    
+    AsyncInvocation inv1 = vm5.invokeAsync(WANTestBase.class, "startSender", new Object[]
{ "ln" });
+    AsyncInvocation inv2 = vm6.invokeAsync(WANTestBase.class, "startSender", new Object[]
{ "ln" });
+
+    inv1.join();
+    inv2.join();
+    // before doing any puts, let the senders be running in order to ensure that
+    // not a single event will be lost
+    
+    vm5.invoke(WANTestBase.class, "waitForSenderRunningState",
+        new Object[] { "ln" });
+    vm6.invoke(WANTestBase.class, "waitForSenderRunningState",
+        new Object[] { "ln" });
+    
+    vm7.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_PR",
+        10000 });
+
+    
+    // verify all buckets drained on all sender nodes.
+    vm5.invoke(WANTestBase.class,
+        "validateParallelSenderQueueAllBucketsDrained", new Object[] { "ln" });
+    vm6.invoke(WANTestBase.class,
+        "validateParallelSenderQueueAllBucketsDrained", new Object[] { "ln" });
+
+    vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_PR", 10000 });
+    vm3.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_PR", 10000 });
+
+    vm5.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_PR", 10000 });
+    vm6.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_PR", 10000 });
+
+    vm7.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+      testName + "_PR", 10000 });
+    
+    vm4.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+      testName + "_PR", 10000 });
+
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelWANPropagationConcurrentOpsDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelWANPropagationConcurrentOpsDUnitTest.java
b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelWANPropagationConcurrentOpsDUnitTest.java
new file mode 100644
index 0000000..8eae664
--- /dev/null
+++ b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelWANPropagationConcurrentOpsDUnitTest.java
@@ -0,0 +1,282 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.internal.cache.wan.parallel;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+import com.gemstone.gemfire.internal.cache.wan.WANTestBase;
+
+import dunit.AsyncInvocation;
+
+public class ParallelWANPropagationConcurrentOpsDUnitTest extends WANTestBase {
+
+  private static final long serialVersionUID = 1L;
+  
+  public ParallelWANPropagationConcurrentOpsDUnitTest(String name) {
+    super(name);
+  }
+  
+  public void setUp() throws Exception {
+    super.setUp();
+  }
+  
+  /**
+   * Normal propagation scenario test case for a PR with only one bucket.
+   * This has been added for bug# 44284.
+   * @throws Exception
+   */
+  public void testParallelPropagationWithSingleBucketPR() throws Exception {
+    Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+        "createFirstLocatorWithDSId", new Object[] { 1 });
+    Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+        "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+    vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+    vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+    vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+    vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ 
+    vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+        true, 100, 10, false, false, null, true });
+    vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+        true, 100, 10, false, false, null, true });
+ 
+    vm4.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", "ln", 1, 1, isOffHeap() });
+    vm5.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", "ln", 1, 1, isOffHeap() });
+ 
+    vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+    vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ 
+    vm2.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", null, 1, 1, isOffHeap() });
+    vm3.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", null, 1, 1, isOffHeap() });
+
+    //before doing any puts, let the senders be running in order to ensure that
+    //not a single event will be lost
+    vm4.invoke(WANTestBase.class, "waitForSenderRunningState", new Object[] { "ln" });
+    vm5.invoke(WANTestBase.class, "waitForSenderRunningState", new Object[] { "ln" });
+    
+    //pause the senders
+    vm4.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln" });
+    vm5.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln" });
+    
+    pause(5000);
+    
+    AsyncInvocation async1 = vm4.invokeAsync(WANTestBase.class, "doPuts", new Object[] {
testName + "_PR", 700 });
+    AsyncInvocation async2 = vm4.invokeAsync(WANTestBase.class, "doPuts", new Object[] {
testName + "_PR", 1000 });
+    AsyncInvocation async3 = vm4.invokeAsync(WANTestBase.class, "doPuts", new Object[] {
testName + "_PR", 800 });
+    AsyncInvocation async4 = vm4.invokeAsync(WANTestBase.class, "doPuts", new Object[] {
testName + "_PR", 1000 });
+    
+    async1.join();
+    async2.join();
+    async3.join();
+    async4.join();
+    
+    int queueSize = (Integer) vm4.invoke(WANTestBase.class, "getQueueContentSize", new Object[]
{ "ln" });
+    assertEquals("Actual queue size is not matching with the expected", 3500, queueSize);
+    
+    //resume the senders now
+    vm4.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln" });
+    vm5.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln" });
+    
+    vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_PR", 1000 });
+  }
+
+  /**
+   * Normal propagation scenario test case for a PR with less number of buckets.
+   * Buckets have been kept to 10 for this test.
+   * This has been added for bug# 44287.
+   * @throws Exception
+   */
+  public void testParallelPropagationWithLowNumberofBuckets() throws Exception {
+    Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+        "createFirstLocatorWithDSId", new Object[] { 1 });
+    Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+        "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+    vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+    vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+    vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+    vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ 
+    vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+        true, 100, 10, false, false, null, true });
+    vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+        true, 100, 10, false, false, null, true });
+ 
+    vm4.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", "ln", 1, 10, isOffHeap() });
+    vm5.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", "ln", 1, 10, isOffHeap() });
+ 
+    vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+    vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ 
+    vm2.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", null, 1, 10, isOffHeap() });
+    vm3.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", null, 1, 10, isOffHeap() });
+
+    //before doing any puts, let the senders be running in order to ensure that
+    //not a single event will be lost
+    vm4.invoke(WANTestBase.class, "waitForSenderRunningState", new Object[] { "ln" });
+    vm5.invoke(WANTestBase.class, "waitForSenderRunningState", new Object[] { "ln" });
+    
+    AsyncInvocation async1 = vm4.invokeAsync(WANTestBase.class, "doPuts", new Object[] {
testName + "_PR", 700 });
+    AsyncInvocation async2 = vm4.invokeAsync(WANTestBase.class, "doPuts", new Object[] {
testName + "_PR", 1000 });
+    AsyncInvocation async3 = vm4.invokeAsync(WANTestBase.class, "doPuts", new Object[] {
testName + "_PR", 800 });
+    AsyncInvocation async4 = vm4.invokeAsync(WANTestBase.class, "doPuts", new Object[] {
testName + "_PR", 1000 });
+    
+    async1.join();
+    async2.join();
+    async3.join();
+    async4.join();
+    
+    vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_PR", 1000 });
+  }
+
+  public void testParalleQueueDrainInOrder_PR() throws Exception {
+    Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+        "createFirstLocatorWithDSId", new Object[] { 1 });
+    Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+        "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+    vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+    vm2.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+      testName + "_PR", null, 3, 4, isOffHeap() });
+  
+    vm2.invoke(WANTestBase.class, "addListenerOnRegion", new Object[] {testName + "_PR"});
+  
+    vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+    vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+    vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+    vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+    vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+        true, 100, 10, false, false, null, true });
+    vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+        true, 100, 10, false, false, null, true });
+    vm6.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+      true, 100, 10, false, false, null, true });
+    vm7.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+      true, 100, 10, false, false, null, true });
+
+    vm4.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", "ln", 3, 4, isOffHeap() });
+    vm5.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", "ln", 3, 4, isOffHeap() });
+    vm6.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", "ln", 3, 4, isOffHeap() });
+    vm7.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", "ln", 3, 4, isOffHeap() });
+    
+    vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+    vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+    vm6.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+    vm7.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+    vm4.invoke(WANTestBase.class, "addQueueListener", new Object[] { "ln", true});
+    vm5.invoke(WANTestBase.class, "addQueueListener", new Object[] { "ln", true});
+    vm6.invoke(WANTestBase.class, "addQueueListener", new Object[] { "ln", true});
+    vm7.invoke(WANTestBase.class, "addQueueListener", new Object[] { "ln", true});
+
+    pause(2000);
+    vm4.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln"});
+    vm5.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln"});
+    vm6.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln"});
+    vm7.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln"});
+    
+    pause(2000);
+    
+    vm6.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_PR",
+      4 });
+    vm4.invoke(WANTestBase.class, "addListenerOnBucketRegion", new Object[] {testName + "_PR",
4});
+    vm5.invoke(WANTestBase.class, "addListenerOnBucketRegion", new Object[] {testName + "_PR",
4});
+    vm6.invoke(WANTestBase.class, "addListenerOnBucketRegion", new Object[] {testName + "_PR",
4 });
+    vm7.invoke(WANTestBase.class, "addListenerOnBucketRegion", new Object[] {testName + "_PR",
4});
+    
+    vm4.invoke(WANTestBase.class, "addListenerOnQueueBucketRegion", new Object[] { "ln" ,
4});
+    vm5.invoke(WANTestBase.class, "addListenerOnQueueBucketRegion", new Object[] { "ln" ,
4});
+    vm6.invoke(WANTestBase.class, "addListenerOnQueueBucketRegion", new Object[] { "ln" ,
4 });
+    vm7.invoke(WANTestBase.class, "addListenerOnQueueBucketRegion", new Object[] { "ln" ,
4});
+    
+    vm6.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_PR",
+      1000 });
+    
+    vm6.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+      testName + "_PR", 1000 });
+    
+    HashMap vm4BRUpdates = (HashMap)vm4.invoke(WANTestBase.class, "checkBR", new Object[]
{testName + "_PR", 4});
+    HashMap vm5BRUpdates = (HashMap)vm5.invoke(WANTestBase.class, "checkBR", new Object[]
{testName + "_PR", 4});
+    HashMap vm6BRUpdates = (HashMap)vm6.invoke(WANTestBase.class, "checkBR", new Object[]
{testName + "_PR", 4});
+    HashMap vm7BRUpdates = (HashMap)vm7.invoke(WANTestBase.class, "checkBR", new Object[]
{testName + "_PR", 4});
+    
+    List b0SenderUpdates = (List)vm4BRUpdates.get("Create0");
+    List b1SenderUpdates = (List)vm4BRUpdates.get("Create1");
+    List b2SenderUpdates = (List)vm4BRUpdates.get("Create2");
+    List b3SenderUpdates = (List)vm4BRUpdates.get("Create3");
+    
+    HashMap vm4QueueBRUpdates = (HashMap)vm4.invoke(WANTestBase.class, "checkQueue_BR", new
Object[] {"ln", 4});
+    HashMap vm5QueueBRUpdates = (HashMap)vm5.invoke(WANTestBase.class, "checkQueue_BR", new
Object[] {"ln", 4});
+    HashMap vm6QueueBRUpdates = (HashMap)vm6.invoke(WANTestBase.class, "checkQueue_BR", new
Object[] {"ln", 4});
+    HashMap vm7QueueBRUpdates = (HashMap)vm7.invoke(WANTestBase.class, "checkQueue_BR", new
Object[] {"ln", 4});
+    
+    assertEquals(vm4QueueBRUpdates, vm5QueueBRUpdates);
+    assertEquals(vm4QueueBRUpdates, vm6QueueBRUpdates);
+    assertEquals(vm4QueueBRUpdates, vm7QueueBRUpdates);
+    
+    vm4.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln"});
+    vm5.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln"});
+    vm6.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln"});
+    vm7.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln"});
+    vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+      testName + "_PR", 1000 });
+    HashMap receiverUpdates = (HashMap)vm2.invoke(WANTestBase.class, "checkPR", new Object[]
{
+      testName + "_PR"});
+    List<Long> createList = (List)receiverUpdates.get("Create");
+    ArrayList<Long> b0ReceiverUpdates = new ArrayList<Long>();
+    ArrayList<Long> b1ReceiverUpdates = new ArrayList<Long>();
+    ArrayList<Long> b2ReceiverUpdates = new ArrayList<Long>();
+    ArrayList<Long> b3ReceiverUpdates = new ArrayList<Long>();
+    for (Long key : createList) {
+      long mod = key % 4;
+      if (mod == 0) {
+        b0ReceiverUpdates.add(key);
+      }
+      else if (mod == 1) {
+        b1ReceiverUpdates.add(key);
+      }
+      else if (mod == 2) {
+        b2ReceiverUpdates.add(key);
+      }
+      else if (mod == 3) {
+        b3ReceiverUpdates.add(key);
+      }
+    }
+    b0ReceiverUpdates.remove(0);
+    b1ReceiverUpdates.remove(0);
+    b2ReceiverUpdates.remove(0);
+    b3ReceiverUpdates.remove(0);
+    
+    assertEquals(b0SenderUpdates, b0ReceiverUpdates);
+    assertEquals(b1SenderUpdates, b1ReceiverUpdates);
+    assertEquals(b2SenderUpdates, b2ReceiverUpdates);
+    assertEquals(b3SenderUpdates, b3ReceiverUpdates);
+  }
+
+}


Mime
View raw message