geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kl...@apache.org
Subject [01/20] incubator-geode git commit: GEODE-1962: Increment notQueuedConflated stat if object unresolved from offheap [Forced Update!]
Date Thu, 13 Oct 2016 17:14:38 GMT
Repository: incubator-geode
Updated Branches:
  refs/heads/feature/GEODE-1466 d26b03cf8 -> d7121b5a9 (forced update)


GEODE-1962: Increment notQueuedConflated stat if object unresolved from offheap


Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/280d2d8f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/280d2d8f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/280d2d8f

Branch: refs/heads/feature/GEODE-1466
Commit: 280d2d8f70ed2b0ae533b2c2d0728130133927ce
Parents: b2d85b4
Author: Jason Huynh <huynhja@gmail.com>
Authored: Wed Oct 5 13:32:19 2016 -0700
Committer: Jason Huynh <huynhja@gmail.com>
Committed: Tue Oct 11 11:08:28 2016 -0700

----------------------------------------------------------------------
 .../parallel/ParallelGatewaySenderQueue.java    |  35 +++--
 .../ParallelGatewaySenderQueueJUnitTest.java    | 134 ++++++++++++++++++-
 2 files changed, 149 insertions(+), 20 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/280d2d8f/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
index 1a9b126..04b6c91 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
@@ -993,7 +993,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
     return null;
   }
 
-  private boolean areLocalBucketQueueRegionsPresent() {
+  protected boolean areLocalBucketQueueRegionsPresent() {
     boolean bucketsAvailable = false;
     for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) {
       if (prQ.getDataStore().getAllLocalBucketRegions().size() > 0)
@@ -1002,10 +1002,6 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
     return false;
   }
   
-  private boolean areLocalBucketQueueRegionsPresent(PartitionedRegion prQ) {
-    return prQ.getDataStore().isLocalBucketRegionPresent();
-  }
-  
   private int pickBucketId;
   
   protected int getRandomPrimaryBucket(PartitionedRegion prQ) {
@@ -1032,8 +1028,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
       while(nTry-- > 0) {
         if(pickBucketId >= thisProcessorBuckets.size())
           pickBucketId = 0;
-        BucketRegionQueue br = (BucketRegionQueue)prQ.getDataStore()
-            .getLocalBucketById(thisProcessorBuckets.get(pickBucketId++));
+        BucketRegionQueue br = getBucketRegionQueueByBucketId(prQ, thisProcessorBuckets.get(pickBucketId++));
         if (br != null && br.isReadyForPeek()) {
           return br.getId();
         }
@@ -1044,7 +1039,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
       /*Collections.shuffle(thisProcessorBuckets);
       for (Integer bucketId : thisProcessorBuckets) {
         BucketRegionQueue br = (BucketRegionQueue)prQ.getDataStore()
-            .getLocalBucketById(bucketId);
+            .getBucketRegionQueueByBucketId(bucketId);
         
         if (br != null && br.isReadyForPeek()) {
           return br.getId();
@@ -1119,8 +1114,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
     boolean isPrimary = prQ.getRegionAdvisor().getBucketAdvisor(bucketId)
         .isPrimary();
     if (isPrimary) {
-      BucketRegionQueue brq = (BucketRegionQueue)prQ.getDataStore()
-          .getLocalBucketById(bucketId);
+      BucketRegionQueue brq = getBucketRegionQueueByBucketId(prQ, bucketId);
       // TODO : Kishor : Make sure we dont need to initalize a bucket
       // before destroying a key from it
       try {
@@ -1303,6 +1297,9 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
         if (object != null) {
           GatewaySenderEventImpl copy = object.makeHeapCopyIfOffHeap();
           if (copy == null) {
+            if (stats != null) {
+              stats.incEventsNotQueuedConflated();
+            }
             continue;
           }
           object = copy;
@@ -1314,11 +1311,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
           }
           batch.add(object);
           peekedEvents.add(object);
-          BucketRegionQueue brq = ((BucketRegionQueue)prQ
-              .getDataStore().getLocalBucketById(bId));
-          
-          //brq.doLockForPrimary(false);
-          
+
         } else {
           // If time to wait is -1 (don't wait) or time interval has elapsed
           long currentTime = System.currentTimeMillis();
@@ -1452,8 +1445,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
 
   protected Object peekAhead(PartitionedRegion prQ, int bucketId) throws CacheException {
     Object object = null;
-    BucketRegionQueue brq = ((BucketRegionQueue)prQ
-        .getDataStore().getLocalBucketById(bucketId));
+    BucketRegionQueue brq = getBucketRegionQueueByBucketId(prQ, bucketId);
 
     if (logger.isDebugEnabled()) {
       logger.debug("{}: Peekahead for the bucket {}",this, bucketId);
@@ -1475,8 +1467,13 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
     }
     return object; // OFFHEAP: ok since callers are careful to do destroys on region queue
after finished with peeked object.
   }
-  
-  
+
+  protected BucketRegionQueue getBucketRegionQueueByBucketId(final PartitionedRegion prQ,
final int bucketId) {
+    return (BucketRegionQueue)prQ
+        .getDataStore().getLocalBucketById(bucketId);
+  }
+
+
   public int localSize() {
     int size = 0;
     for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) {

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/280d2d8f/geode-core/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueueJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueueJUnitTest.java
b/geode-core/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueueJUnitTest.java
index f1d4408..4beace5 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueueJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueueJUnitTest.java
@@ -19,20 +19,35 @@ package org.apache.geode.internal.cache.wan.parallel;
 import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
 
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+import java.util.Set;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 import org.apache.geode.CancelCriterion;
 import org.apache.geode.cache.DataPolicy;
 import org.apache.geode.cache.PartitionAttributesFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.AbstractBucketRegionQueue;
+import org.apache.geode.internal.cache.BucketRegionQueue;
 import org.apache.geode.internal.cache.GemFireCacheImpl;
 import org.apache.geode.internal.cache.PartitionedRegion;
 import org.apache.geode.internal.cache.PartitionedRegionDataStore;
+import org.apache.geode.internal.cache.execute.BucketMovedException;
 import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
+import org.apache.geode.internal.cache.wan.GatewaySenderStats;
 import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderQueue.MetaRegionFactory;
 import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderQueue.ParallelGatewaySenderQueueMetaRegion;
 import org.apache.geode.test.junit.categories.UnitTest;
@@ -43,11 +58,12 @@ public class ParallelGatewaySenderQueueJUnitTest {
   private ParallelGatewaySenderQueue queue;
   private MetaRegionFactory metaRegionFactory;
   private GemFireCacheImpl cache;
+  private AbstractGatewaySender sender;
 
   @Before
   public void createParallelGatewaySenderQueue() {
     cache = mock(GemFireCacheImpl.class);
-    AbstractGatewaySender sender = mock(AbstractGatewaySender.class);
+    sender = mock(AbstractGatewaySender.class);
     CancelCriterion cancelCriterion = mock(CancelCriterion.class);
     when(sender.getCancelCriterion()).thenReturn(cancelCriterion);
     when(sender.getCache()).thenReturn(cache);
@@ -58,6 +74,54 @@ public class ParallelGatewaySenderQueueJUnitTest {
   }
 
   @Test
+  public void whenGatewayEventUnableToResolveFromOffHeapTheStatForNotQueuedConflatedShouldBeIncremented()
throws Exception {
+    GatewaySenderStats stats = mockGatewaySenderStats();
+
+    GatewaySenderEventImpl event = mock(GatewaySenderEventImpl.class);
+    when(event.makeHeapCopyIfOffHeap()).thenReturn(null);
+    GatewaySenderEventImpl eventResolvesFromOffHeap = mock(GatewaySenderEventImpl.class);
+    when(eventResolvesFromOffHeap.makeHeapCopyIfOffHeap()).thenReturn(eventResolvesFromOffHeap);
+    Queue backingList = new LinkedList();
+    backingList.add(event);
+    backingList.add(eventResolvesFromOffHeap);
+
+    BucketRegionQueue bucketRegionQueue = mockBucketRegionQueue(backingList);
+
+    TestableParallelGatewaySenderQueue queue = new TestableParallelGatewaySenderQueue(sender,
Collections.emptySet(), 0, 1, metaRegionFactory);
+    queue.setMockedAbstractBucketRegionQueue(bucketRegionQueue);
+
+    List peeked = queue.peek(1, 1000);
+    assertEquals(1, peeked.size());
+    verify(stats, times(1)).incEventsNotQueuedConflated();
+  }
+
+  private GatewaySenderStats mockGatewaySenderStats() {
+    GatewaySenderStats stats = mock(GatewaySenderStats.class);
+    when(sender.getStatistics()).thenReturn(stats);
+    return stats;
+  }
+
+  @Test
+  public void whenNullPeekedEventFromBucketRegionQueueTheStatForNotQueuedConflatedShouldBeIncremented()
throws Exception {
+    GatewaySenderStats stats = mockGatewaySenderStats();
+
+    GatewaySenderEventImpl eventResolvesFromOffHeap = mock(GatewaySenderEventImpl.class);
+    when(eventResolvesFromOffHeap.makeHeapCopyIfOffHeap()).thenReturn(eventResolvesFromOffHeap);
+    Queue backingList = new LinkedList();
+    backingList.add(null);
+    backingList.add(eventResolvesFromOffHeap);
+
+    BucketRegionQueue bucketRegionQueue = mockBucketRegionQueue(backingList);
+
+    TestableParallelGatewaySenderQueue queue = new TestableParallelGatewaySenderQueue(sender,
Collections.emptySet(), 0, 1, metaRegionFactory);
+    queue.setMockedAbstractBucketRegionQueue(bucketRegionQueue);
+
+    List peeked = queue.peek(1, 1000);
+    assertEquals(1, peeked.size());
+    verify(stats, times(1)).incEventsNotQueuedConflated();
+  }
+
+  @Test
   public void testLocalSize() throws Exception {
     ParallelGatewaySenderQueueMetaRegion mockMetaRegion = mock(ParallelGatewaySenderQueueMetaRegion.class);
     PartitionedRegionDataStore dataStore = mock(PartitionedRegionDataStore.class);
@@ -79,5 +143,73 @@ public class ParallelGatewaySenderQueueJUnitTest {
     when(region.getDataPolicy()).thenReturn(DataPolicy.PARTITION);
     return region;
   }
+  
+  private BucketRegionQueue mockBucketRegionQueue(final Queue backingList) {
+    PartitionedRegion mockBucketRegion = mockPR("bucketRegion");
+    //These next mocked return calls are for when peek is called.  It ends up checking these
on the mocked pr region
+    when(mockBucketRegion.getLocalMaxMemory()).thenReturn(100);
+    when(mockBucketRegion.size()).thenReturn(backingList.size());
+
+    BucketRegionQueue bucketRegionQueue = mock(BucketRegionQueue.class);
+    when (bucketRegionQueue.getPartitionedRegion()).thenReturn(mockBucketRegion);
+    when(bucketRegionQueue.peek()).thenAnswer((Answer) invocation -> backingList.poll());
+    return bucketRegionQueue;
+  }
+
+
+
+  private class TestableParallelGatewaySenderQueue extends ParallelGatewaySenderQueue {
+
+    private BucketRegionQueue mockedAbstractBucketRegionQueue;
+
+    public TestableParallelGatewaySenderQueue(final AbstractGatewaySender sender,
+                                              final Set<Region> userRegions,
+                                              final int idx,
+                                              final int nDispatcher) {
+      super(sender, userRegions, idx, nDispatcher);
+    }
+
+    public TestableParallelGatewaySenderQueue(final AbstractGatewaySender sender,
+                                              final Set<Region> userRegions,
+                                              final int idx,
+                                              final int nDispatcher,
+                                              final MetaRegionFactory metaRegionFactory)
{
+      super(sender, userRegions, idx, nDispatcher, metaRegionFactory);
+    }
+
+
+    public void setMockedAbstractBucketRegionQueue(BucketRegionQueue mocked) {
+      this.mockedAbstractBucketRegionQueue = mocked;
+    }
+
+    public AbstractBucketRegionQueue getBucketRegion(final PartitionedRegion prQ, final int
bucketId) {
+      return mockedAbstractBucketRegionQueue;
+    }
+
+    @Override
+    public boolean areLocalBucketQueueRegionsPresent() {
+      return true;
+    }
+
+    @Override
+    protected PartitionedRegion getRandomShadowPR() {
+      return mockedAbstractBucketRegionQueue.getPartitionedRegion();
+    }
+
+    @Override
+    protected int getRandomPrimaryBucket(PartitionedRegion pr) {
+      return 0;
+    }
+
+    @Override
+    protected BucketRegionQueue getBucketRegionQueueByBucketId(PartitionedRegion prQ, int
bucketId) {
+      return mockedAbstractBucketRegionQueue;
+    }
+
+//    @Override
+//    public int localSizeForProcessor() {
+//      return 1;
+//    }
+  }
 
 }


Mime
View raw message