You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by ds...@apache.org on 2016/04/21 19:54:58 UTC
[1/3] incubator-geode git commit: GEODE-1238: ensure EntryEventImpls
are released
Repository: incubator-geode
Updated Branches:
refs/heads/develop ff914bd95 -> 702213503
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java
index 2cb98ea..ff810ec 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java
@@ -54,6 +54,7 @@ import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.LoggingThreadGroup;
import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
/**
*
@@ -130,7 +131,7 @@ public class ConcurrentSerialGatewaySenderEventProcessor extends
if (sender.getOrderPolicy() == OrderPolicy.KEY || sender.getOrderPolicy() == OrderPolicy.PARTITION) {
// Create copy since the event id will be changed, otherwise the same
// event will be changed for multiple gateways. Fix for bug 44471.
- EntryEventImpl clonedEvent = new EntryEventImpl((EntryEventImpl)event);
+ @Released EntryEventImpl clonedEvent = new EntryEventImpl((EntryEventImpl)event);
try {
EventID originalEventId = clonedEvent.getEventId();
if (logger.isDebugEnabled()) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderQueue.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderQueue.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderQueue.java
index 7c492db..312fc4b 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderQueue.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderQueue.java
@@ -1134,7 +1134,7 @@ public class SerialGatewaySenderQueue implements RegionQueue {
}
if (wasEmpty) continue;
}
-
+ // release not needed since disallowOffHeapValues called
EntryEventImpl event = EntryEventImpl.create((LocalRegion)region,
Operation.DESTROY, (lastDestroyedKey + 1) , null/* newValue */, null, false,
cache.getMyId());
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/test/java/com/gemstone/gemfire/cache30/DistributedAckRegionCCEDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/com/gemstone/gemfire/cache30/DistributedAckRegionCCEDUnitTest.java b/geode-core/src/test/java/com/gemstone/gemfire/cache30/DistributedAckRegionCCEDUnitTest.java
index d642888..8f922a0 100644
--- a/geode-core/src/test/java/com/gemstone/gemfire/cache30/DistributedAckRegionCCEDUnitTest.java
+++ b/geode-core/src/test/java/com/gemstone/gemfire/cache30/DistributedAckRegionCCEDUnitTest.java
@@ -56,7 +56,8 @@ import com.gemstone.gemfire.internal.cache.RegionClearedException;
import com.gemstone.gemfire.internal.cache.RegionEntry;
import com.gemstone.gemfire.internal.cache.Token;
import com.gemstone.gemfire.internal.cache.TombstoneService;
-import com.gemstone.gemfire.internal.cache.UpdateOperation;
+import com.gemstone.gemfire.internal.cache.UpdateOperation;
+import com.gemstone.gemfire.internal.cache.VersionTagHolder;
import com.gemstone.gemfire.internal.cache.tier.sockets.ClientProxyMembershipID;
import com.gemstone.gemfire.internal.cache.versions.VMVersionTag;
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
@@ -472,10 +473,9 @@ public class DistributedAckRegionCCEDUnitTest extends DistributedAckRegionDUnitT
tag.setEntryVersion(0xFFFFFF);
tag.setDistributedSystemId(1);
tag.setRegionVersion(CCRegion.getVersionVector().getNextVersion());
- EntryEventImpl holder = EntryEventImpl.createVersionTagHolder(tag);
+ VersionTagHolder holder = new VersionTagHolder(tag);
ClientProxyMembershipID id = ClientProxyMembershipID.getNewProxyMembership(CCRegion.getDistributionManager().getSystem());
CCRegion.basicBridgePut("cckey0", "newvalue", null, true, null, id, true, holder, false);
- holder.release();
vm0.invoke(new SerializableRunnable("check conflation count") {
public void run() {
assertEquals("expected one conflated event", 1, CCRegion.getCachePerfStats().getConflatedEventsCount());
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/UpdateVersionJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/UpdateVersionJUnitTest.java b/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/UpdateVersionJUnitTest.java
index 99d3973..6c53855 100644
--- a/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/UpdateVersionJUnitTest.java
+++ b/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/UpdateVersionJUnitTest.java
@@ -51,8 +51,8 @@ public class UpdateVersionJUnitTest {
private static final String regionName = "versionedregion";
- private EntryEventImpl createNewEvent(LocalRegion region, VersionTag tag, Object key) {
- EntryEventImpl updateTimeStampEvent = EntryEventImpl.createVersionTagHolder(tag);
+ private VersionTagHolder createNewEvent(LocalRegion region, VersionTag tag, Object key) {
+ VersionTagHolder updateTimeStampEvent = new VersionTagHolder(tag);
updateTimeStampEvent.setOperation(Operation.UPDATE_VERSION_STAMP);
updateTimeStampEvent.setRegion(region);
if (region instanceof PartitionedRegion) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/ha/EventIdOptimizationDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/ha/EventIdOptimizationDUnitTest.java b/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/ha/EventIdOptimizationDUnitTest.java
index 3c3f052..8096f7d 100755
--- a/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/ha/EventIdOptimizationDUnitTest.java
+++ b/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/ha/EventIdOptimizationDUnitTest.java
@@ -41,6 +41,7 @@ import com.gemstone.gemfire.distributed.DistributedSystem;
import com.gemstone.gemfire.internal.AvailablePort;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.EventIDHolder;
import com.gemstone.gemfire.test.dunit.DistributedTestCase;
import com.gemstone.gemfire.test.dunit.Host;
import com.gemstone.gemfire.test.dunit.LogWriterUtils;
@@ -326,9 +327,9 @@ public class EventIdOptimizationDUnitTest extends DistributedTestCase
ServerRegionProxy srp = new ServerRegionProxy(regionName, pool);
for (int i = 0; i < eventIds.length; i++) {
- srp.destroyOnForTestsOnly(connection, "KEY-" + i, null, Operation.DESTROY, new EntryEventImpl(eventIds[i]), null);
+ srp.destroyOnForTestsOnly(connection, "KEY-" + i, null, Operation.DESTROY, new EventIDHolder(eventIds[i]), null);
}
- srp.destroyOnForTestsOnly(connection, LAST_KEY, null, Operation.DESTROY, new EntryEventImpl(eventIdForLastKey), null);
+ srp.destroyOnForTestsOnly(connection, LAST_KEY, null, Operation.DESTROY, new EventIDHolder(eventIdForLastKey), null);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DestroyEntryPropagationDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DestroyEntryPropagationDUnitTest.java b/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DestroyEntryPropagationDUnitTest.java
index 38fcdd3..9a7d5a4 100755
--- a/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DestroyEntryPropagationDUnitTest.java
+++ b/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DestroyEntryPropagationDUnitTest.java
@@ -39,6 +39,7 @@ import com.gemstone.gemfire.distributed.internal.DistributionConfig;
import com.gemstone.gemfire.internal.AvailablePort;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.EventIDHolder;
import com.gemstone.gemfire.test.dunit.Assert;
import com.gemstone.gemfire.test.dunit.DistributedTestCase;
import com.gemstone.gemfire.test.dunit.Host;
@@ -270,8 +271,8 @@ public class DestroyEntryPropagationDUnitTest extends DistributedTestCase {
assertNotNull(conn1);
assertEquals(PORT2, conn1.getServer().getPort());
ServerRegionProxy srp = new ServerRegionProxy(Region.SEPARATOR+REGION_NAME, pool);
- srp.destroyOnForTestsOnly(conn1, "key1", null, Operation.DESTROY, new EntryEventImpl(new EventID(new byte[] {1},100000,1)), null);
- srp.destroyOnForTestsOnly(conn1, "key2", null, Operation.DESTROY, new EntryEventImpl(new EventID(new byte[] {1},100000,2)), null);
+ srp.destroyOnForTestsOnly(conn1, "key1", null, Operation.DESTROY, new EventIDHolder(new EventID(new byte[] {1},100000,1)), null);
+ srp.destroyOnForTestsOnly(conn1, "key2", null, Operation.DESTROY, new EventIDHolder(new EventID(new byte[] {1},100000,2)), null);
}
catch (Exception ex) {
throw new TestException("Failed while setting acquireConnectionsAndDestroyEntry ", ex);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-wan/src/test/java/com/gemstone/gemfire/internal/cache/UpdateVersionDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-wan/src/test/java/com/gemstone/gemfire/internal/cache/UpdateVersionDUnitTest.java b/geode-wan/src/test/java/com/gemstone/gemfire/internal/cache/UpdateVersionDUnitTest.java
index cabacbc..18a89f8 100644
--- a/geode-wan/src/test/java/com/gemstone/gemfire/internal/cache/UpdateVersionDUnitTest.java
+++ b/geode-wan/src/test/java/com/gemstone/gemfire/internal/cache/UpdateVersionDUnitTest.java
@@ -629,8 +629,8 @@ public class UpdateVersionDUnitTest extends DistributedTestCase {
}
- private EntryEventImpl createNewEvent(LocalRegion region, VersionTag tag, Object key, Object value) {
- EntryEventImpl updateEvent = EntryEventImpl.createVersionTagHolder(tag);
+ private VersionTagHolder createNewEvent(LocalRegion region, VersionTag tag, Object key, Object value) {
+ VersionTagHolder updateEvent = new VersionTagHolder(tag);
updateEvent.setOperation(Operation.UPDATE);
updateEvent.setRegion(region);
if (region instanceof PartitionedRegion) {
[2/3] incubator-geode git commit: GEODE-1238: ensure EntryEventImpls
are released
Posted by ds...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/PartitionedRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/PartitionedRegion.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/PartitionedRegion.java
index 049f7df..c75286e 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/PartitionedRegion.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/PartitionedRegion.java
@@ -252,6 +252,7 @@ import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.LoggingThreadGroup;
import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
import com.gemstone.gemfire.internal.offheap.annotations.Unretained;
import com.gemstone.gemfire.internal.sequencelog.RegionLogger;
import com.gemstone.gemfire.internal.util.TransformUtils;
@@ -2341,7 +2342,7 @@ public class PartitionedRegion extends LocalRegion implements
if (isDebugEnabled) {
logger.debug("PR.postPutAll encountered exception at sendMsgByBucket, ",ex);
}
- EntryEventImpl firstEvent = prMsg.getFirstEvent(this);
+ @Released EntryEventImpl firstEvent = prMsg.getFirstEvent(this);
try {
partialKeys.saveFailedKey(firstEvent.getKey(), ex);
} finally {
@@ -2442,7 +2443,7 @@ public class PartitionedRegion extends LocalRegion implements
if (isDebugEnabled) {
logger.debug("PR.postRemoveAll encountered exception at sendMsgByBucket, ",ex);
}
- EntryEventImpl firstEvent = prMsg.getFirstEvent(this);
+ @Released EntryEventImpl firstEvent = prMsg.getFirstEvent(this);
try {
partialKeys.saveFailedKey(firstEvent.getKey(), ex);
} finally {
@@ -2489,7 +2490,7 @@ public class PartitionedRegion extends LocalRegion implements
final boolean isDebugEnabled = logger.isDebugEnabled();
// retry the put remotely until it finds the right node managing the bucket
- EntryEventImpl event = prMsg.getFirstEvent(this);
+ @Released EntryEventImpl event = prMsg.getFirstEvent(this);
try {
RetryTimeKeeper retryTime = null;
InternalDistributedMember currentTarget = getNodeForBucketWrite(bucketId.intValue(), null);
@@ -2626,7 +2627,8 @@ public class PartitionedRegion extends LocalRegion implements
private VersionedObjectList sendMsgByBucket(final Integer bucketId, RemoveAllPRMessage prMsg)
{
// retry the put remotely until it finds the right node managing the bucket
- EntryEventImpl event = prMsg.getFirstEvent(this);
+ @Released EntryEventImpl event = prMsg.getFirstEvent(this);
+ try {
RetryTimeKeeper retryTime = null;
InternalDistributedMember currentTarget = getNodeForBucketWrite(bucketId.intValue(), null);
if (logger.isDebugEnabled()) {
@@ -2751,6 +2753,9 @@ public class PartitionedRegion extends LocalRegion implements
this.prStats.incRemoveAllRetries();
} // for
// NOTREACHED
+ } finally {
+ event.release();
+ }
}
public VersionedObjectList tryToSendOnePutAllMessage(PutAllPRMessage prMsg,InternalDistributedMember currentTarget) throws DataLocationException {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/ProxyRegionMap.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/ProxyRegionMap.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/ProxyRegionMap.java
index 7b4504d..f0a6543 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/ProxyRegionMap.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/ProxyRegionMap.java
@@ -44,6 +44,7 @@ import com.gemstone.gemfire.internal.cache.versions.VersionSource;
import com.gemstone.gemfire.internal.cache.versions.VersionStamp;
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
/**
* Internal implementation of {@link RegionMap}for regions whose DataPolicy is
@@ -279,7 +280,7 @@ final class ProxyRegionMap implements RegionMap {
if (AbstractRegionMap.shouldCreateCBEvent(this.owner,
!inTokenMode)) {
// fix for bug 39526
- EntryEventImpl e = AbstractRegionMap.createCBEvent(this.owner, op,
+ @Released EntryEventImpl e = AbstractRegionMap.createCBEvent(this.owner, op,
key, null, txId, txEvent, eventId, aCallbackArgument,filterRoutingInfo,bridgeContext, txEntryState, versionTag, tailKey);
boolean cbEventInPending = false;
try {
@@ -310,7 +311,7 @@ final class ProxyRegionMap implements RegionMap {
this.owner.isInitialized())) {
// fix for bug 39526
boolean cbEventInPending = false;
- EntryEventImpl e = AbstractRegionMap.createCBEvent(this.owner,
+ @Released EntryEventImpl e = AbstractRegionMap.createCBEvent(this.owner,
localOp ? Operation.LOCAL_INVALIDATE : Operation.INVALIDATE,
key, newValue, txId, txEvent, eventId, aCallbackArgument,filterRoutingInfo,bridgeContext, txEntryState, versionTag, tailKey);
try {
@@ -344,7 +345,7 @@ final class ProxyRegionMap implements RegionMap {
this.owner.isInitialized())) {
// fix for bug 39526
boolean cbEventInPending = false;
- EntryEventImpl e = AbstractRegionMap.createCBEvent(this.owner, putOp, key,
+ @Released EntryEventImpl e = AbstractRegionMap.createCBEvent(this.owner, putOp, key,
newValue, txId, txEvent, eventId, aCallbackArgument,filterRoutingInfo,bridgeContext, txEntryState, versionTag, tailKey);
try {
AbstractRegionMap.switchEventOwnerAndOriginRemote(e, txEntryState == null);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/QueuedOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/QueuedOperation.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/QueuedOperation.java
index fa7aea1..2d5a698 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/QueuedOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/QueuedOperation.java
@@ -18,6 +18,7 @@ package com.gemstone.gemfire.internal.cache;
import com.gemstone.gemfire.internal.cache.versions.ConcurrentCacheModificationException;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
import com.gemstone.gemfire.*;
import com.gemstone.gemfire.cache.*;
import com.gemstone.gemfire.distributed.DistributedMember;
@@ -80,7 +81,7 @@ public class QueuedOperation
else {
// it is an entry operation
//TODO :EventID should be passed from the sender & should be reused here
- EntryEventImpl ee = EntryEventImpl.create(
+ @Released EntryEventImpl ee = EntryEventImpl.create(
lr, this.op, this.key, null,
this.cbArg, true, src);
try {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteDestroyMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteDestroyMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteDestroyMessage.java
index 4a8c101..750953a 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteDestroyMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteDestroyMessage.java
@@ -57,6 +57,7 @@ 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.offheap.StoredObject;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
import com.gemstone.gemfire.internal.offheap.annotations.Unretained;
import static com.gemstone.gemfire.internal.offheap.annotations.OffHeapIdentifier.ENTRY_EVENT_OLD_VALUE;
import static com.gemstone.gemfire.internal.cache.DistributedCacheOperation.VALUE_IS_BYTES;
@@ -359,7 +360,7 @@ public class RemoteDestroyMessage extends RemoteOperationMessageWithDirectReply
if (r.keyRequiresRegionContext()) {
((KeyWithRegionContext)this.key).setRegionContext(r);
}
- EntryEventImpl event = null;
+ @Released EntryEventImpl event = null;
try {
if (this.bridgeContext != null) {
event = EntryEventImpl.create(r, getOperation(), getKey(), null/*newValue*/,
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteInvalidateMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteInvalidateMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteInvalidateMessage.java
index ee01c9b..c4ed1cc 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteInvalidateMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteInvalidateMessage.java
@@ -51,6 +51,8 @@ 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.offheap.annotations.Released;
+
import static com.gemstone.gemfire.internal.cache.DistributedCacheOperation.VALUE_IS_BYTES;
import static com.gemstone.gemfire.internal.cache.DistributedCacheOperation.VALUE_IS_SERIALIZED_OBJECT;
@@ -189,7 +191,7 @@ public final class RemoteInvalidateMessage extends RemoteDestroyMessage {
if (r.keyRequiresRegionContext()) {
((KeyWithRegionContext)key).setRegionContext(r);
}
- final EntryEventImpl event = EntryEventImpl.create(
+ @Released final EntryEventImpl event = EntryEventImpl.create(
r,
getOperation(),
key,
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemotePutAllMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemotePutAllMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemotePutAllMessage.java
index c706e4f..ff09af9 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemotePutAllMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemotePutAllMessage.java
@@ -59,6 +59,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.offheap.annotations.Released;
/**
* A Replicate Region putAll message. Meant to be sent only to
@@ -356,7 +357,7 @@ public final class RemotePutAllMessage extends RemoteOperationMessageWithDirectR
final DistributedRegion dr = (DistributedRegion)r;
// create a base event and a DPAO for PutAllMessage distributed btw redundant buckets
- EntryEventImpl baseEvent = EntryEventImpl.create(
+ @Released EntryEventImpl baseEvent = EntryEventImpl.create(
r, Operation.PUTALL_CREATE,
null, null, this.callbackArg, false, eventSender, !skipCallbacks);
try {
@@ -383,7 +384,7 @@ public final class RemotePutAllMessage extends RemoteOperationMessageWithDirectR
// final boolean requiresRegionContext = dr.keyRequiresRegionContext();
InternalDistributedMember myId = r.getDistributionManager().getDistributionManagerId();
for (int i = 0; i < putAllDataCount; ++i) {
- EntryEventImpl ev = PutAllPRMessage.getEventFromEntry(r, myId, eventSender, i, putAllData, false, bridgeContext, posDup, !skipCallbacks, isPutDML);
+ @Released EntryEventImpl ev = PutAllPRMessage.getEventFromEntry(r, myId, eventSender, i, putAllData, false, bridgeContext, posDup, !skipCallbacks, isPutDML);
try {
ev.setPutAllOperation(dpao);
if (logger.isDebugEnabled()) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemotePutMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemotePutMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemotePutMessage.java
index 9f51b39..3fccc9a 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemotePutMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemotePutMessage.java
@@ -57,6 +57,7 @@ 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.offheap.StoredObject;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
import com.gemstone.gemfire.internal.offheap.annotations.Unretained;
import com.gemstone.gemfire.internal.util.BlobHelper;
import com.gemstone.gemfire.internal.util.Breadcrumbs;
@@ -683,7 +684,7 @@ public final class RemotePutMessage extends RemoteOperationMessageWithDirectRepl
if (r.keyRequiresRegionContext()) {
((KeyWithRegionContext)this.key).setRegionContext(r);
}
- this.event = EntryEventImpl.create(
+ @Released EntryEventImpl eei = EntryEventImpl.create(
r,
getOperation(),
getKey(),
@@ -693,6 +694,7 @@ public final class RemotePutMessage extends RemoteOperationMessageWithDirectRepl
eventSender,
true/*generateCallbacks*/,
false/*initializeId*/);
+ this.event = eei;
try {
if (this.versionTag != null) {
this.versionTag.replaceNullIDs(getSender());
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteRemoveAllMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteRemoveAllMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteRemoveAllMessage.java
index 7f3a138..01e8d9e 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteRemoveAllMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/RemoteRemoveAllMessage.java
@@ -59,6 +59,7 @@ import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
/**
* A Replicate Region removeAll message. Meant to be sent only to
@@ -340,7 +341,7 @@ public final class RemoteRemoveAllMessage extends RemoteOperationMessageWithDire
final DistributedRegion dr = (DistributedRegion)r;
// create a base event and a op for RemoveAllMessage distributed btw redundant buckets
- EntryEventImpl baseEvent = EntryEventImpl.create(
+ @Released EntryEventImpl baseEvent = EntryEventImpl.create(
r, Operation.REMOVEALL_DESTROY,
null, null, this.callbackArg, false, eventSender, true);
try {
@@ -365,7 +366,7 @@ public final class RemoteRemoveAllMessage extends RemoteOperationMessageWithDire
public void run() {
InternalDistributedMember myId = r.getDistributionManager().getDistributionManagerId();
for (int i = 0; i < removeAllDataCount; ++i) {
- EntryEventImpl ev = RemoveAllPRMessage.getEventFromEntry(r, myId, eventSender, i, removeAllData, false, bridgeContext, posDup, false);
+ @Released EntryEventImpl ev = RemoveAllPRMessage.getEventFromEntry(r, myId, eventSender, i, removeAllData, false, bridgeContext, posDup, false);
try {
ev.setRemoveAllOperation(op);
if (logger.isDebugEnabled()) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/SearchLoadAndWriteProcessor.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/SearchLoadAndWriteProcessor.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/SearchLoadAndWriteProcessor.java
index d9729a7..6642ec6 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/SearchLoadAndWriteProcessor.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/SearchLoadAndWriteProcessor.java
@@ -222,8 +222,7 @@ public class SearchLoadAndWriteProcessor implements MembershipListener {
if (scope == Scope.LOCAL && (region.getPartitionAttributes() == null)) {
return false;
}
- @Released
- CacheEvent listenerEvent = getEventForListener(event);
+ @Released CacheEvent listenerEvent = getEventForListener(event);
try {
if (action == BEFOREUPDATE && listenerEvent.getOperation().isCreate()) {
action = BEFORECREATE;
@@ -870,8 +869,7 @@ public class SearchLoadAndWriteProcessor implements MembershipListener {
return false;
}
}
- @Released
- CacheEvent event = getEventForListener(pevent);
+ @Released CacheEvent event = getEventForListener(pevent);
int action = paction;
if (event.getOperation().isCreate() && action == BEFOREUPDATE) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXCommitMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXCommitMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXCommitMessage.java
index d1644c7..4908e09 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXCommitMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXCommitMessage.java
@@ -75,6 +75,7 @@ import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.LoggingThreadGroup;
import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
/** TXCommitMessage is the message that contains all the information
* that needs to be distributed, on commit, to other cache members.
@@ -1307,6 +1308,7 @@ public class TXCommitMessage extends PooledDistributionMessage implements Member
/*
* This happens when we don't have the bucket and are getting adjunct notification
*/
+ // No need to release because it is added to pendingCallbacks and they will be released later
EntryEventImpl eei = AbstractRegionMap.createCBEvent(this.r, entryOp.op, entryOp.key, entryOp.value, this.msg.txIdent, txEvent, getEventId(entryOp), entryOp.callbackArg,entryOp.filterRoutingInfo,this.msg.bridgeContext, null, entryOp.versionTag, entryOp.tailKey);
if(entryOp.filterRoutingInfo!=null) {
eei.setLocalFilterInfo(entryOp.filterRoutingInfo.getFilterInfo(this.r.getCache().getMyId()));
@@ -1407,7 +1409,7 @@ public class TXCommitMessage extends PooledDistributionMessage implements Member
/*
* This happens when we don't have the bucket and are getting adjunct notification
*/
- EntryEventImpl eei = AbstractRegionMap.createCBEvent(this.r, entryOp.op, entryOp.key, entryOp.value, this.msg.txIdent, txEvent, getEventId(entryOp), entryOp.callbackArg,entryOp.filterRoutingInfo,this.msg.bridgeContext, null, entryOp.versionTag, entryOp.tailKey);
+ @Released EntryEventImpl eei = AbstractRegionMap.createCBEvent(this.r, entryOp.op, entryOp.key, entryOp.value, this.msg.txIdent, txEvent, getEventId(entryOp), entryOp.callbackArg,entryOp.filterRoutingInfo,this.msg.bridgeContext, null, entryOp.versionTag, entryOp.tailKey);
try {
if(entryOp.filterRoutingInfo!=null) {
eei.setLocalFilterInfo(entryOp.filterRoutingInfo.getFilterInfo(this.r.getCache().getMyId()));
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXEntryState.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXEntryState.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXEntryState.java
index 1964c28..a4c6004 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXEntryState.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXEntryState.java
@@ -948,16 +948,15 @@ public class TXEntryState implements Releasable
// + " isDis=" + isLocalEventDistributed());
// System.out.flush();
// }
+ @Retained
EntryEvent getEvent(LocalRegion r, Object key, TXState txs)
{
// dumpOp();
- //TODO:ASIF : Shopuld we generate EventID ? At this point not generating
LocalRegion eventRegion = r;
if (r.isUsedForPartitionedRegionBucket()) {
eventRegion = r.getPartitionedRegion();
}
- EntryEventImpl result = new TxEntryEventImpl(eventRegion, key);
- // OFFHEAP: freeOffHeapResources on this event is called from TXEvent.freeOffHeapResources.
+ @Retained EntryEventImpl result = new TxEntryEventImpl(eventRegion, key);
boolean returnedResult = false;
try {
if (this.destroy == DESTROY_NONE || isOpDestroy()) {
@@ -2065,6 +2064,7 @@ public class TXEntryState implements Releasable
/**
* Creates a local tx entry event
*/
+ @Retained
TxEntryEventImpl(LocalRegion r, Object key) {
//TODO:ASIF :Check if the eventID should be created. Currently not
// creating it
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXRegionState.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXRegionState.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXRegionState.java
index c948f55..116edf7 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXRegionState.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXRegionState.java
@@ -477,6 +477,7 @@ public class TXRegionState {
Object eKey = me.getKey();
TXEntryState txes = (TXEntryState)me.getValue();
if (txes.isDirty() && txes.isOpAnyEvent(r)) {
+ // OFFHEAP: these events are released when TXEvent.release is called
events.add(txes.getEvent(r, eKey, txs));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXRmtEvent.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXRmtEvent.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXRmtEvent.java
index c0493ac..a19ce92 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXRmtEvent.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXRmtEvent.java
@@ -28,6 +28,8 @@ import com.gemstone.gemfire.cache.Operation;
import com.gemstone.gemfire.cache.TransactionEvent;
import com.gemstone.gemfire.cache.TransactionId;
import com.gemstone.gemfire.distributed.DistributedMember;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/**
* <p>
@@ -44,6 +46,8 @@ public class TXRmtEvent implements TransactionEvent
private Cache cache;
+ // This list of EntryEventImpls are released by calling freeOffHeapResources
+ @Released
private List events;
TXRmtEvent(TransactionId txId, Cache cache) {
@@ -204,6 +208,7 @@ public class TXRmtEvent implements TransactionEvent
return (events == null) || events.isEmpty();
}
+ @Retained
private EntryEventImpl createEvent(LocalRegion r, Operation op,
RegionEntry re, Object key, Object newValue,Object aCallbackArgument)
{
@@ -214,7 +219,7 @@ public class TXRmtEvent implements TransactionEvent
if (r.isUsedForPartitionedRegionBucket()) {
eventRegion = r.getPartitionedRegion();
}
- EntryEventImpl event = EntryEventImpl.create(
+ @Retained EntryEventImpl event = EntryEventImpl.create(
eventRegion, op, key, newValue,
aCallbackArgument, // callbackArg
true, // originRemote
@@ -264,7 +269,5 @@ public class TXRmtEvent implements TransactionEvent
e.release();
}
}
- // TODO Auto-generated method stub
-
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXState.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXState.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXState.java
index 3bec397..a67d3cc 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXState.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TXState.java
@@ -62,6 +62,7 @@ import com.gemstone.gemfire.internal.cache.tier.sockets.VersionedObjectList;
import com.gemstone.gemfire.internal.cache.tx.TransactionalOperation.ServerRegionOperation;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/** TXState is the entity that tracks the transaction state on a per
@@ -487,7 +488,7 @@ public class TXState implements TXStateInterface {
/*
* The event must contain the bucket region
*/
- EntryEventImpl ev = (EntryEventImpl)o.es.getEvent(o.r, o.key, o.es.getTXRegionState().getTXState());
+ @Released EntryEventImpl ev = (EntryEventImpl)o.es.getEvent(o.r, o.key, o.es.getTXRegionState().getTXState());
try {
/*
* The routing information is derived from the PR advisor, not the bucket advisor.
@@ -1755,7 +1756,7 @@ public class TXState implements TXStateInterface {
// final boolean requiresRegionContext = theRegion.keyRequiresRegionContext();
InternalDistributedMember myId = theRegion.getDistributionManager().getDistributionManagerId();
for (int i = 0; i < putallOp.putAllDataSize; ++i) {
- EntryEventImpl ev = PutAllPRMessage.getEventFromEntry(theRegion, myId,myId, i, putallOp.putAllData, false, putallOp.getBaseEvent().getContext(), false, !putallOp.getBaseEvent().isGenerateCallbacks(), false);
+ @Released EntryEventImpl ev = PutAllPRMessage.getEventFromEntry(theRegion, myId,myId, i, putallOp.putAllData, false, putallOp.getBaseEvent().getContext(), false, !putallOp.getBaseEvent().isGenerateCallbacks(), false);
try {
ev.setPutAllOperation(putallOp);
if (theRegion.basicPut(ev, false, false, null, false)) {
@@ -1786,11 +1787,13 @@ public class TXState implements TXStateInterface {
public void run() {
InternalDistributedMember myId = theRegion.getDistributionManager().getDistributionManagerId();
for (int i = 0; i < op.removeAllDataSize; ++i) {
- EntryEventImpl ev = RemoveAllPRMessage.getEventFromEntry(theRegion, myId, myId, i, op.removeAllData, false, op.getBaseEvent().getContext(), false, !op.getBaseEvent().isGenerateCallbacks());
+ @Released EntryEventImpl ev = RemoveAllPRMessage.getEventFromEntry(theRegion, myId, myId, i, op.removeAllData, false, op.getBaseEvent().getContext(), false, !op.getBaseEvent().isGenerateCallbacks());
ev.setRemoveAllOperation(op);
try {
theRegion.basicDestroy(ev, true/* should we invoke cacheWriter? */, null);
} catch (EntryNotFoundException ignore) {
+ } finally {
+ ev.release();
}
successfulOps.addKeyAndVersion(op.removeAllData[i].key, null);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TimestampedEntryEventImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TimestampedEntryEventImpl.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TimestampedEntryEventImpl.java
index fca7d62..ce5cfab 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TimestampedEntryEventImpl.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/TimestampedEntryEventImpl.java
@@ -17,6 +17,7 @@
package com.gemstone.gemfire.internal.cache;
import com.gemstone.gemfire.cache.util.TimestampedEntryEvent;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/**
* A subclass of EntryEventImpl used in WAN conflict resolution
@@ -30,6 +31,7 @@ public class TimestampedEntryEventImpl extends EntryEventImpl implements
private long newTimestamp;
private long oldTimestamp;
+ @Retained
public TimestampedEntryEventImpl(EntryEventImpl event, int newDSID, int oldDSID, long newTimestamp, long oldTimestamp) {
super(event);
this.newDSID = newDSID;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/UpdateEntryVersionOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/UpdateEntryVersionOperation.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/UpdateEntryVersionOperation.java
index a3d1c19..fce4dee 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/UpdateEntryVersionOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/UpdateEntryVersionOperation.java
@@ -35,6 +35,7 @@ import com.gemstone.gemfire.distributed.internal.DistributionManager;
import com.gemstone.gemfire.internal.cache.versions.ConcurrentCacheModificationException;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/**
* This operation updates Version stamp of an entry if entry is available and
@@ -90,6 +91,7 @@ public class UpdateEntryVersionOperation extends DistributedCacheOperation {
}
@Override
+ @Retained
protected InternalCacheEvent createEvent(DistributedRegion rgn)
throws EntryNotFoundException {
@@ -97,7 +99,7 @@ public class UpdateEntryVersionOperation extends DistributedCacheOperation {
((KeyWithRegionContext)this.key).setRegionContext(rgn);
}
- EntryEventImpl ev = EntryEventImpl.create(rgn, getOperation(), this.key,
+ @Retained EntryEventImpl ev = EntryEventImpl.create(rgn, getOperation(), this.key,
null /* newValue */, this.callbackArg /*callbackArg*/, true /* originRemote*/ , getSender(), false /*generateCallbacks*/);
ev.setEventId(this.eventId);
ev.setVersionTag(this.versionTag);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/UpdateOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/UpdateOperation.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/UpdateOperation.java
index 452c38f..fc9c7ff 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/UpdateOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/UpdateOperation.java
@@ -45,6 +45,7 @@ import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.offheap.MemoryAllocatorImpl;
import com.gemstone.gemfire.internal.offheap.StoredObject;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
import com.gemstone.gemfire.internal.offheap.annotations.Unretained;
import com.gemstone.gemfire.internal.util.BlobHelper;
import com.gemstone.gemfire.internal.util.Breadcrumbs;
@@ -198,6 +199,7 @@ public class UpdateOperation extends AbstractUpdateOperation
}
@Override
+ @Retained
protected InternalCacheEvent createEvent(DistributedRegion rgn)
throws EntryNotFoundException {
EntryEventImpl ev = createEntryEvent(rgn);
@@ -325,6 +327,7 @@ public class UpdateOperation extends AbstractUpdateOperation
}
}
+ @Retained
protected EntryEventImpl createEntryEvent(DistributedRegion rgn)
{
Object argNewValue = null;
@@ -333,7 +336,7 @@ public class UpdateOperation extends AbstractUpdateOperation
if (rgn.keyRequiresRegionContext()) {
((KeyWithRegionContext)this.key).setRegionContext(rgn);
}
- EntryEventImpl result = EntryEventImpl.create(rgn, getOperation(), this.key,
+ @Retained EntryEventImpl result = EntryEventImpl.create(rgn, getOperation(), this.key,
argNewValue, // oldValue,
this.callbackArg, originRemote, getSender(), generateCallbacks);
setOldValueInEvent(result);
@@ -568,6 +571,7 @@ public class UpdateOperation extends AbstractUpdateOperation
protected transient ClientProxyMembershipID clientID;
@Override
+ @Retained
final public EntryEventImpl createEntryEvent(DistributedRegion rgn)
{
// Object oldValue = null;
@@ -579,7 +583,7 @@ public class UpdateOperation extends AbstractUpdateOperation
if (rgn.keyRequiresRegionContext()) {
((KeyWithRegionContext)this.key).setRegionContext(rgn);
}
- EntryEventImpl ev = EntryEventImpl.create(rgn, getOperation(), this.key,
+ @Retained EntryEventImpl ev = EntryEventImpl.create(rgn, getOperation(), this.key,
argNewValue, this.callbackArg, originRemote, getSender(),
generateCallbacks);
ev.setContext(this.clientID);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/VersionTagHolder.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/VersionTagHolder.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/VersionTagHolder.java
new file mode 100644
index 0000000..48258df
--- /dev/null
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/VersionTagHolder.java
@@ -0,0 +1,34 @@
+/*
+ * 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;
+
+import com.gemstone.gemfire.internal.cache.versions.VersionTag;
+
+/**
+ * This class is just used to hold the version tag for an operation.
+ * It does need release to be called (unlike other EntryEventImpls)
+ * because it never has off-heap value references.
+ */
+public class VersionTagHolder extends EntryEventImpl {
+ public VersionTagHolder(VersionTag<?> tag) {
+ setVersionTag(tag);
+ disallowOffHeapValues();
+ }
+ public VersionTagHolder() {
+ this(null);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/DestroyMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/DestroyMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/DestroyMessage.java
index a30b9f2..3a6f04e 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/DestroyMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/DestroyMessage.java
@@ -59,6 +59,8 @@ import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/**
* A class that specifies a destroy operation.
@@ -249,7 +251,7 @@ public class DestroyMessage extends PartitionMessageWithDirectReply {
if (eventSender == null) {
eventSender = getSender();
}
- EntryEventImpl event = null;
+ @Released EntryEventImpl event = null;
try {
if (r.keyRequiresRegionContext()) {
((KeyWithRegionContext)this.key).setRegionContext(r);
@@ -323,7 +325,7 @@ public class DestroyMessage extends PartitionMessageWithDirectReply {
}
}
else {
- EntryEventImpl e2 = createListenerEvent(event, r, dm.getDistributionManagerId());
+ @Released EntryEventImpl e2 = createListenerEvent(event, r, dm.getDistributionManagerId());
try {
r.invokeDestroyCallbacks(EnumListenerEvent.AFTER_DESTROY, e2, r.isInitialized(), true);
} finally {
@@ -411,7 +413,10 @@ public class DestroyMessage extends PartitionMessageWithDirectReply {
return this.eventId;
}
- /** create a new EntryEvent to be used in notifying listeners, bridge servers, etc. */
+ /** create a new EntryEvent to be used in notifying listeners, bridge servers, etc.
+ * Caller must release result if it is != to sourceEvent
+ */
+ @Retained
EntryEventImpl createListenerEvent(EntryEventImpl sourceEvent, PartitionedRegion r,
InternalDistributedMember member) {
final EntryEventImpl e2;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/FetchBulkEntriesMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/FetchBulkEntriesMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/FetchBulkEntriesMessage.java
index 93501aa..d3078a9 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/FetchBulkEntriesMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/FetchBulkEntriesMessage.java
@@ -55,6 +55,7 @@ import com.gemstone.gemfire.internal.cache.InitialImageOperation;
import com.gemstone.gemfire.internal.cache.KeyWithRegionContext;
import com.gemstone.gemfire.internal.cache.PartitionedRegion;
import com.gemstone.gemfire.internal.cache.PartitionedRegionDataStore;
+import com.gemstone.gemfire.internal.cache.VersionTagHolder;
import com.gemstone.gemfire.internal.cache.tier.InterestType;
import com.gemstone.gemfire.internal.cache.versions.VersionSource;
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
@@ -296,8 +297,7 @@ public final class FetchBulkEntriesMessage extends PartitionMessage
while (it.hasNext()) {
Object key = it.next();
- EntryEventImpl clientEvent = EntryEventImpl
- .createVersionTagHolder();
+ VersionTagHolder clientEvent = new VersionTagHolder();
Object value = map.get(key, null, true, true, true, null,
clientEvent, allowTombstones, false);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/GetMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/GetMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/GetMessage.java
index 047d1c7..d7e50f1 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/GetMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/GetMessage.java
@@ -58,6 +58,7 @@ import com.gemstone.gemfire.internal.cache.PrimaryBucketException;
import com.gemstone.gemfire.internal.cache.TXManagerImpl;
import com.gemstone.gemfire.internal.cache.TXStateProxy;
import com.gemstone.gemfire.internal.cache.Token;
+import com.gemstone.gemfire.internal.cache.VersionTagHolder;
import com.gemstone.gemfire.internal.cache.tier.sockets.ClientProxyMembershipID;
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
@@ -182,7 +183,7 @@ public final class GetMessage extends PartitionMessageWithDirectReply
Object val = null;
try {
if (ds != null) {
- EntryEventImpl event = EntryEventImpl.createVersionTagHolder();
+ VersionTagHolder event = new VersionTagHolder();
try {
if (r.keyRequiresRegionContext()) {
((KeyWithRegionContext)this.key).setRegionContext(r);
@@ -214,8 +215,6 @@ public final class GetMessage extends PartitionMessageWithDirectReply
catch (DataLocationException e) {
sendReply(getSender(), getProcessorId(), dm, new ReplyException(e), r, startTime);
return false;
- } finally {
- event.release();
}
if (logger.isTraceEnabled(LogMarker.DM)) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/InvalidateMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/InvalidateMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/InvalidateMessage.java
index 8c14e49..60508c1 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/InvalidateMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/InvalidateMessage.java
@@ -54,6 +54,7 @@ import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
public final class InvalidateMessage extends DestroyMessage {
private static final Logger logger = LogService.getLogger();
@@ -179,7 +180,7 @@ public final class InvalidateMessage extends DestroyMessage {
if (r.keyRequiresRegionContext()) {
((KeyWithRegionContext)key).setRegionContext(r);
}
- final EntryEventImpl event = EntryEventImpl.create(
+ @Released final EntryEventImpl event = EntryEventImpl.create(
r,
getOperation(),
key,
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PRUpdateEntryVersionMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PRUpdateEntryVersionMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PRUpdateEntryVersionMessage.java
index 7fcb031..e1766dc 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PRUpdateEntryVersionMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PRUpdateEntryVersionMessage.java
@@ -130,6 +130,7 @@ public class PRUpdateEntryVersionMessage extends
((KeyWithRegionContext) key).setRegionContext(pr);
}
+ // release not needed because disallowOffHeapValues called
final EntryEventImpl event = EntryEventImpl.create(pr, getOperation(),
getKey(), null, /* newValue */
null, /* callbackargs */
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PutAllPRMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PutAllPRMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PutAllPRMessage.java
index cace5cc..a88f96f 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PutAllPRMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PutAllPRMessage.java
@@ -72,6 +72,8 @@ import com.gemstone.gemfire.internal.cache.versions.ConcurrentCacheModificationE
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/**
* A Partitioned Region update message. Meant to be sent only to
@@ -359,12 +361,13 @@ public final class PutAllPRMessage extends PartitionMessageWithDirectReply
}
/* we need a event with content for waitForNodeOrCreateBucket() */
+ @Retained
public EntryEventImpl getFirstEvent(PartitionedRegion r) {
if (putAllPRDataSize == 0) {
return null;
}
- EntryEventImpl ev = EntryEventImpl.create(r,
+ @Retained EntryEventImpl ev = EntryEventImpl.create(r,
putAllPRData[0].getOp(),
putAllPRData[0].getKey(),
putAllPRData[0].getValue(),
@@ -408,7 +411,7 @@ public final class PutAllPRMessage extends PartitionMessageWithDirectReply
}
DistributedPutAllOperation dpao = null;
- EntryEventImpl baseEvent = null;
+ @Released EntryEventImpl baseEvent = null;
BucketRegion bucketRegion = null;
PartitionedRegionDataStore ds = r.getDataStore();
InternalDistributedMember myId = r.getDistributionManager().getDistributionManagerId();
@@ -486,7 +489,7 @@ public final class PutAllPRMessage extends PartitionMessageWithDirectReply
* in this request, because these request will be blocked by foundKey
*/
for (int i=0; i<putAllPRDataSize; i++) {
- EntryEventImpl ev = getEventFromEntry(r, myId, eventSender, i,putAllPRData,notificationOnly,bridgeContext,posDup,skipCallbacks, this.isPutDML);
+ @Released EntryEventImpl ev = getEventFromEntry(r, myId, eventSender, i,putAllPRData,notificationOnly,bridgeContext,posDup,skipCallbacks, this.isPutDML);
try {
key = ev.getKey();
@@ -596,6 +599,7 @@ public final class PutAllPRMessage extends PartitionMessageWithDirectReply
return true;
}
+ @Retained
public static EntryEventImpl getEventFromEntry(LocalRegion r,
InternalDistributedMember myId, InternalDistributedMember eventSender,
int idx, DistributedPutAllOperation.PutAllEntryData[] data,
@@ -610,7 +614,7 @@ public final class PutAllPRMessage extends PartitionMessageWithDirectReply
// true/* generate Callbacks */,
// prd.getEventID());
- EntryEventImpl ev = EntryEventImpl.create(r, prd.getOp(), prd.getKey(), prd
+ @Retained EntryEventImpl ev = EntryEventImpl.create(r, prd.getOp(), prd.getKey(), prd
.getValue(), null, false, eventSender, !skipCallbacks, prd.getEventID());
boolean evReturned = false;
try {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PutMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PutMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PutMessage.java
index 77a47b7..db11a8e 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PutMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/PutMessage.java
@@ -67,6 +67,8 @@ import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
import com.gemstone.gemfire.internal.offheap.annotations.Unretained;
import com.gemstone.gemfire.internal.util.BlobHelper;
@@ -450,7 +452,10 @@ public final class PutMessage extends PartitionMessageWithDirectReply implements
// }
- /** create a new EntryEvent to be used in notifying listeners, bridge servers, etc. */
+ /** create a new EntryEvent to be used in notifying listeners, bridge servers, etc.
+ * Caller must release result if it is != to sourceEvent
+ */
+ @Retained
EntryEventImpl createListenerEvent(EntryEventImpl sourceEvent, PartitionedRegion r,
InternalDistributedMember member) {
final EntryEventImpl e2;
@@ -795,7 +800,7 @@ public final class PutMessage extends PartitionMessageWithDirectReply implements
if (r.keyRequiresRegionContext()) {
((KeyWithRegionContext)this.key).setRegionContext(r);
}
- final EntryEventImpl ev = EntryEventImpl.create(
+ @Released final EntryEventImpl ev = EntryEventImpl.create(
r,
getOperation(),
getKey(),
@@ -906,7 +911,7 @@ public final class PutMessage extends PartitionMessageWithDirectReply implements
}
}
else { // notificationOnly
- EntryEventImpl e2 = createListenerEvent(ev, r, dm.getDistributionManagerId());
+ @Released EntryEventImpl e2 = createListenerEvent(ev, r, dm.getDistributionManagerId());
final EnumListenerEvent le;
try {
if (e2.getOperation().isCreate()) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/RemoveAllPRMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/RemoveAllPRMessage.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/RemoveAllPRMessage.java
index c907943..fb46e0c 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/RemoveAllPRMessage.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/RemoveAllPRMessage.java
@@ -73,6 +73,8 @@ import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/**
* PR removeAll
@@ -339,12 +341,13 @@ public final class RemoveAllPRMessage extends PartitionMessageWithDirectReply
}
/* we need a event with content for waitForNodeOrCreateBucket() */
+ @Retained
public EntryEventImpl getFirstEvent(PartitionedRegion r) {
if (removeAllPRDataSize == 0) {
return null;
}
- EntryEventImpl ev = EntryEventImpl.create(r,
+ @Retained EntryEventImpl ev = EntryEventImpl.create(r,
removeAllPRData[0].getOp(),
removeAllPRData[0].getKey(),
null /*value*/,
@@ -388,7 +391,7 @@ public final class RemoveAllPRMessage extends PartitionMessageWithDirectReply
}
DistributedRemoveAllOperation op = null;
- EntryEventImpl baseEvent = null;
+ @Released EntryEventImpl baseEvent = null;
BucketRegion bucketRegion = null;
PartitionedRegionDataStore ds = r.getDataStore();
InternalDistributedMember myId = r.getDistributionManager().getDistributionManagerId();
@@ -466,7 +469,7 @@ public final class RemoveAllPRMessage extends PartitionMessageWithDirectReply
* in this request, because these request will be blocked by foundKey
*/
for (int i=0; i<removeAllPRDataSize; i++) {
- EntryEventImpl ev = getEventFromEntry(r, myId, eventSender, i,removeAllPRData,notificationOnly,bridgeContext,posDup,skipCallbacks);
+ @Released EntryEventImpl ev = getEventFromEntry(r, myId, eventSender, i,removeAllPRData,notificationOnly,bridgeContext,posDup,skipCallbacks);
try {
key = ev.getKey();
@@ -574,14 +577,14 @@ public final class RemoveAllPRMessage extends PartitionMessageWithDirectReply
public boolean canStartRemoteTransaction() {
return true;
}
-
+ @Retained
public static EntryEventImpl getEventFromEntry(LocalRegion r,
InternalDistributedMember myId, InternalDistributedMember eventSender,
int idx, DistributedRemoveAllOperation.RemoveAllEntryData[] data,
boolean notificationOnly, ClientProxyMembershipID bridgeContext,
boolean posDup, boolean skipCallbacks) {
RemoveAllEntryData dataItem = data[idx];
- EntryEventImpl ev = EntryEventImpl.create(r, dataItem.getOp(), dataItem.getKey(), null, null, false, eventSender, !skipCallbacks, dataItem.getEventID());
+ @Retained EntryEventImpl ev = EntryEventImpl.create(r, dataItem.getOp(), dataItem.getKey(), null, null, false, eventSender, !skipCallbacks, dataItem.getEventID());
boolean evReturned = false;
try {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/BaseCommand.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/BaseCommand.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/BaseCommand.java
index 0be56f9..c264616 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/BaseCommand.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/BaseCommand.java
@@ -69,6 +69,7 @@ import com.gemstone.gemfire.internal.cache.PartitionedRegionHelper;
import com.gemstone.gemfire.internal.cache.TXManagerImpl;
import com.gemstone.gemfire.internal.cache.TXStateProxy;
import com.gemstone.gemfire.internal.cache.Token;
+import com.gemstone.gemfire.internal.cache.VersionTagHolder;
import com.gemstone.gemfire.internal.cache.tier.CachedRegionHelper;
import com.gemstone.gemfire.internal.cache.tier.Command;
import com.gemstone.gemfire.internal.cache.tier.InterestType;
@@ -1141,7 +1142,7 @@ public abstract class BaseCommand implements Command {
if (region != null) {
if (region.containsKey(entryKey) || region.containsTombstone(entryKey)) {
- EntryEventImpl versionHolder = EntryEventImpl.createVersionTagHolder();
+ VersionTagHolder versionHolder = new VersionTagHolder();
ClientProxyMembershipID id = servConn == null ? null : servConn.getProxyID();
// From Get70.getValueAndIsObject()
Object data = region.get(entryKey, null, true, true, true, id, versionHolder, true, false);
@@ -1237,7 +1238,7 @@ public abstract class BaseCommand implements Command {
}
for (Object key : region.keySet(true)) {
- EntryEventImpl versionHolder = EntryEventImpl.createVersionTagHolder();
+ VersionTagHolder versionHolder = new VersionTagHolder();
if (keyPattern != null) {
if (!(key instanceof String)) {
// key is not a String, cannot apply regex to this entry
@@ -1338,11 +1339,11 @@ public abstract class BaseCommand implements Command {
VersionedObjectList values, Object riKeys, Set keySet, ServerConnection servConn)
throws IOException {
Object key = null;
- EntryEventImpl versionHolder = null;
+ VersionTagHolder versionHolder = null;
ClientProxyMembershipID requestingClient = servConn == null ? null : servConn.getProxyID();
for (Iterator it = keySet.iterator(); it.hasNext();) {
key = it.next();
- versionHolder = EntryEventImpl.createVersionTagHolder();
+ versionHolder = new VersionTagHolder();
Object value = region.get(key, null, true, true, true, requestingClient, versionHolder, true, false);
@@ -1545,8 +1546,7 @@ public abstract class BaseCommand implements Command {
for (Iterator it = keyList.iterator(); it.hasNext();) {
Object key = it.next();
if (region.containsKey(key) || region.containsTombstone(key)) {
- EntryEventImpl versionHolder = EntryEventImpl
- .createVersionTagHolder();
+ VersionTagHolder versionHolder = new VersionTagHolder();
ClientProxyMembershipID id = servConn == null ? null : servConn
.getProxyID();
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientUpdater.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientUpdater.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientUpdater.java
index 8968f62..ce067d7 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientUpdater.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientUpdater.java
@@ -90,6 +90,7 @@ import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.LoggingThreadGroup;
import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
import com.gemstone.gemfire.internal.sequencelog.EntryLogger;
import com.gemstone.gemfire.security.AuthenticationFailedException;
import com.gemstone.gemfire.security.AuthenticationRequiredException;
@@ -771,7 +772,7 @@ public class CacheClientUpdater extends Thread implements ClientUpdater,
else if (region.hasServerProxy()
&& ServerResponseMatrix.checkForValidStateAfterNotification(region,
key, m.getMessageType()) && (withInterest || !withCQs)) {
- EntryEventImpl newEvent = null;
+ @Released EntryEventImpl newEvent = null;
try {
// Create an event and put the entry
newEvent = EntryEventImpl.create(
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Destroy.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Destroy.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Destroy.java
index 888cf62..593cc08 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Destroy.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Destroy.java
@@ -21,6 +21,7 @@ package com.gemstone.gemfire.internal.cache.tier.sockets.command;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.EventIDHolder;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.PartitionedRegion;
import com.gemstone.gemfire.internal.cache.tier.CachedRegionHelper;
@@ -152,7 +153,7 @@ public class Destroy extends BaseCommand {
}
}
region.basicBridgeDestroy(key, callbackArg, servConn.getProxyID(),
- true, new EntryEventImpl(eventId));
+ true, new EventIDHolder(eventId));
servConn.setModificationInfo(true, regionName, key);
}
catch (EntryNotFoundException e) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Destroy65.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Destroy65.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Destroy65.java
index 513e902..1da422a 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Destroy65.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Destroy65.java
@@ -24,6 +24,7 @@ import com.gemstone.gemfire.internal.cache.Token;
import com.gemstone.gemfire.internal.cache.OpType;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.EventIDHolder;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.PartitionedRegion;
import com.gemstone.gemfire.internal.cache.tier.CachedRegionHelper;
@@ -201,7 +202,7 @@ public class Destroy65 extends BaseCommand {
.readEventIdPartsFromOptmizedByteArray(eventIdPartsBuffer);
EventID eventId = new EventID(servConn.getEventMemberIDByteArray(),
threadId, sequenceId);
- EntryEventImpl clientEvent = new EntryEventImpl(eventId);
+ EventIDHolder clientEvent = new EventIDHolder(eventId);
Breadcrumbs.setEventId(eventId);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/GatewayReceiverCommand.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/GatewayReceiverCommand.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/GatewayReceiverCommand.java
index e2fb686..fce1175 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/GatewayReceiverCommand.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/GatewayReceiverCommand.java
@@ -37,6 +37,7 @@ import com.gemstone.gemfire.i18n.LogWriterI18n;
import com.gemstone.gemfire.internal.Version;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.EventIDHolder;
import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
import com.gemstone.gemfire.internal.cache.KeyWithRegionContext;
import com.gemstone.gemfire.internal.cache.LocalRegion;
@@ -194,7 +195,7 @@ public class GatewayReceiverCommand extends BaseCommand {
int actionType = actionTypePart.getInt();
long versionTimeStamp = VersionTag.ILLEGAL_VERSION_TIMESTAMP;
- EntryEventImpl clientEvent = null;
+ EventIDHolder clientEvent = null;
boolean callbackArgExists = false;
@@ -306,7 +307,7 @@ public class GatewayReceiverCommand extends BaseCommand {
if (region == null) {
handleRegionNull(servConn, regionName, batchId);
} else {
- clientEvent = new EntryEventImpl(eventId);
+ clientEvent = new EventIDHolder(eventId);
if (versionTimeStamp > 0) {
VersionTag tag = VersionTag.create(region.getVersionMember());
tag.setIsGatewayTag(true);
@@ -415,7 +416,7 @@ public class GatewayReceiverCommand extends BaseCommand {
if (region == null) {
handleRegionNull(servConn, regionName, batchId);
} else {
- clientEvent = new EntryEventImpl(eventId);
+ clientEvent = new EventIDHolder(eventId);
if (versionTimeStamp > 0) {
VersionTag tag = VersionTag.create(region.getVersionMember());
tag.setIsGatewayTag(true);
@@ -514,7 +515,7 @@ public class GatewayReceiverCommand extends BaseCommand {
if (region == null) {
handleRegionNull(servConn, regionName, batchId);
} else {
- clientEvent = new EntryEventImpl(eventId);
+ clientEvent = new EventIDHolder(eventId);
if (versionTimeStamp > 0) {
VersionTag tag = VersionTag.create(region.getVersionMember());
tag.setIsGatewayTag(true);
@@ -596,7 +597,7 @@ public class GatewayReceiverCommand extends BaseCommand {
handleRegionNull(servConn, regionName, batchId);
} else {
- clientEvent = new EntryEventImpl(eventId);
+ clientEvent = new EventIDHolder(eventId);
if (versionTimeStamp > 0) {
VersionTag tag = VersionTag.create(region.getVersionMember());
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Get70.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Get70.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Get70.java
index e382c57..b2be3c3 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Get70.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Get70.java
@@ -30,6 +30,7 @@ import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.PartitionedRegion;
import com.gemstone.gemfire.internal.cache.Token;
+import com.gemstone.gemfire.internal.cache.VersionTagHolder;
import com.gemstone.gemfire.internal.cache.tier.CachedRegionHelper;
import com.gemstone.gemfire.internal.cache.tier.Command;
import com.gemstone.gemfire.internal.cache.tier.MessageType;
@@ -303,13 +304,9 @@ public class Get70 extends BaseCommand {
// }
// } else {
ClientProxyMembershipID id = servConn == null ? null : servConn.getProxyID();
- EntryEventImpl versionHolder = EntryEventImpl.createVersionTagHolder();
- try {
+ VersionTagHolder versionHolder = new VersionTagHolder();
// TODO OFFHEAP: optimize
data = ((LocalRegion) region).get(key, callbackArg, true, true, true, id, versionHolder, true, true /*allowReadFromHDFS*/);
- }finally {
- versionHolder.release();
- }
// }
versionTag = versionHolder.getVersionTag();
@@ -368,12 +365,8 @@ public class Get70 extends BaseCommand {
@Retained Object data = null;
ClientProxyMembershipID id = servConn == null ? null : servConn.getProxyID();
- EntryEventImpl versionHolder = EntryEventImpl.createVersionTagHolder();
- try {
- data = ((LocalRegion) region).getRetained(key, callbackArg, true, true, id, versionHolder, true);
- }finally {
- versionHolder.release();
- }
+ VersionTagHolder versionHolder = new VersionTagHolder();
+ data = ((LocalRegion) region).getRetained(key, callbackArg, true, true, id, versionHolder, true);
versionTag = versionHolder.getVersionTag();
// If it is Token.REMOVED, Token.DESTROYED,
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Invalidate.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Invalidate.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Invalidate.java
index b037a88..3f313c9 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Invalidate.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Invalidate.java
@@ -28,6 +28,7 @@ import com.gemstone.gemfire.cache.operations.InvalidateOperationContext;
import com.gemstone.gemfire.distributed.internal.DistributionStats;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.EventIDHolder;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.PartitionedRegion;
import com.gemstone.gemfire.internal.cache.tier.CachedRegionHelper;
@@ -153,7 +154,7 @@ public class Invalidate extends BaseCommand {
.invalidateAuthorize(regionName, key, callbackArg);
callbackArg = invalidateContext.getCallbackArg();
}
- EntryEventImpl clientEvent = new EntryEventImpl(eventId);
+ EventIDHolder clientEvent = new EventIDHolder(eventId);
// msg.isRetry might be set by v7.0 and later clients
if (msg.isRetry()) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put.java
index 46f4471..35cc085 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put.java
@@ -21,6 +21,7 @@ package com.gemstone.gemfire.internal.cache.tier.sockets.command;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.EventIDHolder;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.tier.CachedRegionHelper;
import com.gemstone.gemfire.internal.cache.tier.Command;
@@ -192,12 +193,12 @@ public class Put extends BaseCommand {
// isObject
// the true after null doesn't matter and is not used.
result = region.basicBridgeCreate(key, null, true, callbackArg,
- servConn.getProxyID(), true, new EntryEventImpl(eventId), false);
+ servConn.getProxyID(), true, new EventIDHolder(eventId), false);
}
else {
// Put the entry
result = region.basicBridgePut(key, value, null, isObject, callbackArg,
- servConn.getProxyID(), true, new EntryEventImpl(eventId), servConn.isSqlFabricSystem());
+ servConn.getProxyID(), true, new EventIDHolder(eventId), servConn.isSqlFabricSystem());
}
if (result) {
servConn.setModificationInfo(true, regionName, key);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put61.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put61.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put61.java
index 2943091..cc4c1d6 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put61.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put61.java
@@ -30,6 +30,7 @@ import com.gemstone.gemfire.cache.operations.PutOperationContext;
import com.gemstone.gemfire.distributed.internal.DistributionStats;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.EventIDHolder;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.PartitionedRegion;
import com.gemstone.gemfire.internal.cache.tier.CachedRegionHelper;
@@ -223,7 +224,7 @@ public class Put61 extends BaseCommand {
// isObject
// the true after null doesn't matter and is not used.
result = region.basicBridgeCreate(key, null, true, callbackArg,
- servConn.getProxyID(), true, new EntryEventImpl(eventId), false);
+ servConn.getProxyID(), true, new EventIDHolder(eventId), false);
}
else {
// Put the entry
@@ -232,7 +233,7 @@ public class Put61 extends BaseCommand {
delta = valuePart.getSerializedForm();
}
result = region.basicBridgePut(key, value, delta, isObject,
- callbackArg, servConn.getProxyID(), true, new EntryEventImpl(eventId), servConn
+ callbackArg, servConn.getProxyID(), true, new EventIDHolder(eventId), servConn
.isSqlFabricSystem());
}
if (result) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put65.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put65.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put65.java
index 092e2ee..9be1390 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put65.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/Put65.java
@@ -37,6 +37,7 @@ import com.gemstone.gemfire.internal.Version;
import com.gemstone.gemfire.internal.cache.CachedDeserializable;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.EventIDHolder;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.PartitionedRegion;
import com.gemstone.gemfire.internal.cache.TXManagerImpl;
@@ -220,7 +221,7 @@ public class Put65 extends BaseCommand {
long sequenceId = EventID
.readEventIdPartsFromOptmizedByteArray(eventIdPartsBuffer);
- EntryEventImpl clientEvent = new EntryEventImpl(
+ EventIDHolder clientEvent = new EventIDHolder(
new EventID(servConn.getEventMemberIDByteArray(),
threadId, sequenceId));
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tx/DistTxEntryEvent.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tx/DistTxEntryEvent.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tx/DistTxEntryEvent.java
index a108965..6e7c21c 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tx/DistTxEntryEvent.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tx/DistTxEntryEvent.java
@@ -36,6 +36,7 @@ import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.DistributedPutAllOperation.EntryVersionsList;
import com.gemstone.gemfire.internal.cache.DistributedPutAllOperation.PutAllEntryData;
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/**
*
@@ -46,7 +47,13 @@ public class DistTxEntryEvent extends EntryEventImpl {
protected static final byte HAS_PUTALL_OP = 0x1;
protected static final byte HAS_REMOVEALL_OP = 0x2;
- // For Serialization
+ /**
+ * TODO DISTTX: callers of this constructor need to
+ * make sure that release is called. In general
+ * the distributed tx code needs to be reviewed to
+ * see if it correctly handles off-heap.
+ */
+ @Retained
public DistTxEntryEvent(EntryEventImpl entry) {
super(entry);
}
@@ -179,7 +186,7 @@ public class DistTxEntryEvent extends EntryEventImpl {
}
}
}
-
+ // TODO DISTTX: release this event?
EntryEventImpl e = EntryEventImpl.create(
this.region, Operation.PUTALL_CREATE,
null, null, null, true, this.getDistributedMember(), true, true);
@@ -246,7 +253,7 @@ public class DistTxEntryEvent extends EntryEventImpl {
removeAllData[i].versionTag = versionTags.get(i);
}
}
-
+ // TODO DISTTX: release this event
EntryEventImpl e = EntryEventImpl.create(
this.region, Operation.REMOVEALL_DESTROY,
null, null, null, true, this.getDistributedMember(), true, true);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tx/PartitionedTXRegionStub.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tx/PartitionedTXRegionStub.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tx/PartitionedTXRegionStub.java
index 323dcc4..6723646 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tx/PartitionedTXRegionStub.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/tx/PartitionedTXRegionStub.java
@@ -46,6 +46,7 @@ import com.gemstone.gemfire.internal.cache.partitioned.RemoveAllPRMessage;
import com.gemstone.gemfire.internal.cache.tier.sockets.ClientProxyMembershipID;
import com.gemstone.gemfire.internal.cache.tier.sockets.VersionedObjectList;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
import java.util.Collections;
import java.util.HashMap;
@@ -386,7 +387,7 @@ public class PartitionedTXRegionStub extends AbstractPeerTXRegionStub {
partialKeys.consolidate(pre.getResult());
} catch (Exception ex) {
// If failed at other exception
- EntryEventImpl firstEvent = prMsg.getFirstEvent(pr);
+ @Released EntryEventImpl firstEvent = prMsg.getFirstEvent(pr);
try {
partialKeys.saveFailedKey(firstEvent.getKey(), ex);
} finally {
@@ -444,8 +445,12 @@ public class PartitionedTXRegionStub extends AbstractPeerTXRegionStub {
partialKeys.consolidate(pre.getResult());
} catch (Exception ex) {
// If failed at other exception
- EntryEventImpl firstEvent = prMsg.getFirstEvent(pr);
+ @Released EntryEventImpl firstEvent = prMsg.getFirstEvent(pr);
+ try {
partialKeys.saveFailedKey(firstEvent.getKey(), ex);
+ } finally {
+ firstEvent.release();
+ }
}
}
pr.prStats.endRemoveAll(startTime);
@@ -477,7 +482,7 @@ public class PartitionedTXRegionStub extends AbstractPeerTXRegionStub {
// retry the put remotely until it finds the right node managing the bucket
InternalDistributedMember currentTarget = pr.getOrCreateNodeForBucketWrite(bucketId.intValue(), null);
if(!currentTarget.equals(this.state.getTarget())) {
- EntryEventImpl firstEvent = prMsg.getFirstEvent(pr);
+ @Released EntryEventImpl firstEvent = prMsg.getFirstEvent(pr);
try {
throw new TransactionDataNotColocatedException(LocalizedStrings.PartitionedRegion_KEY_0_NOT_COLOCATED_WITH_TRANSACTION.toLocalizedString(firstEvent.getKey()));
} finally {
@@ -507,8 +512,12 @@ public class PartitionedTXRegionStub extends AbstractPeerTXRegionStub {
// retry the put remotely until it finds the right node managing the bucket
InternalDistributedMember currentTarget = pr.getOrCreateNodeForBucketWrite(bucketId.intValue(), null);
if(!currentTarget.equals(this.state.getTarget())) {
- EntryEventImpl firstEvent = prMsg.getFirstEvent(pr);
+ @Released EntryEventImpl firstEvent = prMsg.getFirstEvent(pr);
+ try {
throw new TransactionDataNotColocatedException(LocalizedStrings.PartitionedRegion_KEY_0_NOT_COLOCATED_WITH_TRANSACTION.toLocalizedString(firstEvent.getKey()));
+ } finally {
+ firstEvent.release();
+ }
}
try {
return pr.tryToSendOneRemoveAllMessage(prMsg,currentTarget);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySender.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySender.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySender.java
index 4f3488b..94524bd 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySender.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySender.java
@@ -74,6 +74,7 @@ import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.LoggingThreadGroup;
import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
import com.gemstone.gemfire.internal.offheap.Releasable;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
import com.gemstone.gemfire.internal.offheap.annotations.Retained;
import com.gemstone.gemfire.internal.offheap.annotations.Unretained;
@@ -875,8 +876,8 @@ public abstract class AbstractGatewaySender implements GatewaySender,
getStatistics().incEventsFiltered();
return;
}
-
- EntryEventImpl clonedEvent = new EntryEventImpl(event, false);
+ // released by this method or transfers ownership to TmpQueueEvent
+ @Released EntryEventImpl clonedEvent = new EntryEventImpl(event, false);
boolean freeClonedEvent = true;
try {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySenderEventProcessor.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
index 9cde6dd..2ef5e19 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
@@ -816,10 +816,12 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
if (pdxRegion != null && pdxRegion.size() != pdxEventsMap.size()) {
for (Map.Entry<Object, Object> typeEntry : pdxRegion.entrySet()) {
if(!pdxEventsMap.containsKey(typeEntry.getKey())){
+ // event should never be off-heap so it does not need to be released
EntryEventImpl event = EntryEventImpl.create(
(LocalRegion) pdxRegion, Operation.UPDATE,
typeEntry.getKey(), typeEntry.getValue(), null, false,
cache.getMyId());
+ event.disallowOffHeapValues();
event.setEventId(new EventID(cache.getSystem()));
List<Integer> allRemoteDSIds = new ArrayList<Integer>();
for (GatewaySender sender : cache.getGatewaySenders()) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/BatchDestroyOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/BatchDestroyOperation.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/BatchDestroyOperation.java
index 988c526..717253e 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/BatchDestroyOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/serial/BatchDestroyOperation.java
@@ -44,6 +44,7 @@ import com.gemstone.gemfire.internal.cache.RegionQueue;
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.offheap.annotations.Retained;
/**
* Handles distribution messaging for destroying a batch of entry in a queue region.
@@ -157,6 +158,7 @@ public class BatchDestroyOperation extends DistributedCacheOperation {
}
@Override
+ @Retained
protected final InternalCacheEvent createEvent(DistributedRegion rgn)
throws EntryNotFoundException {
EntryEventImpl ev = createEntryEvent(rgn);
@@ -176,8 +178,9 @@ public class BatchDestroyOperation extends DistributedCacheOperation {
}
}
+ @Retained
EntryEventImpl createEntryEvent(DistributedRegion rgn) {
- EntryEventImpl event = EntryEventImpl.create(rgn, getOperation(), this.key,
+ @Retained EntryEventImpl event = EntryEventImpl.create(rgn, getOperation(), this.key,
null, this.callbackArg, true, getSender());
// event.setNewEventId(); Don't set the event here...
setOldValueInEvent(event);
[3/3] incubator-geode git commit: GEODE-1238: ensure EntryEventImpls
are released
Posted by ds...@apache.org.
GEODE-1238: ensure EntryEventImpls are released
-- reviewed all EntryEventImpl creators
-- created two new subclasses to mark ones that are never off-heap
-- added annotations and comments on all creators
-- added missing release calls
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/70221350
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/70221350
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/70221350
Branch: refs/heads/develop
Commit: 7022135035f5a1804eff38c2a159c0352728373d
Parents: ff914bd
Author: Darrel Schneider <ds...@pivotal.io>
Authored: Fri Apr 15 11:42:19 2016 -0700
Committer: Darrel Schneider <ds...@pivotal.io>
Committed: Thu Apr 21 10:53:30 2016 -0700
----------------------------------------------------------------------
.../client/internal/ServerRegionProxy.java | 3 +-
.../internal/admin/ClientStatsManager.java | 3 +-
.../cache/AbstractBucketRegionQueue.java | 3 +-
.../internal/cache/AbstractLRURegionMap.java | 4 +
.../internal/cache/AbstractRegionEntry.java | 2 +-
.../internal/cache/AbstractRegionMap.java | 15 +-
.../gemfire/internal/cache/BucketRegion.java | 14 +-
.../internal/cache/BucketRegionQueue.java | 7 +-
.../internal/cache/DestroyOperation.java | 6 +-
.../internal/cache/DistPeerTXStateStub.java | 2 +
.../gemfire/internal/cache/DistTXState.java | 9 +-
.../cache/DistTXStateOnCoordinator.java | 2 +
.../DistTXStateProxyImplOnCoordinator.java | 2 +
.../cache/DistributedCacheOperation.java | 4 +-
.../cache/DistributedPutAllOperation.java | 23 ++-
.../internal/cache/DistributedRegion.java | 9 +-
.../cache/DistributedRemoveAllOperation.java | 22 ++-
.../gemfire/internal/cache/EntryEventImpl.java | 41 +-----
.../gemfire/internal/cache/EntryExpiryTask.java | 9 +-
.../gemfire/internal/cache/EventIDHolder.java | 35 +++++
.../gemfire/internal/cache/FilterProfile.java | 13 +-
.../gemfire/internal/cache/HARegion.java | 5 +-
.../internal/cache/InitialImageOperation.java | 2 +-
.../internal/cache/InvalidateOperation.java | 5 +-
.../gemfire/internal/cache/LocalRegion.java | 141 ++++++++++---------
.../internal/cache/PartitionedRegion.java | 13 +-
.../gemfire/internal/cache/ProxyRegionMap.java | 7 +-
.../gemfire/internal/cache/QueuedOperation.java | 3 +-
.../internal/cache/RemoteDestroyMessage.java | 3 +-
.../internal/cache/RemoteInvalidateMessage.java | 4 +-
.../internal/cache/RemotePutAllMessage.java | 5 +-
.../internal/cache/RemotePutMessage.java | 4 +-
.../internal/cache/RemoteRemoveAllMessage.java | 5 +-
.../cache/SearchLoadAndWriteProcessor.java | 6 +-
.../gemfire/internal/cache/TXCommitMessage.java | 4 +-
.../gemfire/internal/cache/TXEntryState.java | 6 +-
.../gemfire/internal/cache/TXRegionState.java | 1 +
.../gemfire/internal/cache/TXRmtEvent.java | 9 +-
.../gemfire/internal/cache/TXState.java | 9 +-
.../cache/TimestampedEntryEventImpl.java | 2 +
.../cache/UpdateEntryVersionOperation.java | 4 +-
.../gemfire/internal/cache/UpdateOperation.java | 8 +-
.../internal/cache/VersionTagHolder.java | 34 +++++
.../cache/partitioned/DestroyMessage.java | 11 +-
.../partitioned/FetchBulkEntriesMessage.java | 4 +-
.../internal/cache/partitioned/GetMessage.java | 5 +-
.../cache/partitioned/InvalidateMessage.java | 3 +-
.../PRUpdateEntryVersionMessage.java | 1 +
.../cache/partitioned/PutAllPRMessage.java | 12 +-
.../internal/cache/partitioned/PutMessage.java | 11 +-
.../cache/partitioned/RemoveAllPRMessage.java | 13 +-
.../cache/tier/sockets/BaseCommand.java | 12 +-
.../cache/tier/sockets/CacheClientUpdater.java | 3 +-
.../cache/tier/sockets/command/Destroy.java | 3 +-
.../cache/tier/sockets/command/Destroy65.java | 3 +-
.../sockets/command/GatewayReceiverCommand.java | 11 +-
.../cache/tier/sockets/command/Get70.java | 15 +-
.../cache/tier/sockets/command/Invalidate.java | 3 +-
.../cache/tier/sockets/command/Put.java | 5 +-
.../cache/tier/sockets/command/Put61.java | 5 +-
.../cache/tier/sockets/command/Put65.java | 3 +-
.../internal/cache/tx/DistTxEntryEvent.java | 13 +-
.../cache/tx/PartitionedTXRegionStub.java | 17 ++-
.../cache/wan/AbstractGatewaySender.java | 5 +-
.../AbstractGatewaySenderEventProcessor.java | 2 +
.../cache/wan/serial/BatchDestroyOperation.java | 5 +-
...urrentSerialGatewaySenderEventProcessor.java | 3 +-
.../wan/serial/SerialGatewaySenderQueue.java | 2 +-
.../DistributedAckRegionCCEDUnitTest.java | 6 +-
.../internal/cache/UpdateVersionJUnitTest.java | 4 +-
.../cache/ha/EventIdOptimizationDUnitTest.java | 5 +-
.../DestroyEntryPropagationDUnitTest.java | 5 +-
.../internal/cache/UpdateVersionDUnitTest.java | 4 +-
73 files changed, 437 insertions(+), 265 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ServerRegionProxy.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ServerRegionProxy.java b/geode-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ServerRegionProxy.java
index 40f240f..23e04fd 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ServerRegionProxy.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ServerRegionProxy.java
@@ -42,6 +42,7 @@ import com.gemstone.gemfire.internal.cache.ClientServerObserver;
import com.gemstone.gemfire.internal.cache.ClientServerObserverHolder;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.EventIDHolder;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.TXCommitMessage;
import com.gemstone.gemfire.internal.cache.TXManagerImpl;
@@ -192,7 +193,7 @@ public class ServerRegionProxy extends ServerProxy implements ServerRegionDataAc
EventID eventId,
Object callbackArg)
{
- EntryEventImpl event = new EntryEventImpl(eventId);
+ EventIDHolder event = new EventIDHolder(eventId);
PutOp.execute(con, this.pool, this.regionName, key, value, event, callbackArg, this.pool.getPRSingleHopEnabled());
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/admin/ClientStatsManager.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/admin/ClientStatsManager.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/admin/ClientStatsManager.java
index 20f6dbf..2db7eb1 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/admin/ClientStatsManager.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/admin/ClientStatsManager.java
@@ -36,6 +36,7 @@ import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.EventID;
import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
import com.gemstone.gemfire.management.internal.cli.CliUtil;
/**
@@ -84,7 +85,7 @@ public class ClientStatsManager {
ServerRegionProxy regionProxy = new ServerRegionProxy(ClientHealthMonitoringRegion.ADMIN_REGION_NAME, pool);
EventID eventId = new EventID(ds);
- EntryEventImpl event = new EntryEventImpl((Object)null);
+ @Released EntryEventImpl event = new EntryEventImpl((Object)null);
try {
event.setEventId(eventId);
regionProxy.putForMetaRegion(ds.getMemberId(), stats, null, event, null, true);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractBucketRegionQueue.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractBucketRegionQueue.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractBucketRegionQueue.java
index 2c8f493..1f8da88 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractBucketRegionQueue.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractBucketRegionQueue.java
@@ -48,6 +48,7 @@ import com.gemstone.gemfire.internal.concurrent.ConcurrentHashSet;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.offheap.OffHeapRegionEntryHelper;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
public abstract class AbstractBucketRegionQueue extends BucketRegion {
protected static final Logger logger = LogService.getLogger();
@@ -232,7 +233,7 @@ public abstract class AbstractBucketRegionQueue extends BucketRegion {
if (logger.isDebugEnabled()) {
logger.debug(" destroying primary key {}", key);
}
- EntryEventImpl event = getPartitionedRegion().newDestroyEntryEvent(key,
+ @Released EntryEventImpl event = getPartitionedRegion().newDestroyEntryEvent(key,
null);
event.setEventId(new EventID(cache.getSystem()));
try {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractLRURegionMap.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractLRURegionMap.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractLRURegionMap.java
index a19ee66..2cc7a55 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractLRURegionMap.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractLRURegionMap.java
@@ -177,6 +177,10 @@ public abstract class AbstractLRURegionMap extends AbstractRegionMap {
// make sure this cached deserializable is still in the entry
// @todo what if a clear is done and this entry is no longer in the region?
{
+ if (_getCCHelper().getEvictionAlgorithm().isLRUEntry()) {
+ // no need to worry about the value changing form with entry LRU.
+ return false;
+ }
Object curVal = le._getValue(); // OFFHEAP: _getValue ok
if (curVal != cd) {
if (cd instanceof StoredObject) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionEntry.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionEntry.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionEntry.java
index e78c8eb..debc3da 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionEntry.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionEntry.java
@@ -2070,7 +2070,7 @@ public abstract class AbstractRegionEntry implements RegionEntry,
newValue[0] = v;
}
};
- TimestampedEntryEventImpl timestampedEvent =
+ @Released TimestampedEntryEventImpl timestampedEvent =
(TimestampedEntryEventImpl)event.getTimestampedEvent(tagDsid, stampDsid, tagTime, stampTime);
// gateway conflict resolvers will usually want to see the old value
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
index 1cb7a20..3286373 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractRegionMap.java
@@ -1626,7 +1626,7 @@ public abstract class AbstractRegionMap implements RegionMap {
// a receipt of a TXCommitMessage AND there are callbacks installed
// for this region
boolean invokeCallbacks = shouldCreateCBEvent(owner, isRegionReady || inRI);
- EntryEventImpl cbEvent = createCBEvent(owner, op,
+ @Released EntryEventImpl cbEvent = createCBEvent(owner, op,
key, null, txId, txEvent, eventId, aCallbackArgument, filterRoutingInfo, bridgeContext, txEntryState, versionTag, tailKey);
try {
@@ -1858,7 +1858,7 @@ public abstract class AbstractRegionMap implements RegionMap {
// the destroy is already applied on the Initial image provider, thus
// causing region entry to be absent.
// Notify clients with client events.
- EntryEventImpl cbEvent = createCBEvent(owner, op,
+ @Released EntryEventImpl cbEvent = createCBEvent(owner, op,
key, null, txId, txEvent, eventId, aCallbackArgument,
filterRoutingInfo, bridgeContext, txEntryState, versionTag, tailKey);
try {
@@ -2397,7 +2397,7 @@ public abstract class AbstractRegionMap implements RegionMap {
final LocalRegion owner = _getOwner();
owner.checkBeforeEntrySync(txEvent);
- EntryEventImpl cbEvent = null;
+ @Released EntryEventImpl cbEvent = null;
boolean forceNewEntry = !owner.isInitialized() && owner.isAllEvents();
final boolean hasRemoteOrigin = !((TXId)txId).getMemberId().equals(owner.getMyId());
@@ -2997,7 +2997,7 @@ public abstract class AbstractRegionMap implements RegionMap {
stats.incEvictionsInProgress();
// set the flag on event saying the entry should be evicted
// and not indexed
- EntryEventImpl destroyEvent = EntryEventImpl.create (owner, Operation.DESTROY, event.getKey(),
+ @Released EntryEventImpl destroyEvent = EntryEventImpl.create (owner, Operation.DESTROY, event.getKey(),
null/* newValue */, null, false, owner.getMyId());
try {
@@ -3340,8 +3340,8 @@ public abstract class AbstractRegionMap implements RegionMap {
final boolean isTXHost = txEntryState != null;
final boolean isClientTXOriginator = owner.cache.isClient() && !hasRemoteOrigin;
final boolean isRegionReady = owner.isInitialized();
- EntryEventImpl cbEvent = null;
- EntryEventImpl sqlfEvent = null;
+ @Released EntryEventImpl cbEvent = null;
+ @Released EntryEventImpl sqlfEvent = null;
boolean invokeCallbacks = shouldCreateCBEvent(owner, isRegionReady);
boolean cbEventInPending = false;
cbEvent = createCBEvent(owner, putOp, key, newValue, txId,
@@ -3785,6 +3785,7 @@ public abstract class AbstractRegionMap implements RegionMap {
}
/** create a callback event for applying a transactional change to the local cache */
+ @Retained
public static final EntryEventImpl createCBEvent(final LocalRegion re,
Operation op, Object key, Object newValue, TransactionId txId,
TXRmtEvent txEvent,EventID eventId, Object aCallbackArgument,FilterRoutingInfo filterRoutingInfo,ClientProxyMembershipID bridgeContext, TXEntryState txEntryState, VersionTag versionTag, long tailKey)
@@ -3799,7 +3800,7 @@ public abstract class AbstractRegionMap implements RegionMap {
eventRegion = re.getPartitionedRegion();
}
- EntryEventImpl retVal = EntryEventImpl.create(
+ @Retained EntryEventImpl retVal = EntryEventImpl.create(
re, op, key, newValue,
aCallbackArgument,
txEntryState == null, originator);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegion.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegion.java
index 3946fda..e0f043b 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegion.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegion.java
@@ -97,6 +97,8 @@ 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.offheap.StoredObject;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
import com.gemstone.gemfire.internal.offheap.annotations.Unretained;
import com.gemstone.gemfire.internal.concurrent.AtomicLong5;
@@ -1118,6 +1120,7 @@ implements Bucket
* @return an event for EVICT_DESTROY
*/
@Override
+ @Retained
protected EntryEventImpl generateEvictDestroyEvent(Object key) {
EntryEventImpl event = super.generateEvictDestroyEvent(key);
event.setInvokePRCallbacks(true); //see bug 40797
@@ -1620,6 +1623,7 @@ implements Bucket
//we already distributed this info.
}
+ @Retained
EntryEventImpl createEventForPR(EntryEventImpl sourceEvent) {
EntryEventImpl e2 = new EntryEventImpl(sourceEvent);
boolean returned = false;
@@ -1663,7 +1667,7 @@ implements Bucket
}
super.invokeTXCallbacks(eventType, event, callThem);
}
- final EntryEventImpl prevent = createEventForPR(event);
+ @Released final EntryEventImpl prevent = createEventForPR(event);
try {
this.partitionedRegion.invokeTXCallbacks(eventType, prevent, this.partitionedRegion.isInitialized() ? callDispatchListenerEvent : false);
} finally {
@@ -1691,7 +1695,7 @@ implements Bucket
}
super.invokeDestroyCallbacks(eventType, event, callThem, notifyGateways);
}
- final EntryEventImpl prevent = createEventForPR(event);
+ @Released final EntryEventImpl prevent = createEventForPR(event);
try {
this.partitionedRegion.invokeDestroyCallbacks(eventType, prevent, this.partitionedRegion.isInitialized() ? callDispatchListenerEvent : false, false);
} finally {
@@ -1718,7 +1722,7 @@ implements Bucket
}
super.invokeInvalidateCallbacks(eventType, event, callThem);
}
- final EntryEventImpl prevent = createEventForPR(event);
+ @Released final EntryEventImpl prevent = createEventForPR(event);
try {
this.partitionedRegion.invokeInvalidateCallbacks(eventType, prevent, this.partitionedRegion.isInitialized() ? callDispatchListenerEvent : false);
} finally {
@@ -1749,7 +1753,7 @@ implements Bucket
super.invokePutCallbacks(eventType, event, callThem, notifyGateways);
}
- final EntryEventImpl prevent = createEventForPR(event);
+ @Released final EntryEventImpl prevent = createEventForPR(event);
try {
this.partitionedRegion.invokePutCallbacks(eventType, prevent,
this.partitionedRegion.isInitialized() ? callDispatchListenerEvent : false, false);
@@ -2577,7 +2581,7 @@ implements Bucket
public boolean customEvictDestroy(Object key)
{
checkReadiness();
- final EntryEventImpl event =
+ @Released final EntryEventImpl event =
generateCustomEvictDestroyEvent(key);
event.setCustomEviction(true);
boolean locked = false;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegionQueue.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegionQueue.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegionQueue.java
index 6b76c31..8b4f240 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegionQueue.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketRegionQueue.java
@@ -53,6 +53,7 @@ import com.gemstone.gemfire.internal.concurrent.Atomics;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.offheap.OffHeapRegionEntryHelper;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
/**
*
@@ -504,10 +505,10 @@ public class BucketRegionQueue extends AbstractBucketRegionQueue {
if (logger.isDebugEnabled()) {
logger.debug(" destroying primary key {}", key);
}
- EntryEventImpl event = getPartitionedRegion().newDestroyEntryEvent(key,
+ @Released EntryEventImpl event = getPartitionedRegion().newDestroyEntryEvent(key,
null);
- event.setEventId(new EventID(cache.getSystem()));
try {
+ event.setEventId(new EventID(cache.getSystem()));
event.setRegion(this);
basicDestroy(event, true, null);
checkReadiness();
@@ -531,8 +532,6 @@ public class BucketRegionQueue extends AbstractBucketRegionQueue {
+ key, rde);
}
} finally {
- //merge42180: are we considering offheap in cedar. Comment freeOffHeapReference intentionally
- //event.freeOffHeapReferences();
event.release();
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DestroyOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DestroyOperation.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DestroyOperation.java
index b0ac963..e267190 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DestroyOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DestroyOperation.java
@@ -34,6 +34,7 @@ import com.gemstone.gemfire.distributed.internal.DistributionManager;
import com.gemstone.gemfire.internal.cache.tier.sockets.ClientProxyMembershipID;
import com.gemstone.gemfire.internal.cache.versions.ConcurrentCacheModificationException;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/**
* Handles distribution messaging for destroying an entry in a region.
@@ -125,6 +126,7 @@ public class DestroyOperation extends DistributedCacheOperation
}
@Override
+ @Retained
protected final InternalCacheEvent createEvent(DistributedRegion rgn)
throws EntryNotFoundException {
if (rgn.keyRequiresRegionContext()) {
@@ -151,9 +153,10 @@ public class DestroyOperation extends DistributedCacheOperation
}
}
+ @Retained
EntryEventImpl createEntryEvent(DistributedRegion rgn)
{
- EntryEventImpl event = EntryEventImpl.create(rgn,
+ @Retained EntryEventImpl event = EntryEventImpl.create(rgn,
getOperation(), this.key, null, this.callbackArg, true, getSender());
// event.setNewEventId(); Don't set the event here...
setOldValueInEvent(event);
@@ -258,6 +261,7 @@ public class DestroyOperation extends DistributedCacheOperation
}
@Override
+ @Retained
EntryEventImpl createEntryEvent(DistributedRegion rgn)
{
EntryEventImpl event = EntryEventImpl.create(rgn, getOperation(),
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistPeerTXStateStub.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistPeerTXStateStub.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistPeerTXStateStub.java
index cd17770..5c53df1 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistPeerTXStateStub.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistPeerTXStateStub.java
@@ -259,6 +259,7 @@ public final class DistPeerTXStateStub extends PeerTXStateStub implements
public void postPutAll(DistributedPutAllOperation putallOp,
VersionedObjectList successfulPuts, LocalRegion region) {
super.postPutAll(putallOp, successfulPuts, region);
+ // TODO DISTTX: event is never released
EntryEventImpl event = EntryEventImpl.createPutAllEvent(putallOp, region,
Operation.PUTALL_CREATE, putallOp.getBaseEvent().getKey(), putallOp
.getBaseEvent().getValue());
@@ -271,6 +272,7 @@ public final class DistPeerTXStateStub extends PeerTXStateStub implements
public void postRemoveAll(DistributedRemoveAllOperation removeAllOp,
VersionedObjectList successfulOps, LocalRegion region) {
super.postRemoveAll(removeAllOp, successfulOps, region);
+ // TODO DISTTX: event is never released
EntryEventImpl event = EntryEventImpl.createRemoveAllEvent(removeAllOp,
region, removeAllOp.getBaseEvent().getKey());
event.setEventId(removeAllOp.getBaseEvent().getEventId());
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXState.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXState.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXState.java
index 19cc175..f8475ae 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXState.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXState.java
@@ -44,6 +44,7 @@ import com.gemstone.gemfire.internal.cache.tx.DistTxKeyInfo;
import com.gemstone.gemfire.internal.cache.tx.DistTxEntryEvent;
import com.gemstone.gemfire.internal.cache.versions.RegionVersionVector;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
/**
* TxState on a datanode VM
@@ -568,7 +569,7 @@ public class DistTXState extends TXState {
InternalDistributedMember myId = theRegion.getDistributionManager()
.getDistributionManagerId();
for (int i = 0; i < putallOp.putAllDataSize; ++i) {
- EntryEventImpl ev = PutAllPRMessage.getEventFromEntry(theRegion,
+ @Released EntryEventImpl ev = PutAllPRMessage.getEventFromEntry(theRegion,
myId, myId, i, putallOp.putAllData, false, putallOp
.getBaseEvent().getContext(), false, !putallOp.getBaseEvent()
.isGenerateCallbacks(), false);
@@ -628,10 +629,11 @@ public class DistTXState extends TXState {
InternalDistributedMember myId = theRegion.getDistributionManager()
.getDistributionManagerId();
for (int i = 0; i < op.removeAllDataSize; ++i) {
- EntryEventImpl ev = RemoveAllPRMessage.getEventFromEntry(theRegion,
+ @Released EntryEventImpl ev = RemoveAllPRMessage.getEventFromEntry(theRegion,
myId, myId, i, op.removeAllData, false, op.getBaseEvent()
.getContext(), false, !op.getBaseEvent()
.isGenerateCallbacks());
+ try {
ev.setRemoveAllOperation(op);
// below if condition returns true on secondary when TXState is
// updated in preCommit only on secondary
@@ -664,6 +666,9 @@ public class DistTXState extends TXState {
} catch (EntryNotFoundException ignore) {
}
successfulOps.addKeyAndVersion(op.removeAllData[i].key, null);
+ } finally {
+ ev.release();
+ }
}
}
}, op.getBaseEvent().getEventId());
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXStateOnCoordinator.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXStateOnCoordinator.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXStateOnCoordinator.java
index 10d892a..33bec1c 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXStateOnCoordinator.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXStateOnCoordinator.java
@@ -240,6 +240,7 @@ public final class DistTXStateOnCoordinator extends DistTXState implements
public void postPutAll(DistributedPutAllOperation putallOp,
VersionedObjectList successfulPuts, LocalRegion region) {
super.postPutAll(putallOp, successfulPuts, region);
+ // TODO DISTTX: event is never released
EntryEventImpl event = EntryEventImpl.createPutAllEvent(putallOp, region,
Operation.PUTALL_CREATE, putallOp.getBaseEvent().getKey(), putallOp
.getBaseEvent().getValue());
@@ -252,6 +253,7 @@ public final class DistTXStateOnCoordinator extends DistTXState implements
public void postRemoveAll(DistributedRemoveAllOperation removeAllOp,
VersionedObjectList successfulOps, LocalRegion region) {
super.postRemoveAll(removeAllOp, successfulOps, region);
+ // TODO DISTTX: event is never released
EntryEventImpl event = EntryEventImpl.createRemoveAllEvent(removeAllOp,
region, removeAllOp.getBaseEvent().getKey());
event.setEventId(removeAllOp.getBaseEvent().getEventId());
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXStateProxyImplOnCoordinator.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXStateProxyImplOnCoordinator.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXStateProxyImplOnCoordinator.java
index 75a1df7..67d63b1 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXStateProxyImplOnCoordinator.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistTXStateProxyImplOnCoordinator.java
@@ -896,6 +896,7 @@ public class DistTXStateProxyImplOnCoordinator extends DistTXStateProxyImpl {
DistributedPutAllOperation putAllForBucket =
bucketToPutallMap.get(bucketId);;
if (putAllForBucket == null) {
+ // TODO DISTTX: event is never released
EntryEventImpl event = EntryEventImpl.createPutAllEvent(null, region,
Operation.PUTALL_CREATE, key,
putallOp.putAllData[i].getValue());
@@ -976,6 +977,7 @@ public class DistTXStateProxyImplOnCoordinator extends DistTXStateProxyImpl {
DistributedRemoveAllOperation removeAllForBucket =
bucketToRemoveAllMap.get(bucketId);
if (removeAllForBucket == null) {
+ // TODO DISTTX: event is never released
EntryEventImpl event = EntryEventImpl.createRemoveAllEvent(op, region, key);
event.setEventId(op.removeAllData[i].getEventID());
removeAllForBucket = new DistributedRemoveAllOperation(
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedCacheOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedCacheOperation.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedCacheOperation.java
index 79ae4ee..063f823 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedCacheOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedCacheOperation.java
@@ -74,6 +74,7 @@ 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.offheap.StoredObject;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
import com.gemstone.gemfire.internal.offheap.annotations.Unretained;
import com.gemstone.gemfire.internal.sequencelog.EntryLogger;
@@ -1104,7 +1105,6 @@ public abstract class DistributedCacheOperation {
protected void basicProcess(DistributionManager dm, LocalRegion lclRgn) {
Throwable thr = null;
boolean sendReply = true;
- InternalCacheEvent event = null;
if (logger.isTraceEnabled()) {
logger.trace("DistributedCacheOperation.basicProcess: {}", this);
@@ -1140,7 +1140,7 @@ public abstract class DistributedCacheOperation {
return;
}
- event = createEvent(rgn);
+ @Released InternalCacheEvent event = createEvent(rgn);
try {
boolean isEntry = event.getOperation().isEntry();
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedPutAllOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedPutAllOperation.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedPutAllOperation.java
index 2e83cb8..4b4d4d3 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedPutAllOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedPutAllOperation.java
@@ -61,6 +61,8 @@ import com.gemstone.gemfire.internal.cache.versions.VersionSource;
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/**
* Handles distribution of a Region.putall operation.
@@ -193,8 +195,9 @@ public class DistributedPutAllOperation extends AbstractUpdateOperation
public boolean hasNext() {
return DistributedPutAllOperation.this.putAllDataSize > position;
};
+ @Retained
public Object next() {
- EntryEventImpl ev = getEventForPosition(position);
+ @Retained EntryEventImpl ev = getEventForPosition(position);
position++;
return ev;
};
@@ -214,7 +217,7 @@ public class DistributedPutAllOperation extends AbstractUpdateOperation
}
}
-
+ @Retained
public EntryEventImpl getEventForPosition(int position) {
PutAllEntryData entry = this.putAllData[position];
if (entry == null) {
@@ -224,7 +227,7 @@ public class DistributedPutAllOperation extends AbstractUpdateOperation
return entry.event;
}
LocalRegion region = (LocalRegion)this.event.getRegion();
- EntryEventImpl ev = EntryEventImpl.create(
+ @Retained EntryEventImpl ev = EntryEventImpl.create(
region,
entry.getOp(),
entry.getKey(), null/* value */, this.event.getCallbackArgument(),
@@ -813,13 +816,17 @@ public class DistributedPutAllOperation extends AbstractUpdateOperation
}
FilterRoutingInfo consolidated = new FilterRoutingInfo();
for (int i=0; i<this.putAllData.length; i++) {
- EntryEventImpl ev = getEventForPosition(i);
+ @Released EntryEventImpl ev = getEventForPosition(i);
if (ev != null) {
+ try {
FilterRoutingInfo eventRouting = advisor.adviseFilterRouting(ev, cacheOpRecipients);
if (eventRouting != null) {
consolidated.addFilterInfo(eventRouting);
}
putAllData[i].filterRouting = eventRouting;
+ } finally {
+ ev.release();
+ }
}
}
// we need to create routing information for each PUT event
@@ -1093,11 +1100,12 @@ public class DistributedPutAllOperation extends AbstractUpdateOperation
* basicOperateOnRegion
*/
@Override
+ @Retained
protected InternalCacheEvent createEvent(DistributedRegion rgn)
throws EntryNotFoundException
{
// Gester: We have to specify eventId for the message of MAP
- EntryEventImpl event = EntryEventImpl.create(
+ @Retained EntryEventImpl event = EntryEventImpl.create(
rgn,
Operation.PUTALL_UPDATE /* op */, null /* key */, null/* value */,
this.callbackArg, true /* originRemote */, getSender());
@@ -1133,7 +1141,7 @@ public class DistributedPutAllOperation extends AbstractUpdateOperation
*/
public void doEntryPut(PutAllEntryData entry, DistributedRegion rgn,
boolean requiresRegionContext, boolean fetchFromHDFS, boolean isPutDML) {
- EntryEventImpl ev = PutAllMessage.createEntryEvent(entry, getSender(),
+ @Released EntryEventImpl ev = PutAllMessage.createEntryEvent(entry, getSender(),
this.context, rgn,
requiresRegionContext, this.possibleDuplicate,
this.needsRouting, this.callbackArg, true, skipCallbacks);
@@ -1164,6 +1172,7 @@ public class DistributedPutAllOperation extends AbstractUpdateOperation
* @param callbackArg
* @return the event to be used in applying the element
*/
+ @Retained
public static EntryEventImpl createEntryEvent(PutAllEntryData entry,
InternalDistributedMember sender, ClientProxyMembershipID context,
DistributedRegion rgn, boolean requiresRegionContext,
@@ -1174,7 +1183,7 @@ public class DistributedPutAllOperation extends AbstractUpdateOperation
((KeyWithRegionContext)key).setRegionContext(rgn);
}
EventID evId = entry.getEventID();
- EntryEventImpl ev = EntryEventImpl.create(rgn, entry.getOp(),
+ @Retained EntryEventImpl ev = EntryEventImpl.create(rgn, entry.getOp(),
key, null/* value */, callbackArg,
originRemote, sender, !skipCallbacks,
evId);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRegion.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRegion.java
index 467efc6..addba8e 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRegion.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRegion.java
@@ -121,6 +121,7 @@ import com.gemstone.gemfire.internal.cache.wan.parallel.ConcurrentParallelGatewa
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.offheap.annotations.Released;
import com.gemstone.gemfire.internal.offheap.annotations.Retained;
import com.gemstone.gemfire.internal.sequencelog.RegionLogger;
import com.gemstone.gemfire.internal.util.concurrent.StoppableCountDownLatch;
@@ -2442,13 +2443,12 @@ public class DistributedRegion extends LocalRegion implements
}
long lastModified = 0L;
boolean fromServer = false;
- EntryEventImpl event = null;
+ @Released EntryEventImpl event = null;
@Retained Object result = null;
try {
{
if (this.srp != null) {
- EntryEventImpl holder = EntryEventImpl.createVersionTagHolder();
- try {
+ VersionTagHolder holder = new VersionTagHolder();
Object value = this.srp.get(key, aCallbackArgument, holder);
fromServer = value != null;
if (fromServer) {
@@ -2461,9 +2461,6 @@ public class DistributedRegion extends LocalRegion implements
clientEvent.setVersionTag(holder.getVersionTag());
}
}
- } finally {
- holder.release();
- }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRemoveAllOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRemoveAllOperation.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRemoveAllOperation.java
index 70d39dc..34889aa 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRemoveAllOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/DistributedRemoveAllOperation.java
@@ -55,6 +55,8 @@ import com.gemstone.gemfire.internal.cache.versions.DiskVersionTag;
import com.gemstone.gemfire.internal.cache.versions.VersionSource;
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.logging.LogService;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/**
* Handles distribution of a Region.removeAll operation.
@@ -175,8 +177,9 @@ public class DistributedRemoveAllOperation extends AbstractUpdateOperation // TO
public boolean hasNext() {
return DistributedRemoveAllOperation.this.removeAllDataSize > position;
};
+ @Retained
public Object next() {
- EntryEventImpl ev = getEventForPosition(position);
+ @Retained EntryEventImpl ev = getEventForPosition(position);
position++;
return ev;
};
@@ -196,6 +199,7 @@ public class DistributedRemoveAllOperation extends AbstractUpdateOperation // TO
}
}
+ @Retained
public EntryEventImpl getEventForPosition(int position) {
RemoveAllEntryData entry = this.removeAllData[position];
if (entry == null) {
@@ -205,7 +209,7 @@ public class DistributedRemoveAllOperation extends AbstractUpdateOperation // TO
return entry.event;
}
LocalRegion region = (LocalRegion)this.event.getRegion();
- EntryEventImpl ev = EntryEventImpl.create(
+ @Retained EntryEventImpl ev = EntryEventImpl.create(
region,
entry.getOp(),
entry.getKey(), null/* value */, this.event.getCallbackArgument(),
@@ -568,13 +572,17 @@ public class DistributedRemoveAllOperation extends AbstractUpdateOperation // TO
}
FilterRoutingInfo consolidated = new FilterRoutingInfo();
for (int i=0; i<this.removeAllData.length; i++) {
- EntryEventImpl ev = getEventForPosition(i);
+ @Released EntryEventImpl ev = getEventForPosition(i);
if (ev != null) {
+ try {
FilterRoutingInfo eventRouting = advisor.adviseFilterRouting(ev, cacheOpRecipients);
if (eventRouting != null) {
consolidated.addFilterInfo(eventRouting);
}
removeAllData[i].filterRouting = eventRouting;
+ } finally {
+ ev.release();
+ }
}
}
// we need to create routing information for each PUT event
@@ -838,11 +846,12 @@ public class DistributedRemoveAllOperation extends AbstractUpdateOperation // TO
* basicOperateOnRegion
*/
@Override
+ @Retained
protected InternalCacheEvent createEvent(DistributedRegion rgn)
throws EntryNotFoundException
{
// Gester: We have to specify eventId for the message of MAP
- EntryEventImpl event = EntryEventImpl.create(
+ @Retained EntryEventImpl event = EntryEventImpl.create(
rgn,
Operation.REMOVEALL_DESTROY, null /* key */, null/* value */,
this.callbackArg, true /* originRemote */, getSender());
@@ -878,7 +887,7 @@ public class DistributedRemoveAllOperation extends AbstractUpdateOperation // TO
*/
public void doEntryRemove(RemoveAllEntryData entry, DistributedRegion rgn,
boolean requiresRegionContext) {
- EntryEventImpl ev = RemoveAllMessage.createEntryEvent(entry, getSender(),
+ @Released EntryEventImpl ev = RemoveAllMessage.createEntryEvent(entry, getSender(),
this.context, rgn,
requiresRegionContext, this.possibleDuplicate,
this.needsRouting, this.callbackArg, true, skipCallbacks);
@@ -918,6 +927,7 @@ public class DistributedRemoveAllOperation extends AbstractUpdateOperation // TO
* @param callbackArg
* @return the event to be used in applying the element
*/
+ @Retained
public static EntryEventImpl createEntryEvent(RemoveAllEntryData entry,
InternalDistributedMember sender, ClientProxyMembershipID context,
DistributedRegion rgn, boolean requiresRegionContext,
@@ -928,7 +938,7 @@ public class DistributedRemoveAllOperation extends AbstractUpdateOperation // TO
((KeyWithRegionContext)key).setRegionContext(rgn);
}
EventID evId = entry.getEventID();
- EntryEventImpl ev = EntryEventImpl.create(rgn, entry.getOp(),
+ @Retained EntryEventImpl ev = EntryEventImpl.create(rgn, entry.getOp(),
key, null/* value */, callbackArg,
originRemote, sender, !skipCallbacks,
evId);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EntryEventImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EntryEventImpl.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EntryEventImpl.java
index 3c87654..e3458e7 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EntryEventImpl.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EntryEventImpl.java
@@ -214,29 +214,6 @@ public class EntryEventImpl
}
/**
- * create a new entry event that will be used for conveying version information
- * and anything else of use while processing another event
- * @return the empty event object
- */
- @Retained
- public static EntryEventImpl createVersionTagHolder() {
- return createVersionTagHolder(null);
- }
-
- /**
- * create a new entry event that will be used for conveying version information
- * and anything else of use while processing another event
- * @return the empty event object
- */
- @Retained
- public static EntryEventImpl createVersionTagHolder(VersionTag tag) {
- EntryEventImpl result = new EntryEventImpl();
- result.setVersionTag(tag);
- result.disallowOffHeapValues();
- return result;
- }
-
- /**
* Reads the contents of this message from the given input.
*/
public void fromData(DataInput in) throws IOException, ClassNotFoundException {
@@ -389,18 +366,6 @@ public class EntryEventImpl
public EntryEventImpl(Object key2) {
this.keyInfo = new KeyInfo(key2, null, null);
}
-
- /**
- * This constructor is used to create a bridge event in server-side
- * command classes. Events created with this are not intended to be
- * used in cache operations.
- * @param id the identity of the client's event
- */
- @Retained
- public EntryEventImpl(EventID id) {
- this.eventID = id;
- this.offHeapOk = false;
- }
/**
* Creates and returns an EntryEventImpl. Generates and assigns a bucket id to the
@@ -493,7 +458,7 @@ public class EntryEventImpl
DistributedPutAllOperation putAllOp, LocalRegion region,
Operation entryOp, Object entryKey, @Retained(ENTRY_EVENT_NEW_VALUE) Object entryNewValue)
{
- EntryEventImpl e;
+ @Retained EntryEventImpl e;
if (putAllOp != null) {
EntryEventImpl event = putAllOp.getBaseEvent();
if (event.isBridgeEvent()) {
@@ -515,11 +480,12 @@ public class EntryEventImpl
return e;
}
+ @Retained
protected static EntryEventImpl createRemoveAllEvent(
DistributedRemoveAllOperation op,
LocalRegion region,
Object entryKey) {
- EntryEventImpl e;
+ @Retained EntryEventImpl e;
final Operation entryOp = Operation.REMOVEALL_DESTROY;
if (op != null) {
EntryEventImpl event = op.getBaseEvent();
@@ -2995,6 +2961,7 @@ public class EntryEventImpl
}
/** returns a copy of this event with the additional fields for WAN conflict resolution */
+ @Retained
public TimestampedEntryEvent getTimestampedEvent(
final int newDSID, final int oldDSID,
final long newTimestamp, final long oldTimestamp) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EntryExpiryTask.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EntryExpiryTask.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EntryExpiryTask.java
index cba7802..64b669a 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EntryExpiryTask.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EntryExpiryTask.java
@@ -35,6 +35,7 @@ import com.gemstone.gemfire.cache.Region;
import com.gemstone.gemfire.cache.TimeoutException;
import com.gemstone.gemfire.internal.InternalStatisticsDisabledException;
import com.gemstone.gemfire.internal.logging.LogService;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
public class EntryExpiryTask extends ExpiryTask {
@@ -120,7 +121,7 @@ public class EntryExpiryTask extends ExpiryTask {
RegionEntry re = getCheckedRegionEntry();
Object key = re.getKey();
LocalRegion lr = getLocalRegion();
- EntryEventImpl event = EntryEventImpl.create(
+ @Released EntryEventImpl event = EntryEventImpl.create(
lr, Operation.EXPIRE_DESTROY, key, null,
createExpireEntryCallback(lr, key), false, lr.getMyId());
try {
@@ -142,7 +143,7 @@ public class EntryExpiryTask extends ExpiryTask {
RegionEntry re = getCheckedRegionEntry();
Object key = re.getKey();
LocalRegion lr = getLocalRegion();
- EntryEventImpl event = EntryEventImpl.create(lr,
+ @Released EntryEventImpl event = EntryEventImpl.create(lr,
Operation.EXPIRE_INVALIDATE, key, null,
createExpireEntryCallback(lr, key), false, lr.getMyId());
try {
@@ -162,7 +163,7 @@ public class EntryExpiryTask extends ExpiryTask {
RegionEntry re = getCheckedRegionEntry();
Object key = re.getKey();
LocalRegion lr = getLocalRegion();
- EntryEventImpl event = EntryEventImpl.create(lr,
+ @Released EntryEventImpl event = EntryEventImpl.create(lr,
Operation.EXPIRE_LOCAL_DESTROY, key, null,
createExpireEntryCallback(lr, key), false, lr.getMyId());
try {
@@ -182,7 +183,7 @@ public class EntryExpiryTask extends ExpiryTask {
RegionEntry re = getCheckedRegionEntry();
Object key = re.getKey();
LocalRegion lr = getLocalRegion();
- EntryEventImpl event = EntryEventImpl.create(lr,
+ @Released EntryEventImpl event = EntryEventImpl.create(lr,
Operation.EXPIRE_LOCAL_INVALIDATE, key, null,
createExpireEntryCallback(lr, key), false, lr.getMyId());
try {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EventIDHolder.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EventIDHolder.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EventIDHolder.java
new file mode 100644
index 0000000..6dd8586
--- /dev/null
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/EventIDHolder.java
@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+/**
+ * This class is an event that just holds an EventID.
+ * Unlike other EntryEventImpls this class does not need to be released
+ * since its values are never off-heap.
+ */
+public class EventIDHolder extends EntryEventImpl {
+ /*
+ * This constructor is used to create a bridge event in server-side
+ * command classes. Events created with this are not intended to be
+ * used in cache operations.
+ * @param id the identity of the client's event
+ */
+ public EventIDHolder(EventID id) {
+ setEventId(id);
+ disallowOffHeapValues();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/FilterProfile.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/FilterProfile.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/FilterProfile.java
index 90c4dc0..65e7a84 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/FilterProfile.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/FilterProfile.java
@@ -73,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.offheap.annotations.Released;
import com.gemstone.gemfire.internal.util.concurrent.CopyOnWriteHashMap;
/**
@@ -1173,7 +1174,8 @@ public class FilterProfile implements DataSerializableFixedID {
for (int idx=0; idx < size; idx++) {
PutAllEntryData pEntry = putAllData[idx];
if (pEntry != null) {
- final EntryEventImpl ev = dpao.getEventForPosition(idx);
+ @Released final EntryEventImpl ev = dpao.getEventForPosition(idx);
+ try {
FilterRoutingInfo fri = pEntry.filterRouting;
FilterInfo fi = null;
if (fri != null) {
@@ -1212,6 +1214,9 @@ public class FilterProfile implements DataSerializableFixedID {
fi.setInterestedClientsInv(clientsInv);
}
ev.setLocalFilterInfo(fi);
+ } finally {
+ ev.release();
+ }
}
}
}
@@ -1230,7 +1235,8 @@ public class FilterProfile implements DataSerializableFixedID {
for (int idx=0; idx < size; idx++) {
RemoveAllEntryData pEntry = removeAllData[idx];
if (pEntry != null) {
- final EntryEventImpl ev = op.getEventForPosition(idx);
+ @Released final EntryEventImpl ev = op.getEventForPosition(idx);
+ try {
FilterRoutingInfo fri = pEntry.filterRouting;
FilterInfo fi = null;
if (fri != null) {
@@ -1272,6 +1278,9 @@ public class FilterProfile implements DataSerializableFixedID {
// this.region.getLogWriterI18n().fine("setting event routing to " + fi);
// }
ev.setLocalFilterInfo(fi);
+ } finally {
+ ev.release();
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/HARegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/HARegion.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/HARegion.java
index fa29bf3..3896800 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/HARegion.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/HARegion.java
@@ -47,6 +47,7 @@ import com.gemstone.gemfire.internal.cache.tier.sockets.ClientProxyMembershipID;
import com.gemstone.gemfire.internal.cache.tier.sockets.HAEventWrapper;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
/**
* This region is being implemented to suppress distribution of puts and to
@@ -229,7 +230,7 @@ public final class HARegion extends DistributedRegion
throws TimeoutException, CacheWriterException {
checkReadiness();
- EntryEventImpl event = EntryEventImpl.create(this, Operation.UPDATE, key,
+ @Released EntryEventImpl event = EntryEventImpl.create(this, Operation.UPDATE, key,
value, aCallbackArgument, false, getMyId());
try {
@@ -409,7 +410,7 @@ public final class HARegion extends DistributedRegion
op = Operation.LOCAL_LOAD_UPDATE;
}
- EntryEventImpl event = EntryEventImpl.create(
+ @Released EntryEventImpl event = EntryEventImpl.create(
this, op, key, value,
aCallbackArgument, false, getMyId(), generateCallbacks);
try {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/InitialImageOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/InitialImageOperation.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/InitialImageOperation.java
index a72ca8e..b43ba2d 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/InitialImageOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/InitialImageOperation.java
@@ -931,7 +931,7 @@ public class InitialImageOperation {
.create((byte[])tmpValue);
}
// dummy EntryEvent to pass for SQLF index maintenance
- final EntryEventImpl ev = EntryEventImpl.create(this.region,
+ @Released final EntryEventImpl ev = EntryEventImpl.create(this.region,
Operation.CREATE, null, null, null, true, null, false, false);
try {
ev.setKeyInfo(this.region.getKeyInfo(entry.key,
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/InvalidateOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/InvalidateOperation.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/InvalidateOperation.java
index de6adda..6e1d91e 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/InvalidateOperation.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/InvalidateOperation.java
@@ -34,6 +34,7 @@ import com.gemstone.gemfire.distributed.internal.DistributionManager;
import com.gemstone.gemfire.internal.cache.tier.sockets.ClientProxyMembershipID;
import com.gemstone.gemfire.internal.cache.versions.ConcurrentCacheModificationException;
import com.gemstone.gemfire.internal.logging.LogService;
+import com.gemstone.gemfire.internal.offheap.annotations.Retained;
/**
* Handles distribution messaging for invalidating an entry in a region.
@@ -108,12 +109,13 @@ public class InvalidateOperation extends DistributedCacheOperation
}
@Override
+ @Retained
protected InternalCacheEvent createEvent(DistributedRegion rgn)
throws EntryNotFoundException {
if (rgn.keyRequiresRegionContext()) {
((KeyWithRegionContext)this.key).setRegionContext(rgn);
}
- EntryEventImpl ev = EntryEventImpl.create(
+ @Retained EntryEventImpl ev = EntryEventImpl.create(
rgn, getOperation(), this.key,
null, this.callbackArg, true, getSender());
ev.setEventId(this.eventId);
@@ -194,6 +196,7 @@ public class InvalidateOperation extends DistributedCacheOperation
transient ClientProxyMembershipID context;
@Override
+ @Retained
protected InternalCacheEvent createEvent(DistributedRegion rgn)
throws EntryNotFoundException {
EntryEventImpl event = (EntryEventImpl)super.createEvent(rgn);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/70221350/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
index 3ff48bb..4360b2a 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/LocalRegion.java
@@ -206,6 +206,7 @@ import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
import com.gemstone.gemfire.internal.offheap.OffHeapHelper;
import com.gemstone.gemfire.internal.offheap.ReferenceCountHelper;
import com.gemstone.gemfire.internal.offheap.StoredObject;
+import com.gemstone.gemfire.internal.offheap.annotations.Released;
import com.gemstone.gemfire.internal.offheap.annotations.Retained;
import com.gemstone.gemfire.internal.offheap.annotations.Unretained;
import com.gemstone.gemfire.internal.sequencelog.EntryLogger;
@@ -1119,12 +1120,12 @@ public class LocalRegion extends AbstractRegion
public final void create(Object key, Object value, Object aCallbackArgument)
throws TimeoutException, EntryExistsException, CacheWriterException {
long startPut = CachePerfStats.getStatTime();
- EntryEventImpl event = newCreateEntryEvent(key, value, aCallbackArgument);
+ @Released EntryEventImpl event = newCreateEntryEvent(key, value, aCallbackArgument);
validatedCreate(event, startPut);
- // TODO OFFHEAP: validatedCreate calls freeOffHeapResources
+ // TODO OFFHEAP: validatedCreate calls release
}
- public final void validatedCreate(EntryEventImpl event, long startPut)
+ public final void validatedCreate(@Released EntryEventImpl event, long startPut)
throws TimeoutException, EntryExistsException, CacheWriterException {
try {
@@ -1152,14 +1153,13 @@ public class LocalRegion extends AbstractRegion
}
}
} finally {
-
event.release();
-
}
}
// split into a separate newCreateEntryEvent since SQLFabric may need to
// manipulate event before doing the put (e.g. posDup flag)
+ @Retained
public final EntryEventImpl newCreateEntryEvent(Object key, Object value,
Object aCallbackArgument) {
@@ -1191,16 +1191,16 @@ public class LocalRegion extends AbstractRegion
public final Object destroy(Object key, Object aCallbackArgument)
throws TimeoutException, EntryNotFoundException, CacheWriterException {
- EntryEventImpl event = newDestroyEntryEvent(key, aCallbackArgument);
+ @Released EntryEventImpl event = newDestroyEntryEvent(key, aCallbackArgument);
return validatedDestroy(key, event);
- // TODO OFFHEAP: validatedDestroy calls freeOffHeapResources
+ // TODO OFFHEAP: validatedDestroy calls release
}
/**
* Destroys entry without performing validations. Call this after validating
* key, callback arg, and runtime state.
*/
- public Object validatedDestroy(Object key, EntryEventImpl event)
+ public Object validatedDestroy(Object key, @Released EntryEventImpl event)
throws TimeoutException, EntryNotFoundException, CacheWriterException
{
try {
@@ -1221,6 +1221,7 @@ public class LocalRegion extends AbstractRegion
// split into a separate newDestroyEntryEvent since SQLFabric may need to
// manipulate event before doing the put (e.g. posDup flag)
+ @Retained
public final EntryEventImpl newDestroyEntryEvent(Object key,
Object aCallbackArgument) {
validateKey(key);
@@ -1731,15 +1732,15 @@ public class LocalRegion extends AbstractRegion
public Object put(Object key, Object value, Object aCallbackArgument)
throws TimeoutException, CacheWriterException {
long startPut = CachePerfStats.getStatTime();
- EntryEventImpl event = newUpdateEntryEvent(key, value, aCallbackArgument);
+ @Released EntryEventImpl event = newUpdateEntryEvent(key, value, aCallbackArgument);
//Since Sqlfire directly calls validatedPut, the freeing is done in
// validatedPut
return validatedPut(event, startPut);
- // TODO OFFHEAP: validatedPut calls freeOffHeapResources
+ // TODO OFFHEAP: validatedPut calls release
}
- public final Object validatedPut(EntryEventImpl event, long startPut)
+ public final Object validatedPut(@Released EntryEventImpl event, long startPut)
throws TimeoutException, CacheWriterException {
try {
@@ -1776,6 +1777,7 @@ public class LocalRegion extends AbstractRegion
// split into a separate newUpdateEntryEvent since SQLFabric may need to
// manipulate event before doing the put (e.g. posDup flag)
+ @Retained
public final EntryEventImpl newUpdateEntryEvent(Object key, Object value,
Object aCallbackArgument) {
@@ -1792,7 +1794,7 @@ public class LocalRegion extends AbstractRegion
// was modified to call the other EntryEventImpl constructor so that
// an id will be generated by default. Null was passed in anyway.
// generate EventID
- final EntryEventImpl event = EntryEventImpl.create(
+ @Retained final EntryEventImpl event = EntryEventImpl.create(
this, Operation.UPDATE, key,
value, aCallbackArgument, false, getMyId());
boolean eventReturned = false;
@@ -1808,6 +1810,7 @@ public class LocalRegion extends AbstractRegion
* Creates an EntryEventImpl that is optimized to not fetch data from HDFS.
* This is meant to be used by PUT dml from GemFireXD.
*/
+ @Retained
public final EntryEventImpl newPutEntryEvent(Object key, Object value,
Object aCallbackArgument) {
EntryEventImpl ev = newUpdateEntryEvent(key, value, aCallbackArgument);
@@ -2392,7 +2395,7 @@ public class LocalRegion extends AbstractRegion
protected void validatedInvalidate(Object key, Object aCallbackArgument)
throws TimeoutException, EntryNotFoundException
{
- EntryEventImpl event = EntryEventImpl.create(
+ @Released EntryEventImpl event = EntryEventImpl.create(
this, Operation.INVALIDATE,
key, null, aCallbackArgument, false, getMyId());
try {
@@ -2411,7 +2414,7 @@ public class LocalRegion extends AbstractRegion
validateKey(key);
checkReadiness();
checkForNoAccess();
- EntryEventImpl event = EntryEventImpl.create(
+ @Released EntryEventImpl event = EntryEventImpl.create(
this,
Operation.LOCAL_DESTROY, key, null, aCallbackArgument, false, getMyId());
if (generateEventID()) {
@@ -2482,7 +2485,7 @@ public class LocalRegion extends AbstractRegion
checkReadiness();
checkForNoAccess();
- EntryEventImpl event = EntryEventImpl.create(
+ @Released EntryEventImpl event = EntryEventImpl.create(
this,
Operation.LOCAL_INVALIDATE, key, null/* newValue */, callbackArgument,
false, getMyId());
@@ -3020,7 +3023,7 @@ public class LocalRegion extends AbstractRegion
final Object aCallbackArgument = keyInfo.getCallbackArg();
Object value = null;
boolean fromServer = false;
- EntryEventImpl holder = null;
+ VersionTagHolder holder = null;
/*
* First lets try the server
@@ -3028,13 +3031,9 @@ public class LocalRegion extends AbstractRegion
{
ServerRegionProxy mySRP = getServerProxy();
if (mySRP != null) {
- holder = EntryEventImpl.createVersionTagHolder();
- try {
- value = mySRP.get(key, aCallbackArgument, holder);
- fromServer = value != null;
- } finally {
- holder.release();
- }
+ holder = new VersionTagHolder();
+ value = mySRP.get(key, aCallbackArgument, holder);
+ fromServer = value != null;
}
}
@@ -3085,7 +3084,7 @@ public class LocalRegion extends AbstractRegion
op = Operation.LOCAL_LOAD_UPDATE;
}
- EntryEventImpl event
+ @Released EntryEventImpl event
= EntryEventImpl.create(this, op, key, value, aCallbackArgument,
false, getMyId(), generateCallbacks);
try {
@@ -4434,7 +4433,7 @@ public class LocalRegion extends AbstractRegion
}
checkReadiness();
validateKey(key);
- EntryEventImpl event = EntryEventImpl.create(this, Operation.LOCAL_DESTROY,
+ @Released EntryEventImpl event = EntryEventImpl.create(this, Operation.LOCAL_DESTROY,
key, false, getMyId(), false /* generateCallbacks */, true);
try {
basicDestroy(event,
@@ -5580,7 +5579,7 @@ public class LocalRegion extends AbstractRegion
//to get Hash. If the partitioning column is different from primary key,
//the resolver for Sqlfabric is not able to obtain the hash object used for creation of KeyInfo
- final EntryEventImpl event = EntryEventImpl.create(this, Operation.CREATE, key,
+ @Released final EntryEventImpl event = EntryEventImpl.create(this, Operation.CREATE, key,
value, callbackArg, false /* origin remote */, client.getDistributedMember(),
true /* generateCallbacks */,
eventId);
@@ -5654,7 +5653,7 @@ public class LocalRegion extends AbstractRegion
}
}
- final EntryEventImpl event = EntryEventImpl.create(this, Operation.UPDATE, key,
+ @Released final EntryEventImpl event = EntryEventImpl.create(this, Operation.UPDATE, key,
null /* new value */, callbackArg,
false /* origin remote */, memberId.getDistributedMember(),
true /* generateCallbacks */,
@@ -5801,7 +5800,7 @@ public class LocalRegion extends AbstractRegion
concurrencyConfigurationCheck(versionTag);
// Create an event and put the entry
- EntryEventImpl event =
+ @Released EntryEventImpl event =
EntryEventImpl.create(this,
Operation.INVALIDATE,
key, null /* newValue */,
@@ -5856,7 +5855,7 @@ public class LocalRegion extends AbstractRegion
concurrencyConfigurationCheck(versionTag);
// Create an event and destroy the entry
- EntryEventImpl event =
+ @Released EntryEventImpl event =
EntryEventImpl.create(this,
Operation.DESTROY,
key, null /* newValue */,
@@ -5938,7 +5937,7 @@ public class LocalRegion extends AbstractRegion
}
// Create an event and put the entry
- final EntryEventImpl event = EntryEventImpl.create(this, Operation.DESTROY, key,
+ @Released final EntryEventImpl event = EntryEventImpl.create(this, Operation.DESTROY, key,
null /* new value */, callbackArg,
false /* origin remote */, memberId.getDistributedMember(),
true /* generateCallbacks */,
@@ -5978,7 +5977,7 @@ public class LocalRegion extends AbstractRegion
}
// Create an event and put the entry
- final EntryEventImpl event = EntryEventImpl.create(this, Operation.INVALIDATE, key,
+ @Released final EntryEventImpl event = EntryEventImpl.create(this, Operation.INVALIDATE, key,
null /* new value */, callbackArg,
false /* origin remote */, memberId.getDistributedMember(),
true /* generateCallbacks */,
@@ -6004,7 +6003,7 @@ public class LocalRegion extends AbstractRegion
ClientProxyMembershipID memberId, boolean fromClient, EntryEventImpl clientEvent) {
// Create an event and update version stamp of the entry
- EntryEventImpl event = EntryEventImpl.create(this, Operation.UPDATE_VERSION_STAMP, key,
+ @Released EntryEventImpl event = EntryEventImpl.create(this, Operation.UPDATE_VERSION_STAMP, key,
null /* new value */, null /*callbackArg*/,
false /* origin remote */, memberId.getDistributedMember(),
false /* generateCallbacks */,
@@ -6368,8 +6367,7 @@ public class LocalRegion extends AbstractRegion
protected void notifyTimestampsToGateways(EntryEventImpl event) {
// Create updateTimeStampEvent from event.
- EntryEventImpl updateTimeStampEvent = EntryEventImpl.createVersionTagHolder(event.getVersionTag());
- try {
+ VersionTagHolder updateTimeStampEvent = new VersionTagHolder(event.getVersionTag());
updateTimeStampEvent.setOperation(Operation.UPDATE_VERSION_STAMP);
updateTimeStampEvent.setKeyInfo(event.getKeyInfo());
updateTimeStampEvent.setGenerateCallbacks(false);
@@ -6394,9 +6392,6 @@ public class LocalRegion extends AbstractRegion
updateTimeStampEvent.setRegion(event.getRegion());
notifyGatewaySender(EnumListenerEvent.TIMESTAMP_UPDATE, updateTimeStampEvent);
}
- } finally {
- updateTimeStampEvent.release();
- }
}
@@ -7276,8 +7271,9 @@ public class LocalRegion extends AbstractRegion
* @param key - the key that this event is related to
* @return an event for EVICT_DESTROY
*/
+ @Retained
protected EntryEventImpl generateEvictDestroyEvent(final Object key) {
- EntryEventImpl event = EntryEventImpl.create(
+ @Retained EntryEventImpl event = EntryEventImpl.create(
this, Operation.EVICT_DESTROY, key, null/* newValue */,
null, false, getMyId());
// Fix for bug#36963
@@ -7287,8 +7283,10 @@ public class LocalRegion extends AbstractRegion
event.setFetchFromHDFS(false);
return event;
}
- protected EntryEventImpl generateCustomEvictDestroyEvent(final Object key) {
- EntryEventImpl event = EntryEventImpl.create(
+
+ @Retained
+ protected EntryEventImpl generateCustomEvictDestroyEvent(final Object key) {
+ @Retained EntryEventImpl event = EntryEventImpl.create(
this, Operation.CUSTOM_EVICT_DESTROY, key, null/* newValue */,
null, false, getMyId());
@@ -7307,7 +7305,7 @@ public class LocalRegion extends AbstractRegion
{
checkReadiness();
- final EntryEventImpl event =
+ @Released final EntryEventImpl event =
generateEvictDestroyEvent(entry.getKey());
try {
return mapDestroy(event,
@@ -8207,7 +8205,7 @@ public class LocalRegion extends AbstractRegion
for (Iterator itr = keySet().iterator(); itr.hasNext();) {
try {
//EventID will not be generated by this constructor
- EntryEventImpl event = EntryEventImpl.create(
+ @Released EntryEventImpl event = EntryEventImpl.create(
this, op, itr.next() /*key*/,
null/* newValue */, null/* callbackArg */, rgnEvent.isOriginRemote(),
rgnEvent.getDistributedMember());
@@ -9267,6 +9265,10 @@ public class LocalRegion extends AbstractRegion
class EventDispatcher implements Runnable
{
+ /**
+ * released by the release method
+ */
+ @Retained
InternalCacheEvent event;
EnumListenerEvent op;
@@ -10091,7 +10093,7 @@ public class LocalRegion extends AbstractRegion
// correct events will be delivered to any callbacks we have.
long startPut = CachePerfStats.getStatTime();
validateKey(key);
- EntryEventImpl event = EntryEventImpl.create(
+ @Released EntryEventImpl event = EntryEventImpl.create(
this, Operation.LOCAL_LOAD_CREATE, key, value,
callback, false, getMyId(), true);
try {
@@ -10186,7 +10188,7 @@ public class LocalRegion extends AbstractRegion
callbackArg = new GatewaySenderEventCallbackArgument(callbackArg);
}
- final EntryEventImpl event = EntryEventImpl.create(this, Operation.PUTALL_CREATE, null,
+ @Released final EntryEventImpl event = EntryEventImpl.create(this, Operation.PUTALL_CREATE, null,
null /* new value */, callbackArg,
false /* origin remote */, memberId.getDistributedMember(),
!skipCallbacks /* generateCallbacks */,
@@ -10222,7 +10224,7 @@ public class LocalRegion extends AbstractRegion
callbackArg = new GatewaySenderEventCallbackArgument(callbackArg);
}
- final EntryEventImpl event = EntryEventImpl.create(this, Operation.REMOVEALL_DESTROY, null,
+ @Released final EntryEventImpl event = EntryEventImpl.create(this, Operation.REMOVEALL_DESTROY, null,
null /* new value */, callbackArg,
false /* origin remote */, memberId.getDistributedMember(),
true /* generateCallbacks */,
@@ -10246,7 +10248,7 @@ public class LocalRegion extends AbstractRegion
long startPut = CachePerfStats.getStatTime();
// generateCallbacks == false
- EntryEventImpl event = EntryEventImpl.create(this, Operation.PUTALL_CREATE,
+ @Released EntryEventImpl event = EntryEventImpl.create(this, Operation.PUTALL_CREATE,
null, null, null, true, getMyId(), !skipCallbacks);
try {
DistributedPutAllOperation putAllOp = new DistributedPutAllOperation(event, map.size(), false);
@@ -10394,7 +10396,7 @@ public class LocalRegion extends AbstractRegion
Runnable r = new Runnable() {
public void run() {
int offset = 0;
- EntryEventImpl tagHolder = EntryEventImpl.createVersionTagHolder();
+ VersionTagHolder tagHolder = new VersionTagHolder();
while (iterator.hasNext()) {
stopper.checkCancelInProgress(null);
Map.Entry mapEntry = (Map.Entry)iterator.next();
@@ -10445,11 +10447,7 @@ public class LocalRegion extends AbstractRegion
}
if (!overwritten) {
- try {
- basicEntryPutAll(key, value, dpao, offset, tagHolder);
- } finally {
- tagHolder.release();
- }
+ basicEntryPutAll(key, value, dpao, offset, tagHolder);
}
// now we must check again since the cache may have closed during
// distribution (causing this process to not receive and queue the
@@ -10613,7 +10611,7 @@ public class LocalRegion extends AbstractRegion
Runnable r = new Runnable() {
public void run() {
int offset = 0;
- EntryEventImpl tagHolder = EntryEventImpl.createVersionTagHolder();
+ VersionTagHolder tagHolder = new VersionTagHolder();
while (iterator.hasNext()) {
stopper.checkCancelInProgress(null);
Object key;
@@ -10766,6 +10764,7 @@ public class LocalRegion extends AbstractRegion
// Create a dummy event for the PutAll operation. Always create a
// PutAll operation, even if there is no distribution, so that individual
// events can be tracked and handed off to callbacks in postPutAll
+ // No need for release since disallowOffHeapValues called.
final EntryEventImpl event = EntryEventImpl.create(this,
Operation.PUTALL_CREATE, null, null, callbackArg, true, getMyId());
@@ -10795,6 +10794,7 @@ public class LocalRegion extends AbstractRegion
// Create a dummy event for the removeAll operation. Always create a
// removeAll operation, even if there is no distribution, so that individual
// events can be tracked and handed off to callbacks in postRemoveAll
+ // No need for release since disallowOffHeapValues called.
final EntryEventImpl event = EntryEventImpl.create(this, Operation.REMOVEALL_DESTROY, null,
null/* newValue */, callbackArg, false, getMyId());
event.disallowOffHeapValues();
@@ -10826,7 +10826,7 @@ public class LocalRegion extends AbstractRegion
validateArguments(key, value, null);
// event is marked as a PUTALL_CREATE but if the entry exists it
// will be changed to a PUTALL_UPDATE later on.
- EntryEventImpl event = EntryEventImpl.createPutAllEvent(
+ @Released EntryEventImpl event = EntryEventImpl.createPutAllEvent(
putallOp, this, Operation.PUTALL_CREATE, key, value);
try {
@@ -10864,7 +10864,7 @@ public class LocalRegion extends AbstractRegion
assert op != null;
checkReadiness();
validateKey(key);
- EntryEventImpl event = EntryEventImpl.createRemoveAllEvent(op, this, key);
+ @Released EntryEventImpl event = EntryEventImpl.createRemoveAllEvent(op, this, key);
try {
if (tagHolder != null) {
event.setVersionTag(tagHolder.getVersionTag());
@@ -10918,13 +10918,17 @@ public class LocalRegion extends AbstractRegion
successfulKeys.add(key);
}
for (Iterator it=putallOp.eventIterator(); it.hasNext(); ) {
- EntryEventImpl event = (EntryEventImpl)it.next();
+ @Released EntryEventImpl event = (EntryEventImpl)it.next();
+ try {
if (successfulKeys.contains(event.getKey())) {
EnumListenerEvent op = event.getOperation().isCreate() ? EnumListenerEvent.AFTER_CREATE
: EnumListenerEvent.AFTER_UPDATE;
invokePutCallbacks(op, event, !event.callbacksInvoked() && !event.isPossibleDuplicate(),
false /* We must notify gateways inside RegionEntry lock, NOT here, to preserve the order of events sent by gateways for same key*/);
}
+ } finally {
+ event.release();
+ }
}
}
@@ -10941,11 +10945,15 @@ public class LocalRegion extends AbstractRegion
successfulKeys.add(key);
}
for (Iterator it=op.eventIterator(); it.hasNext(); ) {
- EntryEventImpl event = (EntryEventImpl)it.next();
+ @Released EntryEventImpl event = (EntryEventImpl)it.next();
+ try {
if (successfulKeys.contains(event.getKey())) {
invokeDestroyCallbacks(EnumListenerEvent.AFTER_DESTROY, event, !event.callbacksInvoked() && !event.isPossibleDuplicate(),
false /* We must notify gateways inside RegionEntry lock, NOT here, to preserve the order of events sent by gateways for same key*/);
}
+ } finally {
+ event.release();
+ }
}
}
@@ -11607,8 +11615,7 @@ public class LocalRegion extends AbstractRegion
}
public void destroyRecoveredEntry(Object key) {
- EntryEventImpl event = EntryEventImpl.create(
- this,
+ @Released EntryEventImpl event = EntryEventImpl.create(this,
Operation.LOCAL_DESTROY, key, null, null, false, getMyId(), false);
try {
event.inhibitCacheListenerNotification(true);
@@ -12337,7 +12344,7 @@ public class LocalRegion extends AbstractRegion
// was modified to call the other EntryEventImpl constructor so that
// an id will be generated by default. Null was passed in anyway.
// generate EventID
- EntryEventImpl event = EntryEventImpl.create(
+ @Released EntryEventImpl event = EntryEventImpl.create(
this, Operation.PUT_IF_ABSENT, key,
value, callbackArgument, false, getMyId());
final Object oldValue = null;
@@ -12398,7 +12405,7 @@ public class LocalRegion extends AbstractRegion
if (value == null) {
value = Token.INVALID;
}
- EntryEventImpl event = EntryEventImpl.create(this,
+ @Released EntryEventImpl event = EntryEventImpl.create(this,
Operation.REMOVE,
key,
null, // newValue
@@ -12455,7 +12462,7 @@ public class LocalRegion extends AbstractRegion
validateArguments(key, newValue, callbackArg);
checkReadiness();
checkForLimitedOrNoAccess();
- EntryEventImpl event = EntryEventImpl.create(this,
+ @Released EntryEventImpl event = EntryEventImpl.create(this,
Operation.REPLACE,
key,
newValue,
@@ -12523,7 +12530,7 @@ public class LocalRegion extends AbstractRegion
validateArguments(key, value, callbackArg);
checkReadiness();
checkForLimitedOrNoAccess();
- EntryEventImpl event = EntryEventImpl.create(this,
+ @Released EntryEventImpl event = EntryEventImpl.create(this,
Operation.REPLACE,
key,
value,
@@ -12571,7 +12578,7 @@ public class LocalRegion extends AbstractRegion
callbackArg = new GatewaySenderEventCallbackArgument(callbackArg);
}
}
- final EntryEventImpl event = EntryEventImpl.create(this, Operation.PUT_IF_ABSENT, key,
+ @Released final EntryEventImpl event = EntryEventImpl.create(this, Operation.PUT_IF_ABSENT, key,
null /* new value */, callbackArg,
false /* origin remote */, client.getDistributedMember(),
true /* generateCallbacks */,
@@ -12648,7 +12655,7 @@ public class LocalRegion extends AbstractRegion
callbackArg = new GatewaySenderEventCallbackArgument(callbackArg);
}
}
- final EntryEventImpl event = EntryEventImpl.create(this, Operation.REPLACE, key,
+ @Released final EntryEventImpl event = EntryEventImpl.create(this, Operation.REPLACE, key,
null /* new value */, callbackArg,
false /* origin remote */, client.getDistributedMember(),
true /* generateCallbacks */,
@@ -12705,7 +12712,7 @@ public class LocalRegion extends AbstractRegion
callbackArg = new GatewaySenderEventCallbackArgument(callbackArg);
}
}
- final EntryEventImpl event = EntryEventImpl.create(this, Operation.REPLACE, key,
+ @Released final EntryEventImpl event = EntryEventImpl.create(this, Operation.REPLACE, key,
null /* new value */, callbackArg,
false /* origin remote */, client.getDistributedMember(),
true /* generateCallbacks */,
@@ -12771,7 +12778,7 @@ public class LocalRegion extends AbstractRegion
}
// Create an event and put the entry
- final EntryEventImpl event = EntryEventImpl.create(this, Operation.REMOVE, key,
+ @Released final EntryEventImpl event = EntryEventImpl.create(this, Operation.REMOVE, key,
null /* new value */, callbackArg,
false /* origin remote */, memberId.getDistributedMember(),
true /* generateCallbacks */,