You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bo...@apache.org on 2016/03/17 18:14:38 UTC
[1/3] incubator-geode git commit: GEODE-1095: GatewaySenderEventImpl
now caches serialized value size
Repository: incubator-geode
Updated Branches:
refs/heads/develop 155f87d5f -> 05cd14437
GEODE-1095: GatewaySenderEventImpl now caches serialized value size
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/05cd1443
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/05cd1443
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/05cd1443
Branch: refs/heads/develop
Commit: 05cd14437ecc3391505acf0ef158be3a5dda619f
Parents: ff69aea
Author: Barry Oglesby <bo...@pivotal.io>
Authored: Wed Mar 16 13:40:31 2016 -0700
Committer: Barry Oglesby <bo...@pivotal.io>
Committed: Thu Mar 17 10:11:47 2016 -0700
----------------------------------------------------------------------
.../internal/cache/wan/GatewaySenderEventImpl.java | 16 +++++++++++++---
1 file changed, 13 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/05cd1443/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventImpl.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventImpl.java
index 6f284b5..d1a887f 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventImpl.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventImpl.java
@@ -213,6 +213,10 @@ public class GatewaySenderEventImpl implements
private static final int OP_DETAIL_REMOVEALL = 14;
+ private static final int DEFAULT_SERIALIZED_VALUE_SIZE = -1;
+
+ private volatile int serializedValueSize = DEFAULT_SERIALIZED_VALUE_SIZE;
+
// /**
// * Is this thread in the process of deserializing this event?
// */
@@ -1223,17 +1227,23 @@ public class GatewaySenderEventImpl implements
}
public int getSerializedValueSize() {
+ int localSerializedValueSize = this.serializedValueSize;
+ if (localSerializedValueSize != DEFAULT_SERIALIZED_VALUE_SIZE) {
+ return localSerializedValueSize;
+ }
@Unretained(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
Object vo = this.valueObj;
if (vo instanceof StoredObject) {
- return ((StoredObject) vo).getSizeInBytes();
+ localSerializedValueSize = ((StoredObject) vo).getSizeInBytes();
} else {
if (this.substituteValue != null) {
- return sizeOf(this.substituteValue);
+ localSerializedValueSize = sizeOf(this.substituteValue);
} else {
- return CachedDeserializableFactory.calcMemSize(getSerializedValue());
+ localSerializedValueSize = CachedDeserializableFactory.calcMemSize(getSerializedValue());
}
}
+ this.serializedValueSize = localSerializedValueSize;
+ return localSerializedValueSize;
}
@Override
[2/3] incubator-geode git commit: GEODE-1067: The dispatcher now
handles IllegalStateException by retrying batch without the released event
Posted by bo...@apache.org.
GEODE-1067: The dispatcher now handles IllegalStateException by retrying batch without the released event
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/ff69aeae
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/ff69aeae
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/ff69aeae
Branch: refs/heads/develop
Commit: ff69aeae9615d7263075079e8e5f5180bc350dce
Parents: 5b6a2cd
Author: Barry Oglesby <bo...@pivotal.io>
Authored: Wed Mar 16 12:27:39 2016 -0700
Committer: Barry Oglesby <bo...@pivotal.io>
Committed: Thu Mar 17 10:11:47 2016 -0700
----------------------------------------------------------------------
.../wan/AbstractGatewaySenderEventProcessor.java | 15 +++++++++++++++
.../internal/cache/wan/GatewaySenderEventImpl.java | 7 +++++++
.../wan/GatewaySenderEventRemoteDispatcher.java | 15 +++++++++++----
3 files changed, 33 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ff69aeae/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 51b125a..5020cf2 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
@@ -520,6 +520,21 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
filteredList = new ArrayList<GatewaySenderEventImpl>();
filteredList.addAll(events);
+
+ // If the exception has been set and its cause is an IllegalStateExcetption,
+ // remove all events whose serialized value is no longer available
+ if (this.exception != null && this.exception.getCause() != null
+ && this.exception.getCause() instanceof IllegalStateException) {
+ for (Iterator<GatewaySenderEventImpl> i = filteredList.iterator(); i.hasNext();) {
+ GatewaySenderEventImpl event = i.next();
+ if (event.isSerializedValueNotAvailable()) {
+ i.remove();
+ }
+ }
+ this.exception = null;
+ }
+
+ // Filter the events
for (GatewayEventFilter filter : sender.getGatewayEventFilters()) {
Iterator<GatewaySenderEventImpl> itr = filteredList.iterator();
while (itr.hasNext()) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ff69aeae/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventImpl.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventImpl.java
index d8922f8..6f284b5 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventImpl.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventImpl.java
@@ -131,6 +131,8 @@ public class GatewaySenderEventImpl implements
protected transient Object valueObj;
protected transient boolean valueObjReleased;
+ private transient boolean serializedValueNotAvailable;
+
/**
* Whether the value is a serialized object or just a byte[]
*/
@@ -662,6 +664,10 @@ public class GatewaySenderEventImpl implements
}
}
+ public boolean isSerializedValueNotAvailable() {
+ return this.serializedValueNotAvailable;
+ }
+
/**
* If the value owned of this event is just bytes return that byte array;
* otherwise serialize the value object and return the serialized bytes.
@@ -698,6 +704,7 @@ public class GatewaySenderEventImpl implements
this.value = result;
} else if (result == null) {
if (this.valueObjReleased) {
+ this.serializedValueNotAvailable = true;
throw new IllegalStateException("Value is no longer available. getSerializedValue must be called before processEvents returns.");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/ff69aeae/geode-wan/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventRemoteDispatcher.java
----------------------------------------------------------------------
diff --git a/geode-wan/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventRemoteDispatcher.java b/geode-wan/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventRemoteDispatcher.java
index 22dff3d..ad2be2b 100644
--- a/geode-wan/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventRemoteDispatcher.java
+++ b/geode-wan/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventRemoteDispatcher.java
@@ -163,7 +163,8 @@ public class GatewaySenderEventRemoteDispatcher implements
} else if (t instanceof IOException
|| t instanceof ServerConnectivityException
|| t instanceof ConnectionDestroyedException
- || t instanceof MessageTooLargeException) {
+ || t instanceof MessageTooLargeException
+ || t instanceof IllegalStateException) {
this.processor.handleException();
// If the cause is an IOException or a ServerException, sleep and retry.
// Sleep for a bit and recheck.
@@ -268,6 +269,12 @@ public class GatewaySenderEventRemoteDispatcher implements
LocalizedStrings.GatewayEventRemoteDispatcher_0_EXCEPTION_DURING_PROCESSING_BATCH_1_ON_CONNECTION_2.toLocalizedString(
new Object[] {this, Integer.valueOf(currentBatchId), connection}), ex);
}
+ catch (IllegalStateException e) {
+ this.processor.setException(new GatewaySenderException(e));
+ throw new GatewaySenderException(
+ LocalizedStrings.GatewayEventRemoteDispatcher_0_EXCEPTION_DURING_PROCESSING_BATCH_1_ON_CONNECTION_2.toLocalizedString(
+ new Object[] {this, Integer.valueOf(currentBatchId), connection}), e);
+ }
catch (Exception e) {
// An Exception has occurred. Get its cause.
Throwable t = e.getCause();
@@ -321,7 +328,7 @@ public class GatewaySenderEventRemoteDispatcher implements
if (cache != null && !cache.isClosed()) {
if (this.sender.isPrimary() && (this.connection != null)) {
if (this.ackReaderThread == null || !this.ackReaderThread.isRunning()) {
- this.ackReaderThread = new AckReaderThread(this.sender);
+ this.ackReaderThread = new AckReaderThread(this.sender, this.processor);
this.ackReaderThread.start();
this.ackReaderThread.waitForRunningAckReaderThreadRunningState();
}
@@ -541,8 +548,8 @@ public class GatewaySenderEventRemoteDispatcher implements
private volatile boolean ackReaderThreadRunning = false;
- public AckReaderThread(GatewaySender sender) {
- super("AckReaderThread for : " + sender.getId());
+ public AckReaderThread(GatewaySender sender, AbstractGatewaySenderEventProcessor processor) {
+ super("AckReaderThread for : " + processor.getName());
this.setDaemon(true);
this.cache = (GemFireCacheImpl)((AbstractGatewaySender)sender).getCache();
}
[3/3] incubator-geode git commit: GEODE-670: GatewaySenderEvents are
now sized using Sizeable if appropriate
Posted by bo...@apache.org.
GEODE-670: GatewaySenderEvents are now sized using Sizeable if appropriate
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/5b6a2cdc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/5b6a2cdc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/5b6a2cdc
Branch: refs/heads/develop
Commit: 5b6a2cdc08f521625e0987267201893735877323
Parents: 155f87d
Author: Barry Oglesby <bo...@pivotal.io>
Authored: Wed Mar 16 11:50:44 2016 -0700
Committer: Barry Oglesby <bo...@pivotal.io>
Committed: Thu Mar 17 10:11:47 2016 -0700
----------------------------------------------------------------------
.../java/com/gemstone/gemfire/internal/cache/BucketRegion.java | 6 ++----
1 file changed, 2 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/5b6a2cdc/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 413fc87..2495464 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
@@ -2141,14 +2141,12 @@ implements Bucket
}
static int calcMemSize(Object value) {
- if (value != null && (value instanceof GatewaySenderEventImpl)) {
- return ((GatewaySenderEventImpl)value).getSerializedValueSize();
- }
if (value == null || value instanceof Token) {
return 0;
}
if (!(value instanceof byte[]) && !(value instanceof CachedDeserializable)
- && !(value instanceof com.gemstone.gemfire.Delta) && !(value instanceof Delta)) {
+ && !(value instanceof com.gemstone.gemfire.Delta) && !(value instanceof Delta)
+ && !(value instanceof GatewaySenderEventImpl)) {
// ezoerner:20090401 it's possible this value is a Delta
throw new InternalGemFireError("DEBUG: calcMemSize: weird value (class "
+ value.getClass() + "): " + value);