You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bs...@apache.org on 2016/01/19 18:03:09 UTC
[1/3] incubator-geode git commit: Removing label from continues in
AbstractRegionMap
Repository: incubator-geode
Updated Branches:
refs/heads/develop 6e6861d14 -> eb28f6186
Removing label from continues in AbstractRegionMap
Hitesh discovered that there are no longer inner "while" loops in this
code and so we no longer need the continuation labels to break out of them.
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/78d8bb32
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/78d8bb32
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/78d8bb32
Branch: refs/heads/develop
Commit: 78d8bb328bce3f7e53e8f614a3e307d9eb5adc0b
Parents: 78d74e7
Author: Bruce Schuchardt <bs...@pivotal.io>
Authored: Tue Jan 19 08:56:08 2016 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Tue Jan 19 09:00:16 2016 -0800
----------------------------------------------------------------------
.../internal/cache/AbstractRegionMap.java | 1014 ++++++++----------
1 file changed, 464 insertions(+), 550 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/78d8bb32/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
index f4fb044..096bd0a 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
@@ -1276,574 +1276,514 @@ abstract class AbstractRegionMap implements RegionMap {
+ " is null for event " + event);
}
- //mbid: this has been added to maintain consistency between the disk region
- // and
- //and the region map after clear() has been called. This will set the
- // reference of
- //the diskSegmentRegion as a ThreadLocal so that if the diskRegionSegment
- // is later changed
- //by another thread, we can do the necessary.
-
boolean retry = true;
-// int retries = -1;
-
-RETRY_LOOP:
- while (retry) {
- retry = false;
- /* this is useful for debugging if you get a hot thread
- retries++;
- if (retries > 0) {
- owner.getCachePerfStats().incRetries();
- if (retries == 1000000) {
- logger.warn(LocalizedMessage.create(
- LocalizedStrings.AbstractRegionMap_RETRIED_1_MILLION_TIMES_FOR_ENTRY_TO_GO_AWAY_0, new Object[] { retryEntry, retryEntry.removeTrace }));
- }
- }
- */
-// boolean lruUpdateCallback = false;
-
- boolean sqlfIndexLocked = false;
- if (indexUpdater != null) {
- // take read lock for SQLF index initializations if required
- sqlfIndexLocked = indexUpdater.lockForIndexGII();
- }
- boolean opCompleted = false;
- boolean doPart3 = false;
-
- // We need to acquire the region entry while holding the lock to avoid #45620.
- // However, we also want to release the lock before distribution to prevent
- // potential deadlocks. The outer try/finally ensures that the lock will be
- // released without fail. I'm avoiding indenting just to preserve the ability
- // to track diffs since the code is fairly complex.
- boolean doUnlock = true;
- lockForCacheModification(owner, event);
- try {
+ while (retry) {
+ retry = false;
+
+ boolean opCompleted = false;
+ boolean doPart3 = false;
+
+ // We need to acquire the region entry while holding the lock to avoid #45620.
+ // However, we also want to release the lock before distribution to prevent
+ // potential deadlocks. The outer try/finally ensures that the lock will be
+ // released without fail. I'm avoiding indenting just to preserve the ability
+ // to track diffs since the code is fairly complex.
+ boolean doUnlock = true;
+ lockForCacheModification(owner, event);
+ try {
-
- RegionEntry re = getOrCreateRegionEntry(owner, event, Token.REMOVED_PHASE1, null, true, true);
- RegionEntry tombstone = null;
- boolean haveTombstone = false;
- /*
- * Execute the test hook runnable inline (not threaded) if it is not null.
- */
- if(null != testHookRunnableFor48182) {
- testHookRunnableFor48182.run();
- }
-
- try {
- if (logger.isTraceEnabled(LogMarker.LRU_TOMBSTONE_COUNT) && !(owner instanceof HARegion)) {
- logger.trace(LogMarker.LRU_TOMBSTONE_COUNT,
- "ARM.destroy() inTokenMode={}; duringRI={}; riLocalDestroy={}; withRepl={}; fromServer={}; concurrencyEnabled={}; isOriginRemote={}; isEviction={}; operation={}; re={}",
- inTokenMode, duringRI, event.isFromRILocalDestroy(), owner.dataPolicy.withReplication(), event.isFromServer(),
- owner.concurrencyChecksEnabled, event.isOriginRemote(), isEviction, event.getOperation(), re);
- }
- if (event.isFromRILocalDestroy()) {
- // for RI local-destroy we don't want to keep tombstones.
- // In order to simplify things we just set this recovery
- // flag to true to force the entry to be removed
- removeRecoveredEntry = true;
- }
- // the logic in this method is already very involved, and adding tombstone
- // permutations to (re != null) greatly complicates it. So, we check
- // for a tombstone here and, if found, pretend for a bit that the entry is null
- if (re != null && re.isTombstone() && !removeRecoveredEntry) {
- tombstone = re;
- haveTombstone = true;
- re = null;
- }
- IndexManager oqlIndexManager = owner.getIndexManager() ;
- if (re == null) {
- // we need to create an entry if in token mode or if we've received
- // a destroy from a peer or WAN gateway and we need to retain version
- // information for concurrency checks
- boolean retainForConcurrency = (!haveTombstone
- && (owner.dataPolicy.withReplication() || event.isFromServer())
- && owner.concurrencyChecksEnabled
- && (event.isOriginRemote() /* destroy received from other must create tombstone */
- || event.isFromWANAndVersioned() /* wan event must create a tombstone */
- || event.isBridgeEvent())); /* event from client must create a tombstone so client has a version # */
- if (inTokenMode
- || retainForConcurrency) {
- // removeRecoveredEntry should be false in this case
- RegionEntry newRe = getEntryFactory().createEntry(owner,
- event.getKey(),
- Token.REMOVED_PHASE1);
- // Fix for Bug #44431. We do NOT want to update the region and wait
- // later for index INIT as region.clear() can cause inconsistency if
- // happened in parallel as it also does index INIT.
- if (oqlIndexManager != null) {
- oqlIndexManager.waitForIndexInit();
+
+ RegionEntry re = getOrCreateRegionEntry(owner, event, Token.REMOVED_PHASE1, null, true, true);
+ RegionEntry tombstone = null;
+ boolean haveTombstone = false;
+ /*
+ * Execute the test hook runnable inline (not threaded) if it is not null.
+ */
+ if(null != testHookRunnableFor48182) {
+ testHookRunnableFor48182.run();
+ }
+
+ try {
+ if (logger.isTraceEnabled(LogMarker.LRU_TOMBSTONE_COUNT) && !(owner instanceof HARegion)) {
+ logger.trace(LogMarker.LRU_TOMBSTONE_COUNT,
+ "ARM.destroy() inTokenMode={}; duringRI={}; riLocalDestroy={}; withRepl={}; fromServer={}; concurrencyEnabled={}; isOriginRemote={}; isEviction={}; operation={}; re={}",
+ inTokenMode, duringRI, event.isFromRILocalDestroy(), owner.dataPolicy.withReplication(), event.isFromServer(),
+ owner.concurrencyChecksEnabled, event.isOriginRemote(), isEviction, event.getOperation(), re);
}
- try {
- synchronized (newRe) {
- RegionEntry oldRe = putEntryIfAbsent(event.getKey(), newRe);
- while (!opCompleted && oldRe != null) {
- synchronized (oldRe) {
- if (oldRe.isRemovedPhase2()) {
- oldRe = putEntryIfAbsent(event.getKey(), newRe);
- if (oldRe != null) {
- owner.getCachePerfStats().incRetries();
- }
- } else {
- event.setRegionEntry(oldRe);
-
- // Last transaction related eviction check. This should
- // prevent
- // transaction conflict (caused by eviction) when the entry
- // is being added to transaction state.
- if (isEviction) {
- if (!confirmEvictionDestroy(oldRe) || (owner.getEvictionCriteria() != null && !owner.getEvictionCriteria().doEvict(event))) {
- opCompleted = false;
- return opCompleted;
+ if (event.isFromRILocalDestroy()) {
+ // for RI local-destroy we don't want to keep tombstones.
+ // In order to simplify things we just set this recovery
+ // flag to true to force the entry to be removed
+ removeRecoveredEntry = true;
+ }
+ // the logic in this method is already very involved, and adding tombstone
+ // permutations to (re != null) greatly complicates it. So, we check
+ // for a tombstone here and, if found, pretend for a bit that the entry is null
+ if (re != null && re.isTombstone() && !removeRecoveredEntry) {
+ tombstone = re;
+ haveTombstone = true;
+ re = null;
+ }
+ IndexManager oqlIndexManager = owner.getIndexManager() ;
+ if (re == null) {
+ // we need to create an entry if in token mode or if we've received
+ // a destroy from a peer or WAN gateway and we need to retain version
+ // information for concurrency checks
+ boolean retainForConcurrency = (!haveTombstone
+ && (owner.dataPolicy.withReplication() || event.isFromServer())
+ && owner.concurrencyChecksEnabled
+ && (event.isOriginRemote() /* destroy received from other must create tombstone */
+ || event.isFromWANAndVersioned() /* wan event must create a tombstone */
+ || event.isBridgeEvent())); /* event from client must create a tombstone so client has a version # */
+ if (inTokenMode
+ || retainForConcurrency) {
+ // removeRecoveredEntry should be false in this case
+ RegionEntry newRe = getEntryFactory().createEntry(owner,
+ event.getKey(),
+ Token.REMOVED_PHASE1);
+ // Fix for Bug #44431. We do NOT want to update the region and wait
+ // later for index INIT as region.clear() can cause inconsistency if
+ // happened in parallel as it also does index INIT.
+ if (oqlIndexManager != null) {
+ oqlIndexManager.waitForIndexInit();
+ }
+ try {
+ synchronized (newRe) {
+ RegionEntry oldRe = putEntryIfAbsent(event.getKey(), newRe);
+ while (!opCompleted && oldRe != null) {
+ synchronized (oldRe) {
+ if (oldRe.isRemovedPhase2()) {
+ oldRe = putEntryIfAbsent(event.getKey(), newRe);
+ if (oldRe != null) {
+ owner.getCachePerfStats().incRetries();
+ }
+ } else {
+ event.setRegionEntry(oldRe);
+
+ // Last transaction related eviction check. This should
+ // prevent
+ // transaction conflict (caused by eviction) when the entry
+ // is being added to transaction state.
+ if (isEviction) {
+ if (!confirmEvictionDestroy(oldRe) || (owner.getEvictionCriteria() != null && !owner.getEvictionCriteria().doEvict(event))) {
+ opCompleted = false;
+ return opCompleted;
+ }
+ }
+ try {
+ //if concurrency checks are enabled, destroy will
+ //set the version tag
+ boolean destroyed = destroyEntry(oldRe, event, inTokenMode, cacheWrite, expectedOldValue, false, removeRecoveredEntry);
+ if (destroyed) {
+ if (retainForConcurrency) {
+ owner.basicDestroyBeforeRemoval(oldRe, event);
+ }
+ owner.basicDestroyPart2(oldRe, event, inTokenMode,
+ false /* conflict with clear */, duringRI, true);
+ lruEntryDestroy(oldRe);
+ doPart3 = true;
+ }
+ }
+ catch (RegionClearedException rce) {
+ // Ignore. The exception will ensure that we do not update
+ // the LRU List
+ owner.basicDestroyPart2(oldRe, event, inTokenMode,
+ true/* conflict with clear */, duringRI, true);
+ doPart3 = true;
+ } catch (ConcurrentCacheModificationException ccme) {
+ VersionTag tag = event.getVersionTag();
+ if (tag != null && tag.isTimeStampUpdated()) {
+ // Notify gateways of new time-stamp.
+ owner.notifyTimestampsToGateways(event);
+ }
+ throw ccme;
+ }
+ re = oldRe;
+ opCompleted = true;
}
- }
+ } // synchronized oldRe
+ } // while
+ if (!opCompleted) {
+ // The following try has a finally that cleans up the newRe.
+ // This is only needed if newRe was added to the map which only
+ // happens if we didn't get completed with oldRe in the above while loop.
try {
- //if concurrency checks are enabled, destroy will
- //set the version tag
- boolean destroyed = destroyEntry(oldRe, event, inTokenMode, cacheWrite, expectedOldValue, false, removeRecoveredEntry);
- if (destroyed) {
- if (retainForConcurrency) {
- owner.basicDestroyBeforeRemoval(oldRe, event);
+ re = newRe;
+ event.setRegionEntry(newRe);
+
+ try {
+ //if concurrency checks are enabled, destroy will
+ //set the version tag
+ if (isEviction) {
+ opCompleted = false;
+ return opCompleted;
}
- owner.basicDestroyPart2(oldRe, event, inTokenMode,
- false /* conflict with clear */, duringRI, true);
-// if (!oldRe.isTombstone() || isEviction) {
- lruEntryDestroy(oldRe);
-// } else { // tombstone
-// lruEntryUpdate(oldRe);
-// lruUpdateCallback = true;
-// }
+ opCompleted = destroyEntry(newRe, event, inTokenMode, cacheWrite, expectedOldValue, true, removeRecoveredEntry);
+ if (opCompleted) {
+ // This is a new entry that was created because we are in
+ // token mode or are accepting a destroy operation by adding
+ // a tombstone. There is no oldValue, so we don't need to
+ // call updateSizeOnRemove
+ // owner.recordEvent(event);
+ event.setIsRedestroyedEntry(true); // native clients need to know if the entry didn't exist
+ if (retainForConcurrency) {
+ owner.basicDestroyBeforeRemoval(oldRe, event);
+ }
+ owner.basicDestroyPart2(newRe, event, inTokenMode,
+ false /* conflict with clear */, duringRI, true);
+ doPart3 = true;
+ }
+ }
+ catch (RegionClearedException rce) {
+ // Ignore. The exception will ensure that we do not update
+ // the LRU List
+ opCompleted = true;
+ EntryLogger.logDestroy(event);
+ owner.basicDestroyPart2(newRe, event, inTokenMode, true /* conflict with clear*/, duringRI, true);
doPart3 = true;
+ } catch (ConcurrentCacheModificationException ccme) {
+ VersionTag tag = event.getVersionTag();
+ if (tag != null && tag.isTimeStampUpdated()) {
+ // Notify gateways of new time-stamp.
+ owner.notifyTimestampsToGateways(event);
+ }
+ throw ccme;
}
- }
- catch (RegionClearedException rce) {
- // region cleared implies entry is no longer there
- // so must throw exception if expecting a particular
- // old value
- // if (expectedOldValue != null) {
- // throw new EntryNotFoundException("entry not found with expected value");
- // }
-
- // Ignore. The exception will ensure that we do not update
- // the LRU List
- owner.basicDestroyPart2(oldRe, event, inTokenMode,
- true/* conflict with clear */, duringRI, true);
- doPart3 = true;
- } catch (ConcurrentCacheModificationException ccme) {
- VersionTag tag = event.getVersionTag();
- if (tag != null && tag.isTimeStampUpdated()) {
- // Notify gateways of new time-stamp.
- owner.notifyTimestampsToGateways(event);
+ // Note no need for LRU work since the entry is destroyed
+ // and will be removed when gii completes
+ } finally {
+ if (!opCompleted && !haveTombstone /* to fix bug 51583 do this for all operations */ ) {
+ removeEntry(event.getKey(), newRe, false);
+ }
+ if (!opCompleted && isEviction) {
+ removeEntry(event.getKey(), newRe, false);
}
- throw ccme;
- }
- re = oldRe;
- opCompleted = true;
- }
- } // synchronized oldRe
- } // while
- if (!opCompleted) {
- // The following try has a finally that cleans up the newRe.
- // This is only needed if newRe was added to the map which only
- // happens if we didn't get completed with oldRe in the above while loop.
- try { // bug #42228 - leaving "removed" entries in the cache
- re = newRe;
- event.setRegionEntry(newRe);
-
- try {
- //if concurrency checks are enabled, destroy will
- //set the version tag
- if (isEviction) {
- opCompleted = false;
- return opCompleted;
- }
- opCompleted = destroyEntry(newRe, event, inTokenMode, cacheWrite, expectedOldValue, true, removeRecoveredEntry);
- if (opCompleted) {
- // This is a new entry that was created because we are in
- // token mode or are accepting a destroy operation by adding
- // a tombstone. There is no oldValue, so we don't need to
- // call updateSizeOnRemove
-// owner.recordEvent(event);
- event.setIsRedestroyedEntry(true); // native clients need to know if the entry didn't exist
- if (retainForConcurrency) {
- owner.basicDestroyBeforeRemoval(oldRe, event);
}
- owner.basicDestroyPart2(newRe, event, inTokenMode,
- false /* conflict with clear */, duringRI, true);
- doPart3 = true;
- }
- }
- catch (RegionClearedException rce) {
- // region cleared implies entry is no longer there
- // so must throw exception if expecting a particular
- // old value
- // if (expectedOldValue != null) {
- // throw new EntryNotFoundException("entry not found with expected value");
- // }
-
- // Ignore. The exception will ensure that we do not update
- // the LRU List
- opCompleted = true;
- EntryLogger.logDestroy(event);
-// owner.recordEvent(event, newRe);
- owner.basicDestroyPart2(newRe, event, inTokenMode, true /* conflict with clear*/, duringRI, true);
- doPart3 = true;
- } catch (ConcurrentCacheModificationException ccme) {
- VersionTag tag = event.getVersionTag();
- if (tag != null && tag.isTimeStampUpdated()) {
- // Notify gateways of new time-stamp.
- owner.notifyTimestampsToGateways(event);
- }
- throw ccme;
+ } // !opCompleted
+ } // synchronized newRe
+ } finally {
+ if (oqlIndexManager != null) {
+ oqlIndexManager.countDownIndexUpdaters();
}
- // Note no need for LRU work since the entry is destroyed
- // and will be removed when gii completes
- } finally { // bug #42228
- if (!opCompleted && !haveTombstone /* to fix bug 51583 do this for all operations */ ) {
-
-// owner.getLogWriterI18n().warning(LocalizedStrings.DEBUG, "BRUCE: removing incomplete entry");
- removeEntry(event.getKey(), newRe, false);
+ }
+ } // inTokenMode or tombstone creation
+ else {
+ if (!isEviction || owner.concurrencyChecksEnabled) {
+ // The following ensures that there is not a concurrent operation
+ // on the entry and leaves behind a tombstone if concurrencyChecksEnabled.
+ // It fixes bug #32467 by propagating the destroy to the server even though
+ // the entry isn't in the client
+ RegionEntry newRe = haveTombstone? tombstone : getEntryFactory().createEntry(owner, event.getKey(),
+ Token.REMOVED_PHASE1);
+ synchronized(newRe) {
+ if (haveTombstone && !tombstone.isTombstone()) {
+ // we have to check this again under synchronization since it may have changed
+ retry = true;
+ //retryEntry = tombstone; // leave this in place for debugging
+ continue;
}
- if (!opCompleted && isEviction) {
- removeEntry(event.getKey(), newRe, false);
+ re = (RegionEntry)_getMap().putIfAbsent(event.getKey(), newRe);
+ if (re != null && re != tombstone) {
+ // concurrent change - try again
+ retry = true;
+ //retryEntry = tombstone; // leave this in place for debugging
+ continue;
}
- }
- } // !opCompleted
- } // synchronized newRe
- } finally {
- if (oqlIndexManager != null) {
- oqlIndexManager.countDownIndexUpdaters();
- }
- }
- } // inTokenMode or tombstone creation
- else {
- if (!isEviction || owner.concurrencyChecksEnabled) {
- // The following ensures that there is not a concurrent operation
- // on the entry and leaves behind a tombstone if concurrencyChecksEnabled.
- // It fixes bug #32467 by propagating the destroy to the server even though
- // the entry isn't in the client
- RegionEntry newRe = haveTombstone? tombstone : getEntryFactory().createEntry(owner, event.getKey(),
- Token.REMOVED_PHASE1);
- synchronized(newRe) {
- if (haveTombstone && !tombstone.isTombstone()) {
- // we have to check this again under synchronization since it may have changed
- retry = true;
- //retryEntry = tombstone; // leave this in place for debugging
- continue RETRY_LOOP;
- }
- re = (RegionEntry)_getMap().putIfAbsent(event.getKey(), newRe);
- if (re != null && re != tombstone) {
- // concurrent change - try again
- retry = true;
- //retryEntry = tombstone; // leave this in place for debugging
- continue RETRY_LOOP;
- }
- else if (!isEviction) {
- boolean throwex = false;
- EntryNotFoundException ex = null;
- try {
- if (!cacheWrite) {
- throwex = true;
- } else {
+ else if (!isEviction) {
+ boolean throwex = false;
+ EntryNotFoundException ex = null;
try {
- if (!removeRecoveredEntry) {
- throwex = !owner.bridgeWriteBeforeDestroy(event, expectedOldValue);
+ if (!cacheWrite) {
+ throwex = true;
+ } else {
+ try {
+ if (!removeRecoveredEntry) {
+ throwex = !owner.bridgeWriteBeforeDestroy(event, expectedOldValue);
+ }
+ } catch (EntryNotFoundException e) {
+ throwex = true;
+ ex = e;
+ }
}
- } catch (EntryNotFoundException e) {
- throwex = true;
- ex = e;
- }
- }
- if (throwex) {
- if (!event.isOriginRemote() && !event.getOperation().isLocal() &&
- (event.isFromBridgeAndVersioned() || // if this is a replayed client event that already has a version
- event.isFromWANAndVersioned())) { // or if this is a WAN event that has been applied in another system
- // we must distribute these since they will update the version information in peers
- if (logger.isDebugEnabled()) {
- logger.debug("ARM.destroy is allowing wan/client destroy of {} to continue", event.getKey());
+ if (throwex) {
+ if (!event.isOriginRemote() && !event.getOperation().isLocal() &&
+ (event.isFromBridgeAndVersioned() || // if this is a replayed client event that already has a version
+ event.isFromWANAndVersioned())) { // or if this is a WAN event that has been applied in another system
+ // we must distribute these since they will update the version information in peers
+ if (logger.isDebugEnabled()) {
+ logger.debug("ARM.destroy is allowing wan/client destroy of {} to continue", event.getKey());
+ }
+ throwex = false;
+ event.setIsRedestroyedEntry(true);
+ // Distribution of this op happens on re and re might me null here before
+ // distributing this destroy op.
+ if (re == null) {
+ re = newRe;
+ }
+ doPart3 = true;
+ }
}
- throwex = false;
- event.setIsRedestroyedEntry(true);
- // Distribution of this op happens on re and re might me null here before
- // distributing this destroy op.
- if (re == null) {
- re = newRe;
+ if (throwex) {
+ if (ex == null) {
+ // Fix for 48182, check cache state and/or region state before sending entry not found.
+ // this is from the server and any exceptions will propogate to the client
+ owner.checkEntryNotFound(event.getKey());
+ } else {
+ throw ex;
+ }
+ }
+ } finally {
+ // either remove the entry or leave a tombstone
+ try {
+ if (!event.isOriginRemote() && event.getVersionTag() != null && owner.concurrencyChecksEnabled) {
+ // this shouldn't fail since we just created the entry.
+ // it will either generate a tag or apply a server's version tag
+ processVersionTag(newRe, event);
+ if (doPart3) {
+ owner.generateAndSetVersionTag(event, newRe);
+ }
+ try {
+ owner.recordEvent(event);
+ newRe.makeTombstone(owner, event.getVersionTag());
+ } catch (RegionClearedException e) {
+ // that's okay - when writing a tombstone into a disk, the
+ // region has been cleared (including this tombstone)
+ }
+ opCompleted = true;
+ // lruEntryCreate(newRe);
+ } else if (!haveTombstone) {
+ try {
+ assert newRe != tombstone;
+ newRe.setValue(owner, Token.REMOVED_PHASE2);
+ removeEntry(event.getKey(), newRe, false);
+ } catch (RegionClearedException e) {
+ // that's okay - we just need to remove the new entry
+ }
+ } else if (event.getVersionTag() != null ) { // haveTombstone - update the tombstone version info
+ processVersionTag(tombstone, event);
+ if (doPart3) {
+ owner.generateAndSetVersionTag(event, newRe);
+ }
+ // This is not conflict, we need to persist the tombstone again with new version tag
+ try {
+ tombstone.setValue(owner, Token.TOMBSTONE);
+ } catch (RegionClearedException e) {
+ // that's okay - when writing a tombstone into a disk, the
+ // region has been cleared (including this tombstone)
+ }
+ owner.recordEvent(event);
+ owner.rescheduleTombstone(tombstone, event.getVersionTag());
+ owner.basicDestroyPart2(tombstone, event, inTokenMode, true /* conflict with clear*/, duringRI, true);
+ opCompleted = true;
+ }
+ } catch (ConcurrentCacheModificationException ccme) {
+ VersionTag tag = event.getVersionTag();
+ if (tag != null && tag.isTimeStampUpdated()) {
+ // Notify gateways of new time-stamp.
+ owner.notifyTimestampsToGateways(event);
+ }
+ throw ccme;
}
- doPart3 = true;
}
}
- if (throwex) {
- if (ex == null) {
- // Fix for 48182, check cache state and/or region state before sending entry not found.
- // this is from the server and any exceptions will propogate to the client
- owner.checkEntryNotFound(event.getKey());
- } else {
- throw ex;
+ } // synchronized(newRe)
+ }
+ }
+ } // no current entry
+ else { // current entry exists
+ if (oqlIndexManager != null) {
+ oqlIndexManager.waitForIndexInit();
+ }
+ try {
+ synchronized (re) {
+ // if the entry is a tombstone and the event is from a peer or a client
+ // then we allow the operation to be performed so that we can update the
+ // version stamp. Otherwise we would retain an old version stamp and may allow
+ // an operation that is older than the destroy() to be applied to the cache
+ // Bug 45170: If removeRecoveredEntry, we treat tombstone as regular entry to be deleted
+ boolean createTombstoneForConflictChecks = (owner.concurrencyChecksEnabled
+ && (event.isOriginRemote() || event.getContext() != null || removeRecoveredEntry));
+ if (!re.isRemoved() || createTombstoneForConflictChecks) {
+ if (re.isRemovedPhase2()) {
+ retry = true;
+ continue;
+ }
+ if (!event.isOriginRemote() && event.getOperation().isExpiration()) {
+ // If this expiration started locally then only do it if the RE is not being used by a tx.
+ if (re.isInUseByTransaction()) {
+ opCompleted = false;
+ return opCompleted;
}
}
- } finally {
- // either remove the entry or leave a tombstone
+ event.setRegionEntry(re);
+
+ // See comment above about eviction checks
+ if (isEviction) {
+ assert expectedOldValue == null;
+ if (!confirmEvictionDestroy(re) || (owner.getEvictionCriteria() != null && !owner.getEvictionCriteria().doEvict(event))) {
+ opCompleted = false;
+ return opCompleted;
+ }
+ }
+
+ boolean removed = false;
try {
- if (!event.isOriginRemote() && event.getVersionTag() != null && owner.concurrencyChecksEnabled) {
- // this shouldn't fail since we just created the entry.
- // it will either generate a tag or apply a server's version tag
- processVersionTag(newRe, event);
- if (doPart3) {
- owner.generateAndSetVersionTag(event, newRe);
- }
- try {
- owner.recordEvent(event);
- newRe.makeTombstone(owner, event.getVersionTag());
- } catch (RegionClearedException e) {
- // that's okay - when writing a tombstone into a disk, the
- // region has been cleared (including this tombstone)
- }
- opCompleted = true;
- // lruEntryCreate(newRe);
- } else if (!haveTombstone) {
- try {
- assert newRe != tombstone;
- newRe.setValue(owner, Token.REMOVED_PHASE2);
- removeEntry(event.getKey(), newRe, false);
- } catch (RegionClearedException e) {
- // that's okay - we just need to remove the new entry
+ opCompleted = destroyEntry(re, event, inTokenMode, cacheWrite, expectedOldValue, false, removeRecoveredEntry);
+ if (opCompleted) {
+ // It is very, very important for Partitioned Regions to keep
+ // the entry in the map until after distribution occurs so that other
+ // threads performing a create on this entry wait until the destroy
+ // distribution is finished.
+ // keeping backup copies consistent. Fix for bug 35906.
+ // -- mthomas 07/02/2007 <-- how about that date, kinda cool eh?
+ owner.basicDestroyBeforeRemoval(re, event);
+
+ // do this before basicDestroyPart2 to fix bug 31786
+ if (!inTokenMode) {
+ if ( re.getVersionStamp() == null) {
+ re.removePhase2();
+ removeEntry(event.getKey(), re, true, event, owner,
+ indexUpdater);
+ removed = true;
+ }
}
- } else if (event.getVersionTag() != null ) { // haveTombstone - update the tombstone version info
- processVersionTag(tombstone, event);
- if (doPart3) {
- owner.generateAndSetVersionTag(event, newRe);
+ if (inTokenMode && !duringRI) {
+ event.inhibitCacheListenerNotification(true);
}
- // This is not conflict, we need to persist the tombstone again with new version tag
- try {
- tombstone.setValue(owner, Token.TOMBSTONE);
- } catch (RegionClearedException e) {
- // that's okay - when writing a tombstone into a disk, the
- // region has been cleared (including this tombstone)
+ doPart3 = true;
+ owner.basicDestroyPart2(re, event, inTokenMode, false /* conflict with clear*/, duringRI, true);
+ // if (!re.isTombstone() || isEviction) {
+ lruEntryDestroy(re);
+ // } else {
+ // lruEntryUpdate(re);
+ // lruUpdateCallback = true;
+ // }
+ } else {
+ if (!inTokenMode) {
+ EntryLogger.logDestroy(event);
+ owner.recordEvent(event);
+ if (re.getVersionStamp() == null) {
+ re.removePhase2();
+ removeEntry(event.getKey(), re, true, event, owner,
+ indexUpdater);
+ lruEntryDestroy(re);
+ } else {
+ if (re.isTombstone()) {
+ // the entry is already a tombstone, but we're destroying it
+ // again, so we need to reschedule the tombstone's expiration
+ if (event.isOriginRemote()) {
+ owner.rescheduleTombstone(re, re.getVersionStamp().asVersionTag());
+ }
+ }
+ }
+ lruEntryDestroy(re);
+ opCompleted = true;
}
- owner.recordEvent(event);
- owner.rescheduleTombstone(tombstone, event.getVersionTag());
- owner.basicDestroyPart2(tombstone, event, inTokenMode, true /* conflict with clear*/, duringRI, true);
- opCompleted = true;
}
- } catch (ConcurrentCacheModificationException ccme) {
- VersionTag tag = event.getVersionTag();
- if (tag != null && tag.isTimeStampUpdated()) {
- // Notify gateways of new time-stamp.
- owner.notifyTimestampsToGateways(event);
+ }
+ catch (RegionClearedException rce) {
+ // Ignore. The exception will ensure that we do not update
+ // the LRU List
+ opCompleted = true;
+ owner.recordEvent(event);
+ if (inTokenMode && !duringRI) {
+ event.inhibitCacheListenerNotification(true);
}
- throw ccme;
+ owner.basicDestroyPart2(re, event, inTokenMode, true /*conflict with clear*/, duringRI, true);
+ doPart3 = true;
}
- }
- }
- } // synchronized(newRe)
- }
- }
- } // no current entry
- else { // current entry exists
- if (oqlIndexManager != null) {
- oqlIndexManager.waitForIndexInit();
- }
- try {
- synchronized (re) {
- // if the entry is a tombstone and the event is from a peer or a client
- // then we allow the operation to be performed so that we can update the
- // version stamp. Otherwise we would retain an old version stamp and may allow
- // an operation that is older than the destroy() to be applied to the cache
- // Bug 45170: If removeRecoveredEntry, we treat tombstone as regular entry to be deleted
- boolean createTombstoneForConflictChecks = (owner.concurrencyChecksEnabled
- && (event.isOriginRemote() || event.getContext() != null || removeRecoveredEntry));
- if (!re.isRemoved() || createTombstoneForConflictChecks) {
- if (re.isRemovedPhase2()) {
- retry = true;
- continue RETRY_LOOP;
- }
- if (!event.isOriginRemote() && event.getOperation().isExpiration()) {
- // If this expiration started locally then only do it if the RE is not being used by a tx.
- if (re.isInUseByTransaction()) {
- opCompleted = false;
- return opCompleted;
- }
- }
- event.setRegionEntry(re);
-
- // See comment above about eviction checks
- if (isEviction) {
- assert expectedOldValue == null;
- if (!confirmEvictionDestroy(re) || (owner.getEvictionCriteria() != null && !owner.getEvictionCriteria().doEvict(event))) {
- opCompleted = false;
- return opCompleted;
- }
- }
-
- boolean removed = false;
- try {
- opCompleted = destroyEntry(re, event, inTokenMode, cacheWrite, expectedOldValue, false, removeRecoveredEntry);
- if (opCompleted) {
- // It is very, very important for Partitioned Regions to keep
- // the entry in the map until after distribution occurs so that other
- // threads performing a create on this entry wait until the destroy
- // distribution is finished.
- // keeping backup copies consistent. Fix for bug 35906.
- // -- mthomas 07/02/2007 <-- how about that date, kinda cool eh?
- owner.basicDestroyBeforeRemoval(re, event);
-
- // do this before basicDestroyPart2 to fix bug 31786
- if (!inTokenMode) {
- if ( re.getVersionStamp() == null) {
- re.removePhase2();
- removeEntry(event.getKey(), re, true, event, owner,
- indexUpdater);
- removed = true;
+ finally {
+ if (re.isRemoved() && !re.isTombstone()) {
+ if (!removed) {
+ removeEntry(event.getKey(), re, true, event, owner,
+ indexUpdater);
+ }
}
}
- if (inTokenMode && !duringRI) {
- event.inhibitCacheListenerNotification(true);
+ } // !isRemoved
+ else { // already removed
+ if (owner.isHDFSReadWriteRegion() && re.isRemovedPhase2()) {
+ // For HDFS region there may be a race with eviction
+ // so retry the operation. fixes bug 49150
+ retry = true;
+ continue;
}
- doPart3 = true;
- owner.basicDestroyPart2(re, event, inTokenMode, false /* conflict with clear*/, duringRI, true);
-// if (!re.isTombstone() || isEviction) {
- lruEntryDestroy(re);
-// } else {
-// lruEntryUpdate(re);
-// lruUpdateCallback = true;
-// }
- } else {
- if (!inTokenMode) {
- EntryLogger.logDestroy(event);
- owner.recordEvent(event);
- if (re.getVersionStamp() == null) {
- re.removePhase2();
- removeEntry(event.getKey(), re, true, event, owner,
- indexUpdater);
- lruEntryDestroy(re);
- } else {
- if (re.isTombstone()) {
- // the entry is already a tombstone, but we're destroying it
- // again, so we need to reschedule the tombstone's expiration
- if (event.isOriginRemote()) {
- owner.rescheduleTombstone(re, re.getVersionStamp().asVersionTag());
- }
- }
+ if (re.isTombstone() && event.getVersionTag() != null) {
+ // if we're dealing with a tombstone and this is a remote event
+ // (e.g., from cache client update thread) we need to update
+ // the tombstone's version information
+ // TODO use destroyEntry() here
+ processVersionTag(re, event);
+ try {
+ re.makeTombstone(owner, event.getVersionTag());
+ } catch (RegionClearedException e) {
+ // that's okay - when writing a tombstone into a disk, the
+ // region has been cleared (including this tombstone)
}
- lruEntryDestroy(re);
- opCompleted = true;
}
- }
- }
- catch (RegionClearedException rce) {
- // Ignore. The exception will ensure that we do not update
- // the LRU List
- opCompleted = true;
- owner.recordEvent(event);
- if (inTokenMode && !duringRI) {
- event.inhibitCacheListenerNotification(true);
- }
- owner.basicDestroyPart2(re, event, inTokenMode, true /*conflict with clear*/, duringRI, true);
- doPart3 = true;
- }
- finally {
- if (re.isRemoved() && !re.isTombstone()) {
- if (!removed) {
- removeEntry(event.getKey(), re, true, event, owner,
- indexUpdater);
+ if (expectedOldValue != null) {
+ // if re is removed then there is no old value, so return false
+ return false;
+ }
+
+ if (!inTokenMode && !isEviction) {
+ owner.checkEntryNotFound(event.getKey());
}
}
+ } // synchronized re
+ } catch (ConcurrentCacheModificationException ccme) {
+ VersionTag tag = event.getVersionTag();
+ if (tag != null && tag.isTimeStampUpdated()) {
+ // Notify gateways of new time-stamp.
+ owner.notifyTimestampsToGateways(event);
}
- } // !isRemoved
- else { // already removed
- if (owner.isHDFSReadWriteRegion() && re.isRemovedPhase2()) {
- // For HDFS region there may be a race with eviction
- // so retry the operation. fixes bug 49150
- retry = true;
- continue RETRY_LOOP;
- }
- if (re.isTombstone() && event.getVersionTag() != null) {
- // if we're dealing with a tombstone and this is a remote event
- // (e.g., from cache client update thread) we need to update
- // the tombstone's version information
- // TODO use destroyEntry() here
- processVersionTag(re, event);
- try {
- re.makeTombstone(owner, event.getVersionTag());
- } catch (RegionClearedException e) {
- // that's okay - when writing a tombstone into a disk, the
- // region has been cleared (including this tombstone)
- }
- }
- if (expectedOldValue != null) {
- // if re is removed then there is no old value, so return false
- return false;
+ throw ccme;
+ } finally {
+ if (oqlIndexManager != null) {
+ oqlIndexManager.countDownIndexUpdaters();
}
+ }
+ // No need to call lruUpdateCallback since the only lru action
+ // we may have taken was lruEntryDestroy. This fixes bug 31759.
- if (!inTokenMode && !isEviction) {
- owner.checkEntryNotFound(event.getKey());
+ } // current entry exists
+ if(opCompleted) {
+ EntryLogger.logDestroy(event);
+ }
+ return opCompleted;
+ }
+ finally {
+ releaseCacheModificationLock(owner, event);
+ doUnlock = false;
+
+ try {
+ // If concurrency conflict is there and event contains gateway version tag then
+ // do NOT distribute.
+ if (event.isConcurrencyConflict() &&
+ (event.getVersionTag() != null && event.getVersionTag().isGatewayTag())) {
+ doPart3 = false;
+ }
+ // distribution and listener notification
+ if (doPart3) {
+ owner.basicDestroyPart3(re, event, inTokenMode, duringRI, true, expectedOldValue);
+ }
+ } finally {
+ if (opCompleted) {
+ if (re != null) {
+ owner.cancelExpiryTask(re);
+ } else if (tombstone != null) {
+ owner.cancelExpiryTask(tombstone);
}
-// if (isEviction && re.isTombstone()) {
-// owner.unscheduleTombstone(re);
-// removeTombstone(re, re.getVersionStamp().getEntryVersion(), true);
-// }
}
- } // synchronized re
- } catch (ConcurrentCacheModificationException ccme) {
- VersionTag tag = event.getVersionTag();
- if (tag != null && tag.isTimeStampUpdated()) {
- // Notify gateways of new time-stamp.
- owner.notifyTimestampsToGateways(event);
- }
- throw ccme;
- } finally {
- if (oqlIndexManager != null) {
- oqlIndexManager.countDownIndexUpdaters();
}
}
- // No need to call lruUpdateCallback since the only lru action
- // we may have taken was lruEntryDestroy. This fixes bug 31759.
- } // current entry exists
- if(opCompleted) {
- EntryLogger.logDestroy(event);
- }
- return opCompleted;
- }
- finally {
- releaseCacheModificationLock(owner, event);
- doUnlock = false;
-
- try {
- // release the SQLF index lock, if acquired
- if (sqlfIndexLocked) {
- indexUpdater.unlockForIndexGII();
- }
- // If concurrency conflict is there and event contains gateway version tag then
- // do NOT distribute.
- if (event.isConcurrencyConflict() &&
- (event.getVersionTag() != null && event.getVersionTag().isGatewayTag())) {
- doPart3 = false;
- }
- // distribution and listener notification
- if (doPart3) {
- owner.basicDestroyPart3(re, event, inTokenMode, duringRI, true, expectedOldValue);
- }
-// if (lruUpdateCallback) {
-// lruUpdateCallback();
-// }
- } finally {
- if (opCompleted) {
- if (re != null) {
- owner.cancelExpiryTask(re);
- } else if (tombstone != null) {
- owner.cancelExpiryTask(tombstone);
- }
+ } finally { // failsafe on the read lock...see comment above
+ if (doUnlock) {
+ releaseCacheModificationLock(owner, event);
}
}
- }
-
- } finally { // failsafe on the read lock...see comment above
- if (doUnlock) {
- releaseCacheModificationLock(owner, event);
- }
- }
} // retry loop
return false;
}
@@ -2339,38 +2279,12 @@ RETRY_LOOP:
// RegionEntry retryEntry = null;
// int retries = -1;
- RETRY_LOOP:
while (retry) {
retry = false;
- /* this is useful for debugging if you get a hot thread
- retries++;
- if (retries > 0) {
- owner.getCachePerfStats().incRetries();
- if (retries == 1000000) {
- logger.warn(LocalizedMessage.create(
- LocalizedStrings.AbstractRegionMap_RETRIED_1_MILLION_TIMES_FOR_ENTRY_TO_GO_AWAY_0, new Object[] { retryEntry, retryEntry.removeTrace }));
- }
- }
- */
boolean entryExisted = false;
RegionEntry re = getEntry(event.getKey());
RegionEntry tombstone = null;
boolean haveTombstone = false;
- /* this test fails when an invalidate(k,v) doesn't leave an entry in the cache:
- parReg/bridge/serialParRegHABridge.conf
- bridgeHosts=5
- bridgeThreadsPerVM=1
- bridgeVMsPerHost=1
- edgeHosts=4
- edgeThreadsPerVM=1
- edgeVMsPerHost=1
- numAccessors=1
- numEmptyClients=1
- numThinClients=1
- numVMsToStop=2
- redundantCopies=3
- hydra.Prms-randomSeed=1328320674613;
- */
if (re != null && re.isTombstone()) {
tombstone = re;
haveTombstone = true;
@@ -2387,7 +2301,7 @@ RETRY_LOOP:
// state of the tombstone has changed so we need to retry
retry = true;
//retryEntry = tombstone; // leave this in place for debugging
- continue RETRY_LOOP;
+ continue;
}
re = putEntryIfAbsent(event.getKey(), newRe);
if (re == tombstone) {
@@ -2401,7 +2315,7 @@ RETRY_LOOP:
// bug 45295: state of the tombstone has changed so we need to retry
retry = true;
//retryEntry = tombstone; // leave this in place for debugging
- continue RETRY_LOOP;
+ continue;
}
// bug #43287 - send event to server even if it's not in the client (LRU may have evicted it)
[2/3] incubator-geode git commit: Converting FilterProfile to use
CopyOnWrite collections
Posted by bs...@apache.org.
Converting FilterProfile to use CopyOnWrite collections
There was a lot of overly-complicated code in FilterProfile to implement its own
copy-on-write but we now have CopyOnWriteHashSet and CopyOnWriteHashMap. This
change-set converts FilterProfile to use those classes.
This change-set also includes refactoring of unregisterClientInterest performed in a
group refactoring session last week. It adds unit tests for unregisterClientInterest
that give 96% code coverage for that algorithm and substantially increased coverage
for registerClientInterest as well.
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/78d74e71
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/78d74e71
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/78d74e71
Branch: refs/heads/develop
Commit: 78d74e71030359884c777afc92985dd654022d5b
Parents: 6e6861d
Author: Bruce Schuchardt <bs...@pivotal.io>
Authored: Tue Jan 19 08:54:20 2016 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Tue Jan 19 09:00:16 2016 -0800
----------------------------------------------------------------------
.../gemfire/internal/cache/FilterProfile.java | 853 ++++++-------------
.../gemfire/internal/cache/LocalRegion.java | 2 +-
.../FilterProfileIntegrationJUnitTest.java | 110 +++
.../tier/sockets/FilterProfileJUnitTest.java | 422 +++++++--
.../internal/cache/tier/sockets/TestFilter.java | 58 ++
.../sanctionedDataSerializables.txt | 5 +-
6 files changed, 800 insertions(+), 650 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/78d74e71/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/FilterProfile.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/FilterProfile.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/FilterProfile.java
index 7e41f68..a8f21b9 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/FilterProfile.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/FilterProfile.java
@@ -57,6 +57,7 @@ import com.gemstone.gemfire.distributed.internal.ReplyMessage;
import com.gemstone.gemfire.distributed.internal.ReplyProcessor21;
import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
import com.gemstone.gemfire.internal.ClassLoadUtil;
+import com.gemstone.gemfire.internal.CopyOnWriteHashSet;
import com.gemstone.gemfire.internal.DataSerializableFixedID;
import com.gemstone.gemfire.internal.InternalDataSerializer;
import com.gemstone.gemfire.internal.Version;
@@ -72,6 +73,7 @@ import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
+import com.gemstone.gemfire.internal.util.concurrent.CopyOnWriteHashMap;
/**
* FilterProfile represents a distributed system member and is used for
@@ -117,61 +119,39 @@ public class FilterProfile implements DataSerializableFixedID {
}
/**
- * this variable is used to ensure that the state of all of the interest
- * variables is consistent with other threads. See
- * http://www.cs.umd.edu/~pugh/java/memoryModel/jsr-133-faq.html#volatile
- */
- @SuppressWarnings("unused")
- private volatile Object volatileBarrier = null;
-
- /**
* The keys in which clients are interested. This is a map keyed on client id,
* with a HashSet of the interested keys as the values.
- *
- * This map is never modified in place. Updaters must synchronize via
- * {@link #interestListLock}.
*/
- private Map<Object, Set> keysOfInterest;
+ private final Map<Object, Set> keysOfInterest = new CopyOnWriteHashMap<>();
- private Map<Object, Set> keysOfInterestInv;
+ private final Map<Object, Set> keysOfInterestInv = new CopyOnWriteHashMap<>();
/**
* The patterns in which clients are interested. This is a map keyed on
* client id, with a HashMap (key name to compiled pattern) as the values.
- *
- * This map is never modified in place. Updaters must synchronize via
- * {@link #interestListLock}.
*/
- private Map<Object, Map<Object, Pattern>> patternsOfInterest;
+ private final Map<Object, Map<Object, Pattern>> patternsOfInterest = new CopyOnWriteHashMap<>();
- private Map<Object, Map<Object, Pattern>> patternsOfInterestInv;
+ private final Map<Object, Map<Object, Pattern>> patternsOfInterestInv = new CopyOnWriteHashMap<>();
/**
* The filtering classes in which clients are interested. This is a map
* keyed on client id, with a HashMap (key name to {@link InterestFilter})
* as the values.
- *
- * This map is never modified in place. Updaters must synchronize via
- * {@link #interestListLock}.
*/
- private Map<Object, Map> filtersOfInterest;
+ private final Map<Object, Map> filtersOfInterest = new CopyOnWriteHashMap<>();
- private Map<Object, Map> filtersOfInterestInv;
+ private final Map<Object, Map> filtersOfInterestInv = new CopyOnWriteHashMap<>();
/**
* Set of clients that we have ALL_KEYS interest for and who want updates
*/
- private Set<Long> allKeyClients = null;
+ private final Set<Long> allKeyClients = new CopyOnWriteHashSet<>();
/**
* Set of clients that we have ALL_KEYS interest for and who want invalidations
*/
- private Set<Long> allKeyClientsInv = null;
-
- /**
- * An object used for synchronizing the interest lists
- */
- private transient final Object interestListLock = new Object();
+ private final Set<Long> allKeyClientsInv = new CopyOnWriteHashSet<>();
/**
* The region associated with this profile
@@ -187,7 +167,7 @@ public class FilterProfile implements DataSerializableFixedID {
AtomicInteger cqCount;
/** CQs that are registered on the remote node **/
- private volatile Map cqs = Collections.EMPTY_MAP;
+ private final Map cqs = new CopyOnWriteHashMap();
/* the ID of the member that this profile describes */
private DistributedMember memberID;
@@ -203,11 +183,13 @@ public class FilterProfile implements DataSerializableFixedID {
*/
IDMap cqMap;
+ private final Object interestListLock = new Object();
+
/**
* Queues the Filter Profile messages that are received during profile
* exchange.
*/
- private volatile Map<InternalDistributedMember, LinkedList<OperationMessage>> filterProfileMsgQueue = new HashMap();
+ private volatile Map<InternalDistributedMember, LinkedList<OperationMessage>> filterProfileMsgQueue = new HashMap<>();
public FilterProfile() {
cqCount = new AtomicInteger();
@@ -266,108 +248,39 @@ public class FilterProfile implements DataSerializableFixedID {
*/
public Set registerClientInterest(Object inputClientID,
Object interest, int typeOfInterest, boolean updatesAsInvalidates) {
- Set keysRegistered = null;
+ Set keysRegistered = new HashSet();
operationType opType = null;
- Long clientID;
- if (inputClientID instanceof Long) {
- clientID = (Long)inputClientID;
- } else {
- clientID = clientMap.getWireID(inputClientID);
- }
+ Long clientID = getClientIDForMaps(inputClientID);
synchronized(this.interestListLock) {
switch (typeOfInterest) {
- case InterestType.KEY: {
+ case InterestType.KEY:
opType = operationType.REGISTER_KEY;
- Set oldInterestList;
Map<Object, Set> koi = updatesAsInvalidates?
- this.getKeysOfInterestInv() : this.getKeysOfInterest();
- oldInterestList = koi.get(clientID);
- Set newInterestList = (oldInterestList == null)?
- new HashSet() : new HashSet(oldInterestList);
- newInterestList.add(interest);
- Map<Object,Set> newMap = new HashMap(koi);
- newMap.put(clientID, newInterestList);
- if (updatesAsInvalidates) {
- this.setKeysOfInterestInv(newMap);
- } else {
- this.setKeysOfInterest(newMap);
- }
- // Create a set of keys to pass to any listeners.
- // There currently is no check to see if the key already exists.
- keysRegistered = new HashSet();
- keysRegistered.add(interest);
+ getKeysOfInterestInv() : getKeysOfInterest();
+ registerKeyInMap(interest, keysRegistered, clientID, koi);
break;
- }
-
- case InterestType.REGULAR_EXPRESSION: {
- keysRegistered = new HashSet();
+ case InterestType.REGULAR_EXPRESSION:
opType = operationType.REGISTER_PATTERN;
if (((String)interest).equals(".*")) {
- // ALL_KEYS
- Set akc = updatesAsInvalidates? this.getAllKeyClientsInv() : this.getAllKeyClients();
- akc = new HashSet(akc);
+ Set akc = updatesAsInvalidates? getAllKeyClientsInv() : getAllKeyClients();
if (akc.add(clientID)) {
keysRegistered.add(interest);
- if (updatesAsInvalidates) {
- this.setAllKeyClientsInv(akc);
- } else {
- this.setAllKeyClients(akc);
- }
}
} else {
- Pattern pattern = Pattern.compile((String) interest);
Map<Object, Map<Object, Pattern>> pats = updatesAsInvalidates?
- this.getPatternsOfInterestInv() : this.getPatternsOfInterest();
- Map<Object, Pattern> oldInterestMap = pats.get(clientID);
- Map<Object, Pattern> newInterestMap =(oldInterestMap == null)?
- new HashMap() : new HashMap(oldInterestMap);
- Pattern oldPattern = newInterestMap.put(interest, pattern);
- if (oldPattern == null) {
- // If the pattern didn't exist, add it to the set of keys to pass to any listeners.
- keysRegistered.add(interest);
- }
- Map<Object, Map<Object, Pattern>> newMap = new HashMap(pats);
- newMap.put(clientID, newInterestMap);
- if(updatesAsInvalidates) {
- this.setPatternsOfInterestInv(newMap);
- } else {
- this.setPatternsOfInterest(newMap);
- }
+ getPatternsOfInterestInv() : getPatternsOfInterest();
+ registerPatternInMap(interest, keysRegistered, clientID, pats);
}
break;
- }
-
case InterestType.FILTER_CLASS: {
- // get instance of the filter
- Class filterClass;
- InterestFilter filter;
- try {
- filterClass = ClassLoadUtil.classFromName((String)interest);
- filter = (InterestFilter)filterClass.newInstance();
- }
- catch (ClassNotFoundException cnfe) {
- throw new RuntimeException(LocalizedStrings.CacheClientProxy_CLASS_0_NOT_FOUND_IN_CLASSPATH.toLocalizedString(interest), cnfe);
- }
- catch (Exception e) {
- throw new RuntimeException(LocalizedStrings.CacheClientProxy_CLASS_0_COULD_NOT_BE_INSTANTIATED.toLocalizedString(interest), e);
- }
opType = operationType.REGISTER_FILTER;
Map<Object, Map>filts = updatesAsInvalidates?
- this.getFiltersOfInterestInv() : this.getFiltersOfInterest();
- Map oldInterestMap = filts.get(clientID);
- Map newInterestMap = (oldInterestMap == null)?
- new HashMap() : new HashMap(oldInterestMap);
- newInterestMap.put(interest, filter);
- HashMap newMap = new HashMap(filts);
- newMap.put(clientID, newInterestMap);
- if (updatesAsInvalidates) {
- this.setFiltersOfInterestInv(newMap);
- } else {
- this.setFiltersOfInterest(newMap);
- }
+ getFiltersOfInterestInv() : getFiltersOfInterest();
+ registerFilterClassInMap(interest, clientID, filts);
break;
}
+
default:
throw new InternalGemFireError(LocalizedStrings.CacheClientProxy_UNKNOWN_INTEREST_TYPE.toLocalizedString());
} // switch
@@ -378,6 +291,55 @@ public class FilterProfile implements DataSerializableFixedID {
return keysRegistered;
}
+ private void registerFilterClassInMap(Object interest, Long clientID,
+ Map<Object, Map> filts) {
+ // get instance of the filter
+ Class filterClass;
+ InterestFilter filter;
+ try {
+ filterClass = ClassLoadUtil.classFromName((String)interest);
+ filter = (InterestFilter)filterClass.newInstance();
+ }
+ catch (ClassNotFoundException cnfe) {
+ throw new RuntimeException(LocalizedStrings.CacheClientProxy_CLASS_0_NOT_FOUND_IN_CLASSPATH.toLocalizedString(interest), cnfe);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(LocalizedStrings.CacheClientProxy_CLASS_0_COULD_NOT_BE_INSTANTIATED.toLocalizedString(interest), e);
+ }
+ Map interestMap = filts.get(clientID);
+ if (interestMap == null) {
+ interestMap = new CopyOnWriteHashMap();
+ filts.put(clientID, interestMap);
+ }
+ interestMap.put(interest, filter);
+ }
+
+ private void registerPatternInMap(Object interest, Set keysRegistered,
+ Long clientID, Map<Object, Map<Object, Pattern>> pats) {
+ Pattern pattern = Pattern.compile((String) interest);
+ Map<Object, Pattern> interestMap = pats.get(clientID);
+ if (interestMap == null) {
+ interestMap = new CopyOnWriteHashMap<Object, Pattern>();
+ pats.put(clientID, interestMap);
+ }
+ Pattern oldPattern = interestMap.put(interest, pattern);
+ if (oldPattern == null) {
+ // If the pattern didn't exist, add it to the set of keys to pass to any listeners.
+ keysRegistered.add(interest);
+ }
+ }
+
+ private void registerKeyInMap(Object interest, Set keysRegistered,
+ Long clientID, Map<Object, Set> koi) {
+ Set interestList = koi.get(clientID);
+ if (interestList == null) {
+ interestList = new CopyOnWriteHashSet();
+ koi.put(clientID, interestList);
+ }
+ interestList.add(interest);
+ keysRegistered.add(interest);
+ }
+
/**
* Unregisters a client's interest
*
@@ -403,238 +365,25 @@ public class FilterProfile implements DataSerializableFixedID {
return null;
}
}
- Set keysUnregistered = null;
+ Set keysUnregistered = new HashSet();
operationType opType = null;
synchronized (this.interestListLock) {
switch (interestType) {
case InterestType.KEY: {
opType = operationType.UNREGISTER_KEY;
- if (interest == UnregisterAllInterest.singleton()) {
- clearInterestFor(inputClientID);
- // Bruce: this code removed since clearInterestFor() is more comprehensive
-// if (this.keysOfInterest.get(clientID) != null) {
-// Map newMap = new HashMap(this.keysOfInterest);
-// Set removed = (Set)newMap.remove(clientID);
-// // If something is removed, create a set of keys to pass
-// // to any listeners
-// if (removed != null) {
-// keysUnregistered = new HashSet();
-// keysUnregistered.addAll(removed);
-// }
-// this.keysOfInterest = newMap;
-// }
-// if (this.keysOfInterestInv.get(clientID) != null) {
-// Map newMap = new HashMap(this.keysOfInterestInv);
-// Set removed = (Set)newMap.remove(clientID);
-// // If something is removed, create a set of keys to pass
-// // to any listeners
-// if (removed != null) {
-// if (keysUnregistered == null) keysUnregistered = new HashSet();
-// keysUnregistered.addAll(removed);
-// }
-// this.keysOfInterestInv = newMap;
-// }
- break;
- }
- Set oldInterestList = this.getKeysOfInterest().get(clientID);
- if (oldInterestList != null) {
- Set newInterestList = new HashSet(oldInterestList);
- boolean removed = newInterestList.remove(interest);
- if (removed) {
- keysUnregistered = new HashSet();
- keysUnregistered.add(interest);
- }
- Map newMap = new HashMap(this.getKeysOfInterest());
- if (newInterestList.size() > 0) {
- newMap.put(clientID, newInterestList);
- } else {
- newMap.remove(clientID);
- }
- this.setKeysOfInterest(newMap);
- }
- oldInterestList = this.getKeysOfInterestInv().get(clientID);
- if (oldInterestList != null) {
- Set newInterestList = new HashSet(oldInterestList);
- boolean removed = newInterestList.remove(interest);
- if (removed) {
- keysUnregistered = new HashSet();
- keysUnregistered.add(interest);
- }
- Map newMap = new HashMap(this.getKeysOfInterestInv());
- if (newInterestList.size() > 0) {
- newMap.put(clientID, newInterestList);
- } else {
- newMap.remove(clientID);
- }
- this.setKeysOfInterestInv(newMap);
- }
+ unregisterClientKeys(inputClientID, interest,
+ clientID, keysUnregistered);
break;
}
case InterestType.REGULAR_EXPRESSION: {
opType = operationType.UNREGISTER_PATTERN;
- if (interest == UnregisterAllInterest.singleton()) {
- if (this.getPatternsOfInterest().get(clientID) != null) {
- HashMap newMap = new HashMap(this.getPatternsOfInterest());
- Map removed = (Map)newMap.remove(clientID);
- if (removed != null) {
- keysUnregistered = new HashSet();
- keysUnregistered.addAll(removed.keySet());
- }
- this.setPatternsOfInterest(newMap);
- }
- if (this.getPatternsOfInterestInv().get(clientID) != null) {
- HashMap newMap = new HashMap(this.getPatternsOfInterestInv());
- Map removed = (Map)newMap.remove(clientID);
- if (removed != null) {
- if (keysUnregistered == null) keysUnregistered = new HashSet();
- keysUnregistered.addAll(removed.keySet());
- }
- this.setPatternsOfInterestInv(newMap);
- }
- Set oldSet = this.getAllKeyClients();
- if (!oldSet.isEmpty()) {
- Set newSet;
- newSet = new HashSet(oldSet);
- if (newSet.remove(clientID)) {
- if (newSet.isEmpty()) {
- newSet = null;
- }
- this.setAllKeyClients(newSet);
- if (keysUnregistered == null) {
- // keysUnregistered won't be null if somebody has registered
- // interest in a specific key, then in '.*'.
- keysUnregistered = new HashSet();
- }
- keysUnregistered.add(".*");
- }
- }
- oldSet = this.getAllKeyClientsInv();
- if (oldSet != null) {
- Set newSet;
- newSet = new HashSet(oldSet);
- if (newSet.remove(clientID)) {
- if (newSet.isEmpty()) {
- newSet = null;
- }
- this.setAllKeyClientsInv(newSet);
- if (keysUnregistered == null) {
- // keysUnregistered won't be null if somebody has registered
- // interest in a specific key, then in '.*'.
- keysUnregistered = new HashSet();
- }
- keysUnregistered.add(".*");
- }
- }
- }
- else if (((String)interest).equals(".*")) { // ALL_KEYS
- Set oldSet = this.getAllKeyClients();
- if (oldSet != null) {
- Set newSet;
- newSet = new HashSet(oldSet);
- if (newSet.remove(clientID)) {
- if (newSet.isEmpty()) {
- newSet = null;
- }
- this.setAllKeyClients(newSet);
- // Since something was removed, create a set of keys to pass to any
- // listeners
- keysUnregistered = new HashSet();
- keysUnregistered.add(interest);
- }
- }
- oldSet = this.getAllKeyClientsInv();
- if (oldSet != null) {
- Set newSet;
- newSet = new HashSet(oldSet);
- if (newSet.remove(clientID)) {
- if (newSet.isEmpty()) {
- newSet = null;
- }
- this.setAllKeyClientsInv(newSet);
- // Since something was removed, create a set of keys to pass to any
- // listeners
- keysUnregistered = new HashSet();
- keysUnregistered.add(interest);
- }
- }
- }
- else {
- Map oldInterestMap = this.getPatternsOfInterest().get(clientID);
- if (oldInterestMap != null) {
- Map newInterestMap = new HashMap(oldInterestMap);
- Object obj = newInterestMap.remove(interest);
- if (obj != null) {
- // Since something was removed, create a set of keys to pass to any
- // listeners
- keysUnregistered = new HashSet();
- keysUnregistered.add(interest);
- }
- Map newMap = new HashMap(this.getPatternsOfInterest());
- if (newInterestMap.size() > 0)
- newMap.put(clientID, newInterestMap);
- else
- newMap.remove(clientID);
- this.setPatternsOfInterest(newMap);
- }
- oldInterestMap = this.getPatternsOfInterestInv().get(clientID);
- if (oldInterestMap != null) {
- Map newInterestMap = new HashMap(oldInterestMap);
- Object obj = newInterestMap.remove(interest);
- if (obj != null) {
- // Since something was removed, create a set of keys to pass to any
- // listeners
- keysUnregistered = new HashSet();
- keysUnregistered.add(interest);
- }
- Map newMap = new HashMap(this.getPatternsOfInterestInv());
- if (newInterestMap.size() > 0)
- newMap.put(clientID, newInterestMap);
- else
- newMap.remove(clientID);
- this.setPatternsOfInterestInv(newMap);
- }
- }
+ unregisterClientPattern(interest, clientID,
+ keysUnregistered);
break;
}
case InterestType.FILTER_CLASS: {
opType = operationType.UNREGISTER_FILTER;
- if (interest == UnregisterAllInterest.singleton()) {
- if (this.getFiltersOfInterest().get(clientID) != null) {
- Map newMap = new HashMap(this.getFiltersOfInterest());
- newMap.remove(clientID);
- this.setFiltersOfInterest(newMap);
- }
- if (this.getFiltersOfInterestInv().get(clientID) != null) {
- Map newMap = new HashMap(this.getFiltersOfInterestInv());
- newMap.remove(clientID);
- this.setFiltersOfInterestInv(newMap);
- }
- break;
- }
- Map oldInterestMap = this.getFiltersOfInterest().get(clientID);
- if (oldInterestMap != null) {
- Map newInterestMap = new HashMap(oldInterestMap);
- newInterestMap.remove(interest);
- Map newMap = new HashMap(this.getFiltersOfInterest());
- if (newInterestMap.size() > 0) {
- newMap.put(clientID, newInterestMap);
- } else {
- newMap.remove(clientID);
- }
- this.setFiltersOfInterest(newMap);
- }
- oldInterestMap = this.getFiltersOfInterestInv().get(clientID);
- if (oldInterestMap != null) {
- Map newInterestMap = new HashMap(oldInterestMap);
- newInterestMap.remove(interest);
- Map newMap = new HashMap(this.getFiltersOfInterestInv());
- if (newInterestMap.size() > 0) {
- newMap.put(clientID, newInterestMap);
- } else {
- newMap.remove(clientID);
- }
- this.setFiltersOfInterestInv(newMap);
- }
+ unregisterClientFilterClass(interest, clientID);
break;
}
default:
@@ -647,6 +396,112 @@ public class FilterProfile implements DataSerializableFixedID {
return keysUnregistered;
}
+ private void unregisterClientFilterClass(Object interest, Long clientID) {
+ if (interest == UnregisterAllInterest.singleton()) {
+ if (getFiltersOfInterest().get(clientID) != null) {
+ getFiltersOfInterest().remove(clientID);
+ }
+ if (getFiltersOfInterestInv().get(clientID) != null) {
+ getFiltersOfInterestInv().remove(clientID);
+ }
+ return;
+ }
+ Map interestMap = getFiltersOfInterest().get(clientID);
+ if (interestMap != null) {
+ interestMap.remove(interest);
+ if (interestMap.isEmpty()) {
+ getFiltersOfInterest().remove(clientID);
+ }
+ }
+ interestMap = getFiltersOfInterestInv().get(clientID);
+ if (interestMap != null) {
+ interestMap.remove(interest);
+ if (interestMap.isEmpty()) {
+ this.getFiltersOfInterestInv().remove(clientID);
+ }
+ }
+ }
+
+ private void unregisterClientPattern(Object interest, Long clientID,
+ Set keysUnregistered) {
+ if (interest instanceof String && ((String)interest).equals(".*")) { // ALL_KEYS
+ unregisterAllKeys(interest, clientID, keysUnregistered);
+ return;
+ }
+ if (interest == UnregisterAllInterest.singleton()) {
+ unregisterClientIDFromMap(clientID, getPatternsOfInterest(), keysUnregistered);
+ unregisterClientIDFromMap(clientID, getPatternsOfInterestInv(), keysUnregistered);
+ if (getAllKeyClients().remove(clientID)) {
+ keysUnregistered.add(".*");
+ }
+ if (getAllKeyClientsInv().remove(clientID)) {
+ keysUnregistered.add(".*");
+ }
+ }
+ else {
+ unregisterPatternFromMap(getPatternsOfInterest(), interest, clientID, keysUnregistered);
+ unregisterPatternFromMap(getPatternsOfInterestInv(), interest, clientID, keysUnregistered);
+ }
+ }
+
+ private void unregisterPatternFromMap(Map<Object, Map<Object, Pattern>> map,
+ Object interest, Long clientID, Set keysUnregistered) {
+ Map interestMap = map.get(clientID);
+ if (interestMap != null) {
+ Object obj = interestMap.remove(interest);
+ if (obj != null) {
+ keysUnregistered.add(interest);
+ }
+ if (interestMap.isEmpty()) {
+ map.remove(clientID);
+ }
+ }
+ }
+
+ private void unregisterClientIDFromMap(Long clientID, Map interestMap, Set keysUnregistered) {
+ if (interestMap.get(clientID) != null) {
+ Map removed = (Map)interestMap.remove(clientID);
+ if (removed != null) {
+ keysUnregistered.addAll(removed.keySet());
+ }
+ }
+ }
+
+ private void unregisterAllKeys(Object interest, Long clientID,
+ Set keysUnregistered) {
+ if (getAllKeyClients().remove(clientID)) {
+ keysUnregistered.add(interest);
+ }
+ if (getAllKeyClientsInv().remove(clientID)) {
+ keysUnregistered.add(interest);
+ }
+ }
+
+ private void unregisterClientKeys(Object inputClientID, Object interest,
+ Long clientID, Set keysUnregistered) {
+ if (interest == UnregisterAllInterest.singleton()) {
+ clearInterestFor(inputClientID);
+ return;
+ }
+ unregisterKeyFromMap(getKeysOfInterest(), interest, clientID, keysUnregistered);
+ unregisterKeyFromMap(getKeysOfInterestInv(), interest, clientID, keysUnregistered);
+ return;
+ }
+
+ private void unregisterKeyFromMap(Map<Object, Set> map, Object interest, Long clientID,
+ Set keysUnregistered) {
+ Set interestList = map.get(clientID);
+ if (interestList != null) {
+ boolean removed = interestList.remove(interest);
+ if (removed) {
+ keysUnregistered.add(interest);
+ }
+ if (interestList.isEmpty()) {
+ map.remove(clientID);
+ }
+ }
+ }
+
/**
* Registers interest in a set of keys for a client
*
@@ -658,31 +513,21 @@ public class FilterProfile implements DataSerializableFixedID {
*/
public Set registerClientInterestList(Object inputClientID,
List keys, boolean updatesAsInvalidates) {
- Long clientID;
- if (inputClientID instanceof Long) {
- clientID = (Long)inputClientID;
- } else {
- clientID = clientMap.getWireID(inputClientID);
- }
+ Long clientID = getClientIDForMaps(inputClientID);
Set keysRegistered = new HashSet();
synchronized (interestListLock) {
Map<Object, Set> koi = updatesAsInvalidates?
- this.getKeysOfInterestInv() : this.getKeysOfInterest();
- Set oldInterestList = koi.get(clientID);
- Set newInterestList = (oldInterestList == null)?
- new HashSet() : new HashSet(oldInterestList);
+ getKeysOfInterestInv() : getKeysOfInterest();
+ Set interestList = koi.get(clientID);
+ if (interestList == null) {
+ interestList = new CopyOnWriteHashSet();
+ koi.put(clientID, interestList);
+ }
for (Object key: keys) {
- if (newInterestList.add(key)) {
+ if (interestList.add(key)) {
keysRegistered.add(key);
}
}
- Map newMap = new HashMap(koi);
- newMap.put(clientID, newInterestList);
- if (updatesAsInvalidates) {
- this.setKeysOfInterestInv(newMap);
- } else {
- this.setKeysOfInterest(newMap);
- }
if (this.region != null && this.isLocalProfile) {
sendProfileOperation(clientID, operationType.REGISTER_KEYS, keys, updatesAsInvalidates);
}
@@ -702,45 +547,32 @@ public class FilterProfile implements DataSerializableFixedID {
*/
public Set unregisterClientInterestList(Object inputClientID,
List keys) {
- Long clientID;
- if (inputClientID instanceof Long) {
- clientID = (Long)inputClientID;
- } else {
- clientID = clientMap.getWireID(inputClientID);
- }
+ Long clientID = getClientIDForMaps(inputClientID);
Set keysUnregistered = new HashSet();
synchronized (interestListLock) {
- Set oldInterestList = this.getKeysOfInterest().get(clientID);
- if (oldInterestList != null) {
- Set newInterestList = new HashSet(oldInterestList);
+ Set interestList = getKeysOfInterest().get(clientID);
+ if (interestList != null) {
for (Iterator i = keys.iterator(); i.hasNext();) {
Object keyOfInterest = i.next();
- if (newInterestList.remove(keyOfInterest)) {
+ if (interestList.remove(keyOfInterest)) {
keysUnregistered.add(keyOfInterest);
}
}
- Map newMap = new HashMap(this.getKeysOfInterest());
- if (newInterestList.size() > 0)
- newMap.put(clientID, newInterestList);
- else
- newMap.remove(clientID);
- this.setKeysOfInterest(newMap);
- }
- oldInterestList = this.getKeysOfInterestInv().get(clientID);
- if (oldInterestList != null) {
- Set newInterestList = new HashSet(oldInterestList);
+ if (interestList.isEmpty()) {
+ getKeysOfInterest().remove(clientID);
+ }
+ }
+ interestList = getKeysOfInterestInv().get(clientID);
+ if (interestList != null) {
for (Iterator i = keys.iterator(); i.hasNext();) {
Object keyOfInterest = i.next();
- if (newInterestList.remove(keyOfInterest)) {
+ if (interestList.remove(keyOfInterest)) {
keysUnregistered.add(keyOfInterest);
}
}
- Map newMap = new HashMap(this.getKeysOfInterestInv());
- if (newInterestList.size() > 0)
- newMap.put(clientID, newInterestList);
- else
- newMap.remove(clientID);
- this.setKeysOfInterestInv(newMap);
+ if (interestList.isEmpty()) {
+ getKeysOfInterestInv().remove(clientID);
+ }
}
if (this.region != null && this.isLocalProfile) {
sendProfileOperation(clientID, operationType.UNREGISTER_KEYS, keys, false);
@@ -750,13 +582,7 @@ public class FilterProfile implements DataSerializableFixedID {
}
public Set getKeysOfInterestFor(Object inputClientID) {
- Long clientID;
- if (inputClientID instanceof Long) {
- clientID = (Long)inputClientID;
- } else {
- clientID = clientMap.getWireID(inputClientID);
- }
- volatileBarrier();
+ Long clientID = getClientIDForMaps(inputClientID);
Set keys1 = this.getKeysOfInterest().get(clientID);
Set keys2 = this.getKeysOfInterestInv().get(clientID);
if (keys1 == null) {
@@ -774,13 +600,7 @@ public class FilterProfile implements DataSerializableFixedID {
}
public Map<String, Pattern> getPatternsOfInterestFor(Object inputClientID) {
- Long clientID;
- if (inputClientID instanceof Long) {
- clientID = (Long)inputClientID;
- } else {
- clientID = clientMap.getWireID(inputClientID);
- }
- volatileBarrier();
+ Long clientID = getClientIDForMaps(inputClientID);
Map patterns1 = this.getPatternsOfInterest().get(clientID);
Map patterns2 = this.getPatternsOfInterestInv().get(clientID);
if (patterns1 == null) {
@@ -798,53 +618,29 @@ public class FilterProfile implements DataSerializableFixedID {
}
public boolean hasKeysOfInterestFor(Object inputClientID, boolean wantInvalidations) {
- Long clientID;
- if (inputClientID instanceof Long) {
- clientID = (Long)inputClientID;
- } else {
- clientID = clientMap.getWireID(inputClientID);
- }
- volatileBarrier();
+ Long clientID = getClientIDForMaps(inputClientID);
if (wantInvalidations) {
return this.getKeysOfInterestInv().containsKey(clientID);
}
- return this.getKeysOfInterestInv().containsKey(clientID);
+ return this.getKeysOfInterest().containsKey(clientID);
}
public boolean hasAllKeysInterestFor(Object inputClientID) {
- Long clientID;
- if (inputClientID instanceof Long) {
- clientID = (Long)inputClientID;
- } else {
- clientID = clientMap.getWireID(inputClientID);
- }
- volatileBarrier();
+ Long clientID = getClientIDForMaps(inputClientID);
return hasAllKeysInterestFor(clientID, false) ||
hasAllKeysInterestFor(clientID, true);
}
public boolean hasAllKeysInterestFor(Object inputClientID, boolean wantInvalidations) {
- Long clientID;
- if (inputClientID instanceof Long) {
- clientID = (Long)inputClientID;
- } else {
- clientID = clientMap.getWireID(inputClientID);
- }
+ Long clientID = getClientIDForMaps(inputClientID);
if (wantInvalidations) {
- volatileBarrier();
return getAllKeyClientsInv().contains(clientID);
}
return getAllKeyClients().contains(clientID);
}
public boolean hasRegexInterestFor(Object inputClientID, boolean wantInvalidations) {
- Long clientID;
- if (inputClientID instanceof Long) {
- clientID = (Long)inputClientID;
- } else {
- clientID = clientMap.getWireID(inputClientID);
- }
- volatileBarrier();
+ Long clientID = getClientIDForMaps(inputClientID);
if (wantInvalidations) {
return (this.getPatternsOfInterestInv().containsKey(clientID));
}
@@ -852,17 +648,11 @@ public class FilterProfile implements DataSerializableFixedID {
}
public boolean hasFilterInterestFor(Object inputClientID, boolean wantInvalidations) {
- Long clientID;
- if (inputClientID instanceof Long) {
- clientID = (Long)inputClientID;
- } else {
- clientID = clientMap.getWireID(inputClientID);
- }
- volatileBarrier();
+ Long clientID = getClientIDForMaps(inputClientID);
if (wantInvalidations) {
return this.getFiltersOfInterestInv().containsKey(clientID);
}
- return this.getFiltersOfInterestInv().containsKey(clientID);
+ return this.getFiltersOfInterest().containsKey(clientID);
}
@@ -924,65 +714,49 @@ public class FilterProfile implements DataSerializableFixedID {
{
Set akc = this.getAllKeyClients();
if (akc.contains(clientID)) {
- akc = new HashSet(akc);
akc.remove(clientID);
- this.setAllKeyClients(akc);
}
}
{
Set akci = this.getAllKeyClientsInv();
if (akci.contains(clientID)) {
- akci = new HashSet(akci);
akci.remove(clientID);
- this.setAllKeyClientsInv(akci);
}
}
{
Map<Object, Set> keys = this.getKeysOfInterest();
if (keys.containsKey(clientID)) {
- Map newkeys = new HashMap(keys);
- newkeys.remove(clientID);
- this.setKeysOfInterest(newkeys);
+ keys.remove(clientID);
}
}
{
Map<Object, Set> keys = this.getKeysOfInterestInv();
if (keys.containsKey(clientID)) {
- Map newkeys = new HashMap(keys);
- newkeys.remove(clientID);
- this.setKeysOfInterestInv(newkeys);
+ keys.remove(clientID);
}
}
{
Map<Object, Map<Object, Pattern>> pats = this.getPatternsOfInterest();
if (pats.containsKey(clientID)) {
- Map newpats = new HashMap(pats);
- newpats.remove(clientID);
- this.setPatternsOfInterest(newpats);
+ pats.remove(clientID);
}
}
{
Map<Object, Map<Object, Pattern>> pats = this.getPatternsOfInterestInv();
if (pats.containsKey(clientID)) {
- Map newpats = new HashMap(pats);
- newpats.remove(clientID);
- this.setPatternsOfInterestInv(newpats);
+ pats.remove(clientID);
}
}
{
Map<Object, Map> filters = this.getFiltersOfInterest();
if (filters.containsKey(clientID)) {
- Map newfilts = new HashMap(filters);
- newfilts.remove(clientID);
- this.setFiltersOfInterest(newfilts);
+ filters.remove(clientID);
}
}
{
Map<Object, Map> filters = this.getFiltersOfInterestInv();
if (filters.containsKey(clientID)) {
- Map newfilts = new HashMap(filters);
- newfilts.remove(clientID);
- this.setFiltersOfInterestInv(newfilts);
+ filters.remove(clientID);
}
}
if (clientMap != null) {
@@ -1035,9 +809,7 @@ public class FilterProfile implements DataSerializableFixedID {
if (logger.isDebugEnabled()) {
logger.debug("Adding CQ {} to this members FilterProfile.", cq.getServerCqName());
}
- Map newCqs = new HashMap(this.cqs);
- newCqs.put(cq.getServerCqName(), cq);
- this.cqs = newCqs;
+ this.cqs.put(cq.getServerCqName(), cq);
this.incCqCount();
//cq.setFilterID(cqMap.getWireID(cq.getServerCqName()));
@@ -1087,9 +859,7 @@ public class FilterProfile implements DataSerializableFixedID {
logger.debug("Adding CQ to remote members FilterProfile using name: {}", serverCqName);
}
if (addToCqMap) {
- Map newCqs = new HashMap(this.cqs);
- newCqs.put(serverCqName, cq);
- this.cqs = newCqs;
+ this.cqs.put(serverCqName, cq);
}
// The region's FilterProfile is accessed through CQ reference as the
@@ -1113,9 +883,7 @@ public class FilterProfile implements DataSerializableFixedID {
serverCqName, ex.getMessage(), ex);
}
}
- Map newCqs = new HashMap(cqs);
- newCqs.remove(serverCqName);
- this.cqs = newCqs;
+ this.cqs.remove(serverCqName);
cq.getCqBaseRegion().getFilterProfile().decCqCount();
}
}
@@ -1154,9 +922,7 @@ public class FilterProfile implements DataSerializableFixedID {
public void closeCq(ServerCQ cq) {
ensureCqID(cq);
String serverCqName = cq.getServerCqName();
- Map newCqs = new HashMap(this.cqs);
- newCqs.remove(serverCqName);
- this.cqs = newCqs;
+ this.cqs.remove(serverCqName);
if (this.cqMap != null) {
this.cqMap.removeIDMapping(cq.getFilterID());
}
@@ -1309,8 +1075,6 @@ public class FilterProfile implements DataSerializableFixedID {
return null;
}
- volatileBarrier();
-
FilterRoutingInfo frInfo = null;
CqService cqService = getCqService(event.getRegion());
@@ -1342,7 +1106,6 @@ public class FilterProfile implements DataSerializableFixedID {
*/
public FilterRoutingInfo getFilterRoutingInfoPart2(FilterRoutingInfo part1Info,
CacheEvent event) {
- volatileBarrier();
FilterRoutingInfo result = part1Info;
if (localProfile.hasCacheServer) {
// bug #45520 - CQ events arriving out of order causes result set
@@ -1402,7 +1165,6 @@ public class FilterProfile implements DataSerializableFixedID {
final boolean isDebugEnabled = logger.isDebugEnabled();
if (this.region != null && this.localProfile.hasCacheServer) {
- volatileBarrier();
Set clientsInv = null;
Set clients = null;
int size = putAllData.length;
@@ -1461,7 +1223,6 @@ public class FilterProfile implements DataSerializableFixedID {
*/
public void getLocalFilterRoutingForRemoveAllOp(DistributedRemoveAllOperation op, RemoveAllEntryData[] removeAllData) {
if (this.region != null && this.localProfile.hasCacheServer) {
- volatileBarrier();
Set clientsInv = null;
Set clients = null;
int size = removeAllData.length;
@@ -1704,39 +1465,33 @@ public class FilterProfile implements DataSerializableFixedID {
}
-
-
- /* DataSerializableFixedID methods ---------------------------------------- */
-
public void fromData(DataInput in) throws IOException, ClassNotFoundException {
InternalDistributedMember id = new InternalDistributedMember();
InternalDataSerializer.invokeFromData(id, in);
this.memberID = id;
- this.allKeyClients = InternalDataSerializer.readSetOfLongs(in);
- this.keysOfInterest = DataSerializer.readHashMap(in);
- this.patternsOfInterest = DataSerializer.readHashMap(in);
- this.filtersOfInterest = DataSerializer.readHashMap(in);
-
- this.allKeyClientsInv = InternalDataSerializer.readSetOfLongs(in);
- this.keysOfInterestInv = DataSerializer.readHashMap(in);
- this.patternsOfInterestInv = DataSerializer.readHashMap(in);
- this.filtersOfInterestInv = DataSerializer.readHashMap(in);
+ this.allKeyClients.addAll(InternalDataSerializer.readSetOfLongs(in));
+ this.keysOfInterest.putAll(DataSerializer.readHashMap(in));
+ this.patternsOfInterest.putAll(DataSerializer.readHashMap(in));
+ this.filtersOfInterest.putAll(DataSerializer.readHashMap(in));
+
+ this.allKeyClientsInv.addAll(InternalDataSerializer.readSetOfLongs(in));
+ this.keysOfInterestInv.putAll(DataSerializer.readHashMap(in));
+ this.patternsOfInterestInv.putAll(DataSerializer.readHashMap(in));
+ this.filtersOfInterestInv.putAll(DataSerializer.readHashMap(in));
// Read CQ Info.
int numCQs = InternalDataSerializer.readArrayLength(in);
if (numCQs > 0) {
- Map theCQs = new HashMap(numCQs);
int oldLevel = LocalRegion.setThreadInitLevelRequirement(LocalRegion.ANY_INIT); // do this before CacheFactory.getInstance for bug 33471
try {
for (int i=0; i < numCQs; i++){
String serverCqName = DataSerializer.readString(in);
ServerCQ cq = CqServiceProvider.readCq(in);
processRegisterCq(serverCqName, cq, false);
- theCQs.put(serverCqName, cq);
+ this.cqs.put(serverCqName, cq);
}
} finally {
- this.cqs = theCQs;
LocalRegion.setThreadInitLevelRequirement(oldLevel);
}
}
@@ -1751,14 +1506,14 @@ public class FilterProfile implements DataSerializableFixedID {
public void toData(DataOutput out) throws IOException {
InternalDataSerializer.invokeToData(((InternalDistributedMember)memberID), out);
InternalDataSerializer.writeSetOfLongs(this.allKeyClients, this.clientMap.hasLongID, out);
- DataSerializer.writeHashMap((HashMap)this.keysOfInterest, out);
- DataSerializer.writeHashMap((HashMap)this.patternsOfInterest, out);
- DataSerializer.writeHashMap((HashMap)this.filtersOfInterest, out);
+ DataSerializer.writeHashMap(this.keysOfInterest, out);
+ DataSerializer.writeHashMap(this.patternsOfInterest, out);
+ DataSerializer.writeHashMap(this.filtersOfInterest, out);
InternalDataSerializer.writeSetOfLongs(this.allKeyClientsInv, this.clientMap.hasLongID, out);
- DataSerializer.writeHashMap((HashMap)this.keysOfInterestInv, out);
- DataSerializer.writeHashMap((HashMap)this.patternsOfInterestInv, out);
- DataSerializer.writeHashMap((HashMap)this.filtersOfInterestInv, out);
+ DataSerializer.writeHashMap(this.keysOfInterestInv, out);
+ DataSerializer.writeHashMap(this.patternsOfInterestInv, out);
+ DataSerializer.writeHashMap(this.filtersOfInterestInv, out);
// Write CQ info.
Map theCQs = this.cqs;
@@ -1777,151 +1532,77 @@ public class FilterProfile implements DataSerializableFixedID {
* @return the keysOfInterest
*/
private Map<Object, Set> getKeysOfInterest() {
- Map<Object,Set> keysOfInterestRef = this.keysOfInterest;
- return keysOfInterestRef == null? Collections.EMPTY_MAP : keysOfInterestRef;
- }
-
- /**
- * @param keysOfInterest the keysOfInterest to set
- */
- private void setKeysOfInterest(Map keysOfInterest) {
- this.keysOfInterest = keysOfInterest;
+ return this.keysOfInterest;
}
/**
* @return the keysOfInterestInv
*/
private Map<Object, Set> getKeysOfInterestInv() {
- Map<Object, Set> keysOfInterestInvRef = this.keysOfInterestInv;
- return keysOfInterestInvRef == null? Collections.EMPTY_MAP : keysOfInterestInvRef;
- }
-
- /**
- * @param keysOfInterestInv the keysOfInterestInv to set
- */
- private void setKeysOfInterestInv(Map keysOfInterestInv) {
- this.keysOfInterestInv = keysOfInterestInv;
+ return this.keysOfInterestInv;
}
/**
* @return the patternsOfInterest
*/
private Map<Object, Map<Object, Pattern>> getPatternsOfInterest() {
- Map<Object, Map<Object, Pattern>> patternsOfInterestRef = this.patternsOfInterest;
- return patternsOfInterestRef == null? Collections.EMPTY_MAP : patternsOfInterestRef;
- }
-
- /**
- * @param patternsOfInterest the patternsOfInterest to set
- */
- private void setPatternsOfInterest(Map patternsOfInterest) {
- this.patternsOfInterest = patternsOfInterest;
+ return this.patternsOfInterest;
}
/**
* @return the patternsOfInterestInv
*/
private Map<Object, Map<Object, Pattern>> getPatternsOfInterestInv() {
- Map<Object, Map<Object, Pattern>> patternsOfInterestInvRef = this.patternsOfInterestInv;
- return patternsOfInterestInvRef == null? Collections.EMPTY_MAP : patternsOfInterestInvRef;
- }
-
- /**
- * @param patternsOfInterestInv the patternsOfInterestInv to set
- */
- private void setPatternsOfInterestInv(Map patternsOfInterestInv) {
- this.patternsOfInterestInv = patternsOfInterestInv;
+ return this.patternsOfInterestInv;
}
/**
* @return the filtersOfInterestInv
*/
Map<Object, Map> getFiltersOfInterestInv() {
- Map<Object, Map> filtersOfInterestInvRef = filtersOfInterestInv;
- return filtersOfInterestInvRef == null? Collections.EMPTY_MAP : filtersOfInterestInvRef;
+ return this.filtersOfInterestInv;
}
/**
- * @param filtersOfInterestInv the filtersOfInterestInv to set
- */
- void setFiltersOfInterestInv(Map filtersOfInterestInv) {
- this.filtersOfInterestInv = filtersOfInterestInv;
- }
-
-
-
- /**
* @return the filtersOfInterest
*/
private Map<Object, Map> getFiltersOfInterest() {
- Map<Object, Map> filtersOfInterestRef = this.filtersOfInterest;
- return filtersOfInterestRef == null? Collections.EMPTY_MAP : filtersOfInterestRef;
+ return this.filtersOfInterest;
}
- /**
- * @param filtersOfInterest the filtersOfInterest to set
- */
- private void setFiltersOfInterest(Map filtersOfInterest) {
- this.filtersOfInterest = filtersOfInterest;
- }
-
- /**
- * So far all calls to setAllKeyClients has occurred under synchronized blocks, locking on interestListLock
- * @param akc the allKeyClients to set
- */
- private void setAllKeyClients(Set akc) {
- this.allKeyClients = akc;
- if (logger.isDebugEnabled()) {
- logger.debug("{}: updated allKeyClients to {}", this, akc);
- }
- }
-
- /**
- * perform a volatile read to ensure that all state is consistent
- */
- private void volatileBarrier() {
- volatileBarrier = this.cqCount; // volatile write
- }
-
- /**
- * It is possible to do a get outside of a synch block, so it can change if another thread
- * calls setAllKeyClients. So instead we store a ref to allKeyClients and if it changes
- * we at least have the one we expected to return.
- * @return the allKeyClients
- */
private Set<Object> getAllKeyClients() {
Set allKeysRef = this.allKeyClients;
if (testHook != null) {
testHook.await();
}
- return allKeysRef == null? Collections.EMPTY_SET : allKeysRef;
+ return allKeysRef;
}
public int getAllKeyClientsSize(){
return this.getAllKeyClients().size();
}
- /**
- * @param akc the allKeyClientsInv to set
- */
- private void setAllKeyClientsInv(Set akc) {
- this.allKeyClientsInv = akc;
- }
-
- /**
- * It is possible to do a get outside of a synch block, so it can change if another thread
- * calls setAllKeyClients. So instead we store a ref to allKeyClientsInv and if it changes
- * we at least have the one we expected to return.
- * @return the allKeyClientsInv
- */
- private Set<Object> getAllKeyClientsInv() {
- Set allKeysInvRef = this.allKeyClientsInv;
- return allKeysInvRef == null? Collections.EMPTY_SET : allKeysInvRef;
+ private Set<Long> getAllKeyClientsInv() {
+ return this.allKeyClientsInv;
}
public int getAllKeyClientsInvSize(){
return this.getAllKeyClientsInv().size();
}
+
+ /**
+ * When clients are registered they are assigned a Long identifier.
+ * This method maps between the real client ID and its Long identifier.
+ */
+ private Long getClientIDForMaps(Object inputClientID) {
+ Long clientID;
+ if (inputClientID instanceof Long) {
+ clientID = (Long)inputClientID;
+ } else {
+ clientID = clientMap.getWireID(inputClientID);
+ }
+ return clientID;
+ }
@Override
public String toString() {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/78d74e71/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
index 2bc2f05..2d9f720 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
@@ -11338,7 +11338,7 @@ public class LocalRegion extends AbstractRegion
* @return this region's GemFireCache instance
*/
@Override
- public final GemFireCacheImpl getGemFireCache() {
+ public GemFireCacheImpl getGemFireCache() {
return this.cache;
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/78d74e71/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/FilterProfileIntegrationJUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/FilterProfileIntegrationJUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/FilterProfileIntegrationJUnitTest.java
new file mode 100755
index 0000000..b96b597
--- /dev/null
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/FilterProfileIntegrationJUnitTest.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.gemstone.gemfire.internal.cache.tier.sockets;
+
+import java.text.ParseException;
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import com.gemstone.gemfire.cache.AttributesFactory;
+import com.gemstone.gemfire.cache.Cache;
+import com.gemstone.gemfire.cache.DataPolicy;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.RegionAttributes;
+import com.gemstone.gemfire.cache.query.CacheUtils;
+import com.gemstone.gemfire.internal.cache.FilterProfile;
+import com.gemstone.gemfire.internal.cache.LocalRegion;
+import com.gemstone.gemfire.internal.cache.tier.InterestType;
+import com.gemstone.gemfire.test.junit.categories.IntegrationTest;
+
+@Category(IntegrationTest.class)
+public class FilterProfileIntegrationJUnitTest {
+
+ private static String regionName = "test";
+ private static int numElem = 120;
+
+ @Test
+ public void testFilterProfile() throws Exception {
+ Cache cache = CacheUtils.getCache();
+ try {
+ createLocalRegion();
+ LocalRegion region = (LocalRegion) cache.getRegion(regionName);
+ final FilterProfile filterProfile = new FilterProfile(region);
+ filterProfile.registerClientInterest("clientId", ".*",
+ InterestType.REGULAR_EXPRESSION, false);
+
+ final FilterProfileTestHook hook = new FilterProfileTestHook();
+ FilterProfile.testHook = hook;
+
+ new Thread(new Runnable() {
+ public void run() {
+ while (hook.getCount() != 1) {
+
+ }
+ filterProfile.unregisterClientInterest("clientId", ".*",
+ InterestType.REGULAR_EXPRESSION);
+
+ }
+ }).start();
+ filterProfile.hasAllKeysInterestFor("clientId");
+ } finally {
+ cache.getDistributedSystem().disconnect();
+ cache.close();
+ }
+ }
+
+ class FilterProfileTestHook implements FilterProfile.TestHook {
+
+ CountDownLatch latch = new CountDownLatch(2);
+
+ // On first time, we know the first thread will reduce count by one
+ // this allows us to start the second thread, by checking the current count
+ public void await() {
+ try {
+ latch.countDown();
+ latch.await();
+ } catch (Exception e) {
+ e.printStackTrace();
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ public long getCount() {
+ return latch.getCount();
+ }
+
+ public void release() {
+ latch.countDown();
+ }
+
+ };
+
+ /**
+ * Helper Methods
+ */
+
+ private void createLocalRegion() throws ParseException {
+ Cache cache = CacheUtils.getCache();
+ AttributesFactory attributesFactory = new AttributesFactory();
+ attributesFactory.setDataPolicy(DataPolicy.NORMAL);
+ RegionAttributes regionAttributes = attributesFactory.create();
+ Region region = cache.createRegion(regionName, regionAttributes);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/78d74e71/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/FilterProfileJUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/FilterProfileJUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/FilterProfileJUnitTest.java
index 66bb0bc..1314d42 100644
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/FilterProfileJUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/FilterProfileJUnitTest.java
@@ -16,97 +16,397 @@
*/
package com.gemstone.gemfire.internal.cache.tier.sockets;
-import java.text.ParseException;
-import java.util.concurrent.CountDownLatch;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
-import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.categories.Category;
-import static org.junit.Assert.*;
-
-import junit.framework.TestCase;
-
-import com.gemstone.gemfire.cache.AttributesFactory;
-import com.gemstone.gemfire.cache.Cache;
-import com.gemstone.gemfire.cache.DataPolicy;
-import com.gemstone.gemfire.cache.Region;
-import com.gemstone.gemfire.cache.RegionAttributes;
-import com.gemstone.gemfire.cache.query.CacheUtils;
import com.gemstone.gemfire.internal.cache.FilterProfile;
+import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.tier.InterestType;
-import com.gemstone.gemfire.test.junit.categories.IntegrationTest;
+import com.gemstone.gemfire.test.junit.categories.UnitTest;
-@Category(IntegrationTest.class)
+@Category(UnitTest.class)
public class FilterProfileJUnitTest {
- private static String regionName = "test";
- private static int numElem = 120;
+ LocalRegion mockRegion;
+ FilterProfile fprofile;
+
+ @Before
+ public void setUp() {
+ mockRegion = mock(LocalRegion.class);
+ GemFireCacheImpl mockCache = mock(GemFireCacheImpl.class);
+ when(mockCache.getCacheServers()).thenReturn(Collections.emptyList());
+ when(mockRegion.getGemFireCache()).thenReturn(mockCache);
+ fprofile = new FilterProfile(mockRegion);
+ }
+
+
+ @Test
+ public void testUnregisterKey() {
+ unregisterKey(false);
+ }
@Test
- public void testFilterProfile() throws Exception {
- Cache cache = CacheUtils.getCache();
- createLocalRegion();
- LocalRegion region = (LocalRegion) cache.getRegion(regionName);
- final FilterProfile filterProfile = new FilterProfile(region);
- filterProfile.registerClientInterest("clientId", ".*",
- InterestType.REGULAR_EXPRESSION, false);
+ public void testUnregisterKeyInv() {
+ unregisterKey(true);
+ }
+
+ private void unregisterKey(boolean inv) {
+ unregisterKey(inv, false);
+ unregisterKey(inv, true);
+ }
+
+ private void unregisterKey(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ fprofile.registerClientInterest(clientId, "Object1234", InterestType.KEY, inv);
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", "Object1234", InterestType.KEY, inv);
+ }
+ boolean isClientInterested = fprofile.hasKeysOfInterestFor(clientId, inv);
+ assertTrue(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, "Object1234", InterestType.KEY);
+ assertFalse(fprofile.hasKeysOfInterestFor(clientId, inv));
+ }
- final FilterProfileTestHook hook = new FilterProfileTestHook();
- FilterProfile.testHook = hook;
+ @Test
+ public void testUnregisterTwoKeys() {
+ unregisterTwoKeys(false);
+ }
- new Thread(new Runnable() {
- public void run() {
- while (hook.getCount() != 1) {
+ @Test
+ public void testUnregisterTwoKeysInv() {
+ unregisterTwoKeys(true);
+ }
+
+ private void unregisterTwoKeys(boolean inv) {
+ unregisterTwoKeys(inv, false);
+ unregisterTwoKeys(inv, true);
+ }
+
+ private void unregisterTwoKeys(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ fprofile.registerClientInterest(clientId, "Object1234", InterestType.KEY, inv);
+ fprofile.registerClientInterest(clientId, "Object4567", InterestType.KEY, inv);
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", "Object1234", InterestType.KEY, inv);
+ }
+ boolean isClientInterested = fprofile.hasKeysOfInterestFor(clientId, inv);
+ assertTrue(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, "Object1234", InterestType.KEY);
+ fprofile.unregisterClientInterest(clientId, "Object4567", InterestType.KEY);
+ assertFalse("still has this interest: " + fprofile.getKeysOfInterestFor(clientId),
+ fprofile.hasKeysOfInterestFor(clientId, inv));
+ }
+
+ @Test
+ public void testUnregisterAllKey() {
+ unregisterAllKey(false);
+ }
+
+ @Test
+ public void testUnregisterAllKeyInv() {
+ unregisterAllKey(true);
+ }
+
+ private void unregisterAllKey(boolean inv) {
+ unregisterAllKey(inv, false);
+ unregisterAllKey(inv, true);
+ }
+
+ private void unregisterAllKey(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ fprofile.registerClientInterest(clientId, "Object1234", InterestType.KEY, inv);
+ fprofile.registerClientInterest(clientId, "Object4567", InterestType.KEY, inv);
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", "Object1234", InterestType.KEY, inv);
+ }
+ boolean isClientInterested = fprofile.hasKeysOfInterestFor(clientId, inv);
+ assertTrue(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, UnregisterAllInterest.singleton(), InterestType.KEY);
+ assertFalse(fprofile.hasKeysOfInterestFor(clientId, inv));
+ }
+
+
+ @Test
+ public void testUnregisterRegex() {
+ unregisterRegex(false);
+ }
+
+ @Test
+ public void testUnregisterRegexInv() {
+ unregisterRegex(true);
+ }
+
+ private void unregisterRegex(boolean inv) {
+ unregisterRegex(inv, false);
+ unregisterRegex(inv, true);
+ }
+
+ private void unregisterRegex(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ fprofile.registerClientInterest(clientId, "Object.*", InterestType.REGULAR_EXPRESSION, inv);
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", "Object.*", InterestType.REGULAR_EXPRESSION, inv);
+ }
+ boolean isClientInterested = fprofile.hasRegexInterestFor(clientId, inv);
+ assertTrue(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, "Object.*", InterestType.REGULAR_EXPRESSION);
+ assertFalse(fprofile.hasRegexInterestFor(clientId, inv));
+ }
+
+ @Test
+ public void testUnregisterAllRegex() {
+ unregisterAllRegex(false);
+ }
+
+ @Test
+ public void testUnregisterAllRegexInv() {
+ unregisterAllRegex(true);
+ }
+
+ private void unregisterAllRegex(boolean inv) {
+ unregisterAllRegex(inv, false);
+ unregisterAllRegex(inv, true);
+ }
+
+ private void unregisterAllRegex(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ fprofile.registerClientInterest(clientId, ".*", InterestType.REGULAR_EXPRESSION, inv);
+ fprofile.registerClientInterest(clientId, "Object.*", InterestType.REGULAR_EXPRESSION, inv);
+ fprofile.registerClientInterest(clientId, "Key.*", InterestType.REGULAR_EXPRESSION, inv);
+ if (twoClients) {
+ String clientId2 = "client2";
+ fprofile.registerClientInterest(clientId2, ".*", InterestType.REGULAR_EXPRESSION, inv);
+ fprofile.registerClientInterest(clientId2, "Object.*", InterestType.REGULAR_EXPRESSION, inv);
+ fprofile.registerClientInterest(clientId2, "Key.*", InterestType.REGULAR_EXPRESSION, inv);
+ }
+ boolean isClientInterested = fprofile.hasRegexInterestFor(clientId, inv);
+ assertTrue(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, UnregisterAllInterest.singleton(), InterestType.REGULAR_EXPRESSION);
+ assertFalse(fprofile.hasRegexInterestFor(clientId, inv));
+ }
+
+
+ @Test
+ public void testUnregisterAllKeys() {
+ unregisterAllKeys(false);
+ }
+
+ @Test
+ public void testUnregisterAllKeysInv() {
+ unregisterAllKeys(true);
+ }
- }
- filterProfile.unregisterClientInterest("clientId", ".*",
- InterestType.REGULAR_EXPRESSION);
+ private void unregisterAllKeys(boolean inv) {
+ unregisterAllKeys(inv, false);
+ unregisterAllKeys(inv, true);
+ }
+
+ private void unregisterAllKeys(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ fprofile.registerClientInterest(clientId, ".*", InterestType.REGULAR_EXPRESSION, inv);
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", ".*", InterestType.REGULAR_EXPRESSION, inv);
+ }
+ boolean isClientInterested = fprofile.hasAllKeysInterestFor(clientId, inv);
+ assertTrue(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, ".*", InterestType.REGULAR_EXPRESSION);
+ assertFalse(fprofile.hasAllKeysInterestFor(clientId, inv));
+ }
+
+ @Test
+ public void testUnregisterFilterClass() {
+ unregisterFilterClass(false);
+ }
+
+ @Test
+ public void testUnregisterFilterClassInv() {
+ unregisterFilterClass(true);
+ }
+
+ private void unregisterFilterClass(boolean inv) {
+ unregisterFilterClass(inv, false);
+ unregisterFilterClass(inv, true);
+ }
- }
- }).start();
- filterProfile.hasAllKeysInterestFor("clientId");
+ public void unregisterFilterClass(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ fprofile.registerClientInterest(clientId, "com.gemstone.gemfire.internal.cache.tier.sockets.TestFilter", InterestType.FILTER_CLASS, inv);
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", "com.gemstone.gemfire.internal.cache.tier.sockets.TestFilter", InterestType.FILTER_CLASS, inv);
+ }
+ boolean isClientInterested = fprofile.hasFilterInterestFor(clientId, inv);
+ assertTrue(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, "com.gemstone.gemfire.internal.cache.tier.sockets.TestFilter", InterestType.FILTER_CLASS);
+ assertFalse(fprofile.hasFilterInterestFor(clientId, inv));
}
- class FilterProfileTestHook implements FilterProfile.TestHook {
+ @Test
+ public void testUnregisterAllFilterClass() {
+ unregisterAllFilterClass(false);
+ }
+
+ @Test
+ public void testUnregisterAllFilterClassInv() {
+ unregisterAllFilterClass(true);
+ }
+
+ private void unregisterAllFilterClass(boolean inv) {
+ unregisterAllFilterClass(inv, false);
+ unregisterAllFilterClass(inv, true);
+ }
- CountDownLatch latch = new CountDownLatch(2);
+ public void unregisterAllFilterClass(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ fprofile.registerClientInterest(clientId, "com.gemstone.gemfire.internal.cache.tier.sockets.TestFilter", InterestType.FILTER_CLASS, inv);
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", "com.gemstone.gemfire.internal.cache.tier.sockets.TestFilter", InterestType.FILTER_CLASS, inv);
+ }
+ boolean isClientInterested = fprofile.hasFilterInterestFor(clientId, inv);
+ assertTrue(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, UnregisterAllInterest.singleton(), InterestType.FILTER_CLASS);
+ assertFalse(fprofile.hasFilterInterestFor(clientId, inv));
+
+ }
+
+ @Test
+ public void testUnregisterRegexNotRegistered() {
+ unregisterRegexNotRegistered(false);
+ }
- // On first time, we know the first thread will reduce count by one
- // this allows us to start the second thread, by checking the current count
- public void await() {
- try {
- latch.countDown();
- latch.await();
- } catch (Exception e) {
- e.printStackTrace();
- Thread.currentThread().interrupt();
- }
+ @Test
+ public void testUnregisterRegexNotRegisteredInv() {
+ unregisterRegexNotRegistered(true);
+ }
+
+ private void unregisterRegexNotRegistered(boolean inv) {
+ unregisterRegexNotRegistered(inv, false);
+ unregisterRegexNotRegistered(inv, true);
+ }
+
+ private void unregisterRegexNotRegistered(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ fprofile.registerClientInterest(clientId, "Object.*", InterestType.REGULAR_EXPRESSION, inv);
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", "Object.*", InterestType.REGULAR_EXPRESSION, inv);
}
+ boolean isClientInterested = fprofile.hasRegexInterestFor(clientId, inv);
+ assertTrue(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, "Key.*", InterestType.REGULAR_EXPRESSION);
+ assertTrue(fprofile.hasRegexInterestFor(clientId, inv));
+ }
- public long getCount() {
- return latch.getCount();
+ @Test
+ public void testUnregisterKeyNotRegistered() {
+ unregisterKeyNotRegistered(false);
+ }
+
+ @Test
+ public void testUnregisterKeyNotRegisteredInv() {
+ unregisterKeyNotRegistered(true);
+ }
+
+ private void unregisterKeyNotRegistered(boolean inv) {
+ unregisterKeyNotRegistered(inv, false);
+ unregisterKeyNotRegistered(inv, true);
+ }
+
+ private void unregisterKeyNotRegistered(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ fprofile.registerClientInterest(clientId, "Object1234", InterestType.KEY, inv);
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", "Object1234", InterestType.KEY, inv);
}
+ boolean isClientInterested = fprofile.hasKeysOfInterestFor(clientId, inv);
+ assertTrue(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, "Object5678", InterestType.KEY);
+ assertTrue(fprofile.hasKeysOfInterestFor(clientId, inv));
+ }
+
+ @Test
+ public void testUnregisterFilterNotRegistered() {
+ unregisterFilterNotRegistered(false);
+ }
- public void release() {
- latch.countDown();
+ @Test
+ public void testUnregisterFilterNotRegisteredInv() {
+ unregisterFilterNotRegistered(true);
+ }
+
+ private void unregisterFilterNotRegistered(boolean inv) {
+ unregisterFilterNotRegistered(inv, false);
+ unregisterFilterNotRegistered(inv, true);
+ }
+
+ private void unregisterFilterNotRegistered(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ fprofile.registerClientInterest(clientId, "com.gemstone.gemfire.internal.cache.tier.sockets.TestFilter", InterestType.FILTER_CLASS, inv);
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", "com.gemstone.gemfire.internal.cache.tier.sockets.TestFilter", InterestType.FILTER_CLASS, inv);
}
+ boolean isClientInterested = fprofile.hasFilterInterestFor(clientId, inv);
+ assertTrue(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, "hmm", InterestType.FILTER_CLASS);
+ assertTrue(fprofile.hasFilterInterestFor(clientId, inv));
+ }
- };
+ @Test
+ public void testUnregisterAllKeysNotRegistered() {
+ unregisterAllKeysNotRegistered(false);
+ }
+
+ @Test
+ public void testUnregisterAllKeysNotRegisteredInv() {
+ unregisterAllKeysNotRegistered(true);
+ }
+
+ private void unregisterAllKeysNotRegistered(boolean inv) {
+ unregisterAllKeysNotRegistered(inv, false);
+ unregisterAllKeysNotRegistered(inv, true);
+ }
+
+ private void unregisterAllKeysNotRegistered(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", ".*", InterestType.REGULAR_EXPRESSION, inv);
+ }
+ boolean isClientInterested = fprofile.hasAllKeysInterestFor(clientId, inv);
+ assertFalse(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, ".*", InterestType.REGULAR_EXPRESSION);
+ assertFalse(fprofile.hasAllKeysInterestFor(clientId, inv));
+ }
+ @Test
+ public void testUnregisterAllFilterNotRegistered() {
+ unregisterAllFilterNotRegistered(false);
+ }
- /**
- * Helper Methods
- */
+ @Test
+ public void testUnregisterAllFilterNotRegisteredInv() {
+ unregisterAllFilterNotRegistered(true);
+ }
- private void createLocalRegion() throws ParseException {
- Cache cache = CacheUtils.getCache();
- AttributesFactory attributesFactory = new AttributesFactory();
- attributesFactory.setDataPolicy(DataPolicy.NORMAL);
- RegionAttributes regionAttributes = attributesFactory.create();
- Region region = cache.createRegion(regionName, regionAttributes);
+ private void unregisterAllFilterNotRegistered(boolean inv) {
+ unregisterAllFilterNotRegistered(inv, false);
+ unregisterAllFilterNotRegistered(inv, true);
}
+ private void unregisterAllFilterNotRegistered(boolean inv, boolean twoClients) {
+ String clientId = "client";
+ if (twoClients) {
+ fprofile.registerClientInterest("client2", "com.gemstone.gemfire.internal.cache.tier.sockets.TestFilter", InterestType.FILTER_CLASS, inv);
+ }
+ boolean isClientInterested = fprofile.hasFilterInterestFor(clientId, inv);
+ assertFalse(isClientInterested);
+ fprofile.unregisterClientInterest(clientId, "com.gemstone.gemfire.internal.cache.tier.sockets.TestFilter", InterestType.FILTER_CLASS);
+ assertFalse(fprofile.hasFilterInterestFor(clientId, inv));
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/78d74e71/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/TestFilter.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/TestFilter.java b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/TestFilter.java
new file mode 100755
index 0000000..01da9a3
--- /dev/null
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/TestFilter.java
@@ -0,0 +1,58 @@
+package com.gemstone.gemfire.internal.cache.tier.sockets;
+
+import com.gemstone.gemfire.internal.cache.InterestEvent;
+import com.gemstone.gemfire.internal.cache.InterestFilter;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * TestFilter is used by FilterProfileJUnitTest
+ */
+public class TestFilter implements InterestFilter {
+ public TestFilter() {
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public boolean notifyOnCreate(InterestEvent event) {
+ return true;
+ }
+
+ @Override
+ public boolean notifyOnUpdate(InterestEvent event) {
+ return true;
+ }
+
+ @Override
+ public boolean notifyOnDestroy(InterestEvent event) {
+ return false;
+ }
+
+ @Override
+ public boolean notifyOnInvalidate(InterestEvent event) {
+ return false;
+ }
+
+ @Override
+ public boolean notifyOnRegister(InterestEvent event) {
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/78d74e71/gemfire-core/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt b/gemfire-core/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt
index d3a141a..3bc7ac3 100644
--- a/gemfire-core/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt
+++ b/gemfire-core/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt
@@ -1,3 +1,4 @@
+
com/gemstone/gemfire/admin/RegionSubRegionSnapshot,2
fromData,62,2a2bb80023b500082a2bb900240100b5000b2a2bb80025b500052ab40005b9002601004d2cb9002701009900132cb900280100c000292ab6001ba7ffeab1
toData,30,2ab400082bb800202b2ab4000bb9002102002ab40005c000032bb80022b1
@@ -1065,8 +1066,8 @@ fromData,33,2a2bb9001b0100b500072a2bb8001cc0001db500052a2bb8001cc0001eb50003b1
toData,27,2b2ab40007b9001902002ab400052bb8001a2ab400032bb8001ab1
com/gemstone/gemfire/internal/cache/FilterProfile,2
-fromData,188,bb012959b7012a4d2c2bb8012b2a2cb500182a2bb8012cb500042a2bb8012db500f42a2bb8012db500f52a2bb8012db500f62a2bb8012cb500052a2bb8012db500f02a2bb8012db500f12a2bb8012db500f22bb8012e3e1d9e0063bb000a591db7012f3a0405b80131360503360615061da2002a2bb801323a072bb801333a082a1907190803b60134190419071908b90037030057840601a7ffd62a1904b500091505b8013157a700143a092a1904b500091505b80131571909bfb1
-toData,199,2ab40018c001292bb801362ab400042ab4001bb401112bb801372ab400f4c0000a2bb801382ab400f5c0000a2bb801382ab400f6c0000a2bb801382ab400052ab4001bb401112bb801372ab400f0c0000a2bb801382ab400f1c0000a2bb801382ab400f2c0000a2bb801382ab400094d2cb9006501003e1d2bb801392cb900a50100b900a601003a041904b9006901009900361904b9006a0100c000a73a051905b901140100c0003b3a061905b900a80100c0007d3a0719062bb8013a19072bb80136a7ffc6b1
+fromData,210,bb012b59b7012c4d2c2bb8012d2a2cb500202ab4000d2bb8012eb900300200572ab400052bb8012fb9007702002ab400072bb8012fb9007702002ab400092bb8012fb9007702002ab4000e2bb8012eb900300200572ab400062bb8012fb9007702002ab400082bb8012fb9007702002ab4000a2bb8012fb9007702002bb801303e1d9e004f05b80132360403360515051da2002c2bb801333a062bb801343a072a1906190703b601352ab4000f19061907b90056030057840501a7ffd41504b8013257a7000e3a081504b80132571908bfb1
+toData,181,2ab40020c0012b2bb801372ab4000d2ab40023b401122bb801382ab400052bb801392ab400072bb801392ab400092bb801392ab4000e2ab40023b401122bb801382ab400062bb801392ab400082bb801392ab4000a2bb801392ab4000f4d2cb900b401003e1d2bb8013a2cb900ab0100b900ac01003a041904b9007001009900361904b900710100c000ad3a051905b901150100c0003c3a061905b900ae0100c000833a0719062bb8013b19072bb80137a7ffc6b1
com/gemstone/gemfire/internal/cache/FilterProfile$OperationMessage,2
fromData,129,2a2bb700522a2bb900530100b5000a2a2bb900540100b500412ab800552bb90056010032b500232a2bb900570100b500292a2bb900580100b5004a2ab40023b8004c99002c2a2bb900540100b500322ab40023b2004ea5000d2ab40023b2004fa600202a2bb80059b50033a700152a2bb900580100b500252a2bb8005ab50027b1
[3/3] incubator-geode git commit: refactoring
RegionVersionHolder.recordVersion()
Posted by bs...@apache.org.
refactoring RegionVersionHolder.recordVersion()
This is from a group refactoring session during a class taught by Michael Feathers.
The old recordVersion code was difficult to understand and was poorly tested. New
JUnit tests are now included that give 100% code coverage for this algorithm.
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/eb28f618
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/eb28f618
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/eb28f618
Branch: refs/heads/develop
Commit: eb28f61866ab42d351e22417eb30feb9e76a0cfe
Parents: 78d8bb3
Author: Bruce Schuchardt <bs...@pivotal.io>
Authored: Tue Jan 19 08:58:14 2016 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Tue Jan 19 09:00:18 2016 -0800
----------------------------------------------------------------------
.../cache/versions/RegionVersionHolder.java | 81 ++++++----
.../versions/RegionVersionHolder2JUnitTest.java | 158 +++++++++++++++++++
2 files changed, 209 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/eb28f618/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/versions/RegionVersionHolder.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/versions/RegionVersionHolder.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/versions/RegionVersionHolder.java
old mode 100644
new mode 100755
index 4872df2..c996031
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/versions/RegionVersionHolder.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/versions/RegionVersionHolder.java
@@ -225,8 +225,13 @@ public class RegionVersionHolder<T> implements Cloneable, DataSerializable {
}
void flushBitSetDuringRecording(long version) {
+ if (this.bitSetVersion + BIT_SET_WIDTH - 1 >= version) {
+ return; // it fits in this bitset
+ }
+
int length = BIT_SET_WIDTH;
int bitCountToFlush = length * 3 / 4;
+
if (logger.isTraceEnabled(LogMarker.RVV)) {
logger.trace(LogMarker.RVV, "flushing RVV bitset bitSetVersion={}; bits={}", this.bitSetVersion, this.bitSet);
}
@@ -316,42 +321,58 @@ public class RegionVersionHolder<T> implements Cloneable, DataSerializable {
}
synchronized void recordVersion(long version) {
-
- if (this.version != version) {
- if (this.bitSet == null) {
- if (this.version < version-1) {
- this.addException(this.version, version);
- if (logger.isTraceEnabled(LogMarker.RVV)) {
- logger.trace(LogMarker.RVV, "Added rvv exception e<rv{} - rv{}>", this.version, version);
- }
- } else if (this.version > version) {
- this.addOlderVersion(version);
- }
- } else { // have a bitSet
- if (this.bitSetVersion + BIT_SET_WIDTH - 1 < version) {
- this.flushBitSetDuringRecording(version);
- }
- if (version < this.bitSetVersion) {
- this.addOlderVersion(version);
- } else {
- // If there's special exception, version maybe >= this.bitSetVersion. We need to fill the hole
- // in the special exception. For example, holder=R5(3,6), bitSetVersion=3, bs=[0]. Adding version=4
- // will become: holder=R5(4,6), bitsetVersion=3, bs[0,1]
- if (this.getSpecialException() != null) {
- this.addOlderVersion(version);
- }
- this.bitSet.set((int)(version-this.bitSetVersion));
- }
- }
- this.version = Math.max(this.version, version);
+ if (this.bitSet != null) {
+ recordVersionWithBitSet(version);
} else {
- if (this.bitSet != null && version>=this.bitSetVersion) {
- this.bitSet.set((int)(version-this.bitSetVersion));
+ recordVersionWithoutBitSet(version);
+ }
+ }
+
+ private void recordVersionWithoutBitSet(long version) {
+ if ( (version - this.version) > 1) {
+ this.addException(this.version, version);
+ logRecordVersion(version);
+ this.version = version;
+ return;
+ }
+ this.addOlderVersion(version);
+ this.version = Math.max(this.version, version);
+ }
+
+ private void recordVersionWithBitSet(long version) {
+ if (this.version == version) {
+ if (version >= this.bitSetVersion) {
+ setVersionInBitSet(version);
}
this.addOlderVersion(version);
+ return;
+ }
+
+ flushBitSetDuringRecording(version);
+
+ if (version >= this.bitSetVersion) {
+ if (this.getSpecialException() != null) {
+ this.addOlderVersion(version);
+ }
+ setVersionInBitSet(version);
+ this.version = Math.max(this.version, version);
+ return;
+ }
+ this.addOlderVersion(version);
+ this.version = Math.max(this.version, version);
+ }
+
+ private void setVersionInBitSet(long version) {
+ this.bitSet.set((int)(version-this.bitSetVersion));
+ }
+
+ private void logRecordVersion(long version) {
+ if (logger.isTraceEnabled(LogMarker.RVV)) {
+ logger.trace(LogMarker.RVV, "Added rvv exception e<rv{} - rv{}>", this.version, version);
}
}
+
/**
* Add an exception that is older than this.bitSetVersion.
*/
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/eb28f618/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/versions/RegionVersionHolder2JUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/versions/RegionVersionHolder2JUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/versions/RegionVersionHolder2JUnitTest.java
new file mode 100755
index 0000000..1c7466d
--- /dev/null
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/versions/RegionVersionHolder2JUnitTest.java
@@ -0,0 +1,158 @@
+package com.gemstone.gemfire.internal.cache.versions;
+
+import static org.junit.Assert.*;
+
+import java.util.Iterator;
+
+import org.junit.Test;
+
+public class RegionVersionHolder2JUnitTest {
+
+ @Test
+ public void testCreateHolderWithBitSet() {
+ createHolderTest(true);
+ }
+
+ @Test
+ public void testCreateHolderWithoutBitSet() {
+ createHolderTest(false);
+ }
+
+ private void createHolderTest(boolean useBitSet) {
+ RegionVersionHolder h = createHolder(useBitSet);
+ assertEquals(0, h.getExceptionCount());
+ }
+
+
+ @Test
+ public void testRecordZeroDoesNothingWithBitSet() {
+ recordZeroDoesNothing(true);
+ }
+
+ @Test
+ public void testRecordZeroDoesNothingWithoutBitSet() {
+ recordZeroDoesNothing(false);
+ }
+
+ private void recordZeroDoesNothing(boolean useBitSet) {
+ RegionVersionHolder h = createHolder(useBitSet);
+ h.recordVersion(0l);
+ assertEquals(0, h.getExceptionCount());
+ assertEquals(0l, h.getVersion());
+ }
+
+ @Test
+ public void testRecordSequentialVersionsWithBitSet() {
+ recordSequentialVersions(true);
+ }
+
+ @Test
+ public void testRecordSequentialVersionsWithoutBitSet() {
+ recordSequentialVersions(false);
+ }
+
+ private void recordSequentialVersions(boolean useBitSet) {
+ RegionVersionHolder h = createHolder(useBitSet);
+ h.recordVersion(1l);
+ h.recordVersion(2l);
+ h.recordVersion(3l);
+ assertEquals(0, h.getExceptionCount());
+ assertEquals(3l, h.getVersion());
+ }
+
+ @Test
+ public void testSkippedVersionCreatesExceptionWithBitSet() {
+ skippedVersionCreatesException(true);
+ }
+
+ @Test
+ public void testSkippedVersionCreatesExceptionWithoutBitSet() {
+ skippedVersionCreatesException(false);
+ }
+
+ private void skippedVersionCreatesException(boolean useBitSet) {
+ RegionVersionHolder h = createHolder(useBitSet);
+ h.recordVersion(2l);
+ assertEquals(1, h.getExceptionCount());
+ assertEquals(2l, h.getVersion());
+ }
+
+ @Test
+ public void testFillExceptionWithBitSet() {
+ fillException(true);
+ }
+
+ @Test
+ public void testFillExceptionWithoutBitSet() {
+ fillException(false);
+ }
+
+ private void fillException(boolean useBitSet) {
+ RegionVersionHolder h = createHolder(useBitSet);
+ h.recordVersion(3l);
+ h.recordVersion(1l);
+ h.recordVersion(2l);
+ h.recordVersion(2l);
+ assertEquals(0, h.getExceptionCount());
+ assertEquals(3l, h.getVersion());
+ }
+
+ @Test
+ public void testFillLargeExceptionWithBitSet() {
+ fillLargeException(true);
+ }
+
+ @Test
+ public void testFillLargeExceptionWithoutBitSet() {
+ fillLargeException(false);
+ }
+
+ private void fillLargeException(boolean useBitSet) {
+ RegionVersionHolder h = createHolder(useBitSet);
+ long bigVersion = RegionVersionHolder.BIT_SET_WIDTH + 1;
+ h.recordVersion(bigVersion);
+ for (long i=0l; i<bigVersion; i++) {
+ h.recordVersion(i);
+ }
+ assertEquals("expected no exceptions in " + h, 0, h.getExceptionCount());
+ assertEquals(bigVersion, h.getVersion());
+ }
+
+ @Test
+ public void testReceiveDuplicateAfterBitSetFlushWithBitSet() {
+ RegionVersionHolder h = createHolder(true);
+ long bigVersion = RegionVersionHolder.BIT_SET_WIDTH + 1;
+ h.recordVersion(bigVersion);
+ for (long i=0l; i<bigVersion; i++) {
+ h.recordVersion(i);
+ }
+ h.recordVersion(bigVersion);
+ assertEquals("expected no exceptions in " + h, 0, h.getExceptionCount());
+ assertEquals(bigVersion, h.getVersion());
+ }
+
+ @Test
+ public void testFillSpecialExceptionWithBitSet() {
+ RegionVersionHolder h = createHolder(true);
+ h.recordVersion(1l);
+ createSpecialException(h);
+ assertEquals(1, h.getExceptionCount());
+ RVVException e = (RVVException)h.getExceptionForTest().iterator().next();
+ assertTrue(h.isSpecialException(e, h));
+ h.recordVersion(2l);
+ // BUG: the exception is not removed
+// assertEquals("unexpected RVV exception : " + h, 0, h.getExceptionCount());
+ }
+
+ private void createSpecialException(RegionVersionHolder h) {
+ h.addException(h.getVersion()-1, h.getVersion()+1);
+ }
+
+ private RegionVersionHolder createHolder(boolean useBitSet) {
+ if (useBitSet) {
+ return new RegionVersionHolder("id");
+ }
+ return new RegionVersionHolder(0l);
+ }
+
+}