geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kl...@apache.org
Subject [5/6] incubator-geode git commit: GEODE-1495: Changes are made to remove the cached destroyed token/events from the CQ.
Date Thu, 09 Jun 2016 02:37:18 GMT
GEODE-1495: Changes are made to remove the cached destroyed token/events from the CQ.

The CQEvents as seen by CQs are cached in order to avoid applying CQ queries on old values.

In case of a destory CQEvent, the CQEvents are marked with destroy tokens and removed from
the cache after the CQEvent is added to HAQueue.
This works fine for the CQs registered locally, but for the CQs registered on peer server,
the
CQs weren't removed from the cache, which resulted in generating wrong CQEvent for subsequent
operation.
This change removes the destroy CQevent from the cache after the CQEvent is distributed to
peer server.


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

Branch: refs/heads/feature/GEODE-837
Commit: e4994c7b3fd42b6804b909796b8589729e9861ea
Parents: afa7cc8
Author: Anil <agingade@pivotal.io>
Authored: Tue Jun 7 18:31:17 2016 -0700
Committer: Anil <agingade@pivotal.io>
Committed: Wed Jun 8 12:23:07 2016 -0700

----------------------------------------------------------------------
 .../cache/DistributedCacheOperation.java        | 52 +++++++++++++
 .../query/cq/dunit/PrCqUsingPoolDUnitTest.java  | 78 +++++++++++++++++++-
 2 files changed, 129 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e4994c7b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedCacheOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedCacheOperation.java
b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedCacheOperation.java
index 83d4c5a..f51717d 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedCacheOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedCacheOperation.java
@@ -43,6 +43,7 @@ import com.gemstone.gemfire.cache.EntryNotFoundException;
 import com.gemstone.gemfire.cache.Operation;
 import com.gemstone.gemfire.cache.RegionDestroyedException;
 import com.gemstone.gemfire.cache.query.internal.cq.CqService;
+import com.gemstone.gemfire.cache.query.internal.cq.ServerCQ;
 import com.gemstone.gemfire.distributed.internal.DM;
 import com.gemstone.gemfire.distributed.internal.DirectReplyProcessor;
 import com.gemstone.gemfire.distributed.internal.DistributionManager;
@@ -59,12 +60,14 @@ import com.gemstone.gemfire.internal.Assert;
 import com.gemstone.gemfire.internal.CopyOnWriteHashSet;
 import com.gemstone.gemfire.internal.InternalDataSerializer;
 import com.gemstone.gemfire.internal.Version;
+import com.gemstone.gemfire.internal.cache.CacheDistributionAdvisor.CacheProfile;
 import com.gemstone.gemfire.internal.cache.DistributedPutAllOperation.PutAllMessage;
 import com.gemstone.gemfire.internal.cache.EntryEventImpl.OldValueImporter;
 import com.gemstone.gemfire.internal.cache.FilterRoutingInfo.FilterInfo;
 import com.gemstone.gemfire.internal.cache.UpdateOperation.UpdateMessage;
 import com.gemstone.gemfire.internal.cache.partitioned.PartitionMessage;
 import com.gemstone.gemfire.internal.cache.persistence.PersistentMemberID;
+import com.gemstone.gemfire.internal.cache.tier.MessageType;
 import com.gemstone.gemfire.internal.cache.versions.DiskVersionTag;
 import com.gemstone.gemfire.internal.cache.versions.RegionVersionVector;
 import com.gemstone.gemfire.internal.cache.versions.VersionSource;
@@ -632,6 +635,10 @@ public abstract class DistributedCacheOperation {
         }
       }
 
+      if (region.isUsedForPartitionedRegionBucket() && filterRouting != null) {
+        removeDestroyTokensFromCqResultKeys(filterRouting);
+      }
+
     } catch (CancelException e) {
       if (logger.isDebugEnabled()) {
         logger.debug("distribution of message aborted by shutdown: {}", this);
@@ -651,6 +658,51 @@ public abstract class DistributedCacheOperation {
     }
   }
 
+
+  /**
+   * Cleanup destroyed events in CQ result cache for remote CQs.
+   * While maintaining the CQ results key caching. the destroy event
+   * keys are marked as destroyed instead of removing them, this is
+   * to take care, arrival of duplicate events. The key marked as
+   * destroyed are  removed after the event is placed in clients 
+   * HAQueue or distributed to the peers.
+   *
+   * This is similar to CacheClientNotifier.removeDestroyTokensFromCqResultKeys()
+   * where the destroyed events for local CQs are handled.
+   */
+  private void removeDestroyTokensFromCqResultKeys(FilterRoutingInfo filterRouting) {
+    for (InternalDistributedMember m : filterRouting.getMembers()) {
+      FilterInfo filterInfo = filterRouting.getFilterInfo(m);
+      if (filterInfo.getCQs() == null) {
+        continue;
+      }
+
+      CacheProfile cf = (CacheProfile) ((BucketRegion)getRegion()).getPartitionedRegion()
+          .getCacheDistributionAdvisor().getProfile(m);
+
+      if (cf == null || cf.filterProfile == null || cf.filterProfile.isLocalProfile() 
+          || cf.filterProfile.getCqMap().isEmpty()) {
+        continue;
+      }
+
+
+      for (Object value : cf.filterProfile.getCqMap().values()) {
+        ServerCQ cq = (ServerCQ)value;
+
+        for (Map.Entry<Long, Integer> e: filterInfo.getCQs().entrySet()) {
+          Long cqID = e.getKey();
+          // For the CQs satisfying the event with destroy CQEvent, remove
+          // the entry form CQ cache.
+          if (cq.getFilterID() == cqID && (e.getValue().equals(Integer.valueOf(
+              MessageType.LOCAL_DESTROY)))) {
+            cq.removeFromCqResultKeys(((EntryEventImpl)event).getKey(), true);
+          }
+        }
+      }
+    }
+  }
+
+
   /**
    * Get the adjunct receivers for a partitioned region operation
    * 

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e4994c7b/geode-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/PrCqUsingPoolDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/PrCqUsingPoolDUnitTest.java
b/geode-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/PrCqUsingPoolDUnitTest.java
index ffdd734..0af556f 100644
--- a/geode-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/PrCqUsingPoolDUnitTest.java
+++ b/geode-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/PrCqUsingPoolDUnitTest.java
@@ -317,7 +317,70 @@ public class PrCqUsingPoolDUnitTest extends CacheTestCase {
     cqHelper.closeServer(server2);
     cqHelper.closeServer(server1);
   }
-  
+
+  /**
+   * Test for registering cqs on a bridge server with local max memory zero.
+   */
+  public void testCqOnAccessorServerWithUpdatesResultingInDestroyedCQEvents() throws Exception
{
+
+    final Host host = Host.getHost(0);
+    VM server1 = host.getVM(0);
+    VM server2 = host.getVM(1);
+    VM client = host.getVM(2);
+
+    // creating an accessor vm with Bridge Server installed.
+    createServer(server1,true);
+
+    createServer(server2);
+
+    // create client
+    final int port = server1.invoke(() -> PrCqUsingPoolDUnitTest.getCacheServerPort());
+    final String host0 = NetworkUtils.getServerHostName(server1.getHost());
+
+    String poolName = "testPartitionedCqOnAccessorBridgeServer";
+    createPool(client, poolName, host0, port);
+
+    // register cq.
+    createCQ(client, poolName, "testCQEvents_0", cqs[0]);
+    cqHelper.executeCQ(client, "testCQEvents_0", false, null);
+
+    // create values
+    final int size = 10;
+    createValues(server1, regions[0], size);
+
+    // do updates that results in destroy CQEvents.
+    updateValuesToGenerateDestroyCQEvent(server1, regions[0], size);
+
+    // validate cq for expected Destroy CQ events.
+    cqHelper.validateCQ(client, "testCQEvents_0",
+        /* resultSize: */ CqQueryUsingPoolDUnitTest.noTest,
+        /* creates: */ size,
+        /* updates: */ size,
+        /* deletes; */ 0,
+        /* queryInserts: */ size,
+        /* queryUpdates: */ 0,
+        /* queryDeletes: */ size,
+        /* totalEvents: */ (size+size));
+
+    // Create values to generate Create CQ Events.
+    createValues(server1, regions[0], size);
+
+    cqHelper.validateCQ(client, "testCQEvents_0",
+        /* resultSize: */ CqQueryUsingPoolDUnitTest.noTest,
+        /* creates: */ size,
+        /* updates: */ size * 2,
+        /* deletes; */ 0,
+        /* queryInserts: */ size * 2,
+        /* queryUpdates: */ 0,
+        /* queryDeletes: */ size,
+        /* totalEvents: */ CqQueryUsingPoolDUnitTest.noTest);
+
+    cqHelper.closeClient(client);
+    cqHelper.closeServer(server2);
+    cqHelper.closeServer(server1);
+  }
+
+
   /**
    * test for registering cqs on single Bridge server hosting all the data. This
    * will generate all the events locally and should always have the old value 
@@ -1994,7 +2057,20 @@ public class PrCqUsingPoolDUnitTest extends CacheTestCase {
       }
     });
   }
+
  
+  public void updateValuesToGenerateDestroyCQEvent(VM vm, final String regionName, final
int size) {
+    vm.invoke(new CacheSerializableRunnable("Update values for region : "+regionName) {
+      public void run2() throws CacheException {
+        Region region1 = getRootRegion().getSubregion(regionName);
+        for (int i = 1; i <= size; i++) {
+          region1.put(KEY+i, new Portfolio(i * -1));
+        }
+        LogWriterUtils.getLogWriter().info("### Number of Entries in Region :" + region1.keys().size());
+      }
+    });
+  }
+
   public void createValuesPutall(VM vm, final String regionName, final int size) {
     vm.invoke(new CacheSerializableRunnable("Create values for region : "+regionName) {
       public void run2() throws CacheException {


Mime
View raw message