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 2015/07/09 19:02:50 UTC
[32/57] [partial] incubator-geode git commit: Initial import of
geode-1.0.0.0-SNAPSHOT-2. All the new sub-project directories (like jvsd)
were not imported. A diff was done to confirm that this commit is exactly the
same as the open directory the snapsho
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/GFJGBasicAdapter.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/GFJGBasicAdapter.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/GFJGBasicAdapter.java
index 8a8413b..897d642 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/GFJGBasicAdapter.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/GFJGBasicAdapter.java
@@ -20,6 +20,7 @@ import java.util.Properties;
import com.gemstone.gemfire.DataSerializer;
import com.gemstone.gemfire.ForcedDisconnectException;
import com.gemstone.gemfire.GemFireConfigException;
+import com.gemstone.gemfire.InternalGemFireError;
import com.gemstone.gemfire.SystemConnectException;
import com.gemstone.gemfire.cache.UnsupportedVersionException;
import com.gemstone.gemfire.distributed.DistributedMember;
@@ -40,8 +41,9 @@ import com.gemstone.gemfire.security.AuthenticationFailedException;
import com.gemstone.org.jgroups.Address;
import com.gemstone.org.jgroups.Header;
import com.gemstone.org.jgroups.Message;
-import com.gemstone.org.jgroups.stack.GFBasicAdapter;
+import com.gemstone.org.jgroups.spi.GFBasicAdapter;
import com.gemstone.org.jgroups.stack.IpAddress;
+import com.gemstone.org.jgroups.util.ExternalStrings;
import com.gemstone.org.jgroups.util.GFLogWriter;
import com.gemstone.org.jgroups.util.GemFireTracer;
import com.gemstone.org.jgroups.util.SockCreator;
@@ -86,7 +88,11 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
@Override
public <T> T readObject(DataInput in) throws IOException,
ClassNotFoundException {
- return DataSerializer.readObject(in);
+ try {
+ return DataSerializer.readObject(in);
+ } catch (InternalGemFireError e) {
+ throw new IOException(e.getMessage(), e);
+ }
}
@Override
@@ -138,11 +144,14 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
static final byte CACHE_OP = 1; // GemStoneAddition
static final byte HIGH_PRIORITY = 8; // GemStoneAddition
+
+ static boolean DEBUG_SERIALIZATION = false;
@Override
public void serializeJGMessage(Message msg, DataOutputStream out) throws IOException {
+ GemFireTracer log = GemFireTracer.getLog(GFJGBasicAdapter.class);
//int begIdx = out.size(); // GemStoneAddition
byte leading=0;
@@ -157,6 +166,9 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
out = new VersionedDataOutputStream(out, Version.fromOrdinalNoThrow(serVersion, false));
}
Version.writeOrdinal(out, serVersion, true);
+ if (DEBUG_SERIALIZATION) {
+ log.info("wrote version " + serVersion);
+ }
if(msg.getSrc() != null) {
leading+=SRC_SET;
@@ -172,6 +184,9 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
// 1. write the leading byte first
out.write(leading);
+ if (DEBUG_SERIALIZATION) {
+ log.info("wrote leading byte " + leading);
+ }
// 2. dest_addr
// if(dest_addr != null) {
@@ -185,9 +200,15 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
if(msg.getSrc() != null) {
if(msg.getSrc() instanceof IpAddress) {
InternalDataSerializer.invokeToData(((IpAddress)msg.getSrc()), out); // GemStoneAddition
+ if (DEBUG_SERIALIZATION) {
+ log.info("wrote Ipaddress " + msg.getSrc());
+ }
}
else {
Util.writeAddress(msg.getSrc(), out);
+ if (DEBUG_SERIALIZATION) {
+ log.info("wrote address using Util " + msg.getSrc());
+ }
}
//eidx = out.size();
//if ( (eidx - sidx) > src_addr.size() ) {
@@ -202,20 +223,27 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
if (msg.isHighPriority)
gfFlags += HIGH_PRIORITY;
out.write(gfFlags);
+ if (DEBUG_SERIALIZATION) {
+ log.info("wrote gfFlags " + gfFlags);
+ }
// 4. buf
if(msg.getRawBuffer() != null) {
- out.writeInt(msg.getLength()-msg.getOffset());
- out.write(msg.getRawBuffer(), msg.getOffset(), msg.getLength());
+ int length = msg.getLength();
+ out.writeInt(length);
+ out.write(msg.getRawBuffer(), msg.getOffset(), length);
+ if (DEBUG_SERIALIZATION) {
+ log.info("wrote " + length + " payload bytes");
+ }
}
// 5. headers
short size=(short)msg.getHeaders().size();
out.writeShort(size);
Map.Entry entry;
-// if (log.isTraceEnabled()) {
-// log.trace("writing " + size + " headers");
-// }
+ if (DEBUG_SERIALIZATION) {
+ log.info("writing " + size + " headers");
+ }
//long estSize, startPos, endPos; // GemStoneAddition
for(Iterator it=msg.getHeaders().entrySet().iterator(); it.hasNext();) {
entry=(Map.Entry)it.next();
@@ -234,9 +262,9 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
msg.writeHeader((Header)entry.getValue(), hdos);
hdos.flush();
byte[] headerBytes = baos.toByteArray();
-// if (log.isTraceEnabled()) {
-// log.trace("writing header " + entry.getKey() + " length=" + headerBytes.length);
-// }
+ if (DEBUG_SERIALIZATION) {
+ log.info("writing header " + entry.getKey() + " length=" + headerBytes.length);
+ }
out.writeInt(headerBytes.length);
out.write(headerBytes);
//endPos = out.size();
@@ -257,9 +285,14 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
String hdr_name;
Header hdr = null;
+ GemFireTracer log = GemFireTracer.getLog(GFJGBasicAdapter.class);
+
// create a versioned stream if a different version
// was used to multicast this message
short sv = Version.readOrdinal(in);
+ if (DEBUG_SERIALIZATION) {
+ log.info("read version " + sv);
+ }
msg.setVersion(sv);
if (0 < msg.getVersion() && msg.getVersion() < Version.CURRENT_ORDINAL) {
@@ -272,6 +305,9 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
// 1. read the leading byte first
leading=in.readByte();
+ if (DEBUG_SERIALIZATION) {
+ log.info("read leading byte " + leading);
+ }
// 2. dest_addr
// if((leading & DEST_SET) == DEST_SET) {
@@ -291,6 +327,9 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
IpAddress src_addr=new IpAddress();
try {
InternalDataSerializer.invokeFromData(((IpAddress)src_addr), in);
+ if (DEBUG_SERIALIZATION) {
+ log.info("read src_addr " + src_addr);
+ }
} catch (ClassNotFoundException e) {
throw new IOException(e);
} // GemStoneAddition
@@ -299,11 +338,17 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
}
else {
msg.setSrc(Util.readAddress(in));
+ if (DEBUG_SERIALIZATION) {
+ log.info("read address using Util " + msg.getSrc());
+ }
}
}
// GemStoneAddition
byte gfFlags = in.readByte();
+ if (DEBUG_SERIALIZATION) {
+ log.info("read gfFlags " + gfFlags);
+ }
if ( (gfFlags & CACHE_OP) != 0 )
msg.isCacheOperation = true;
if ( (gfFlags & HIGH_PRIORITY) != 0 )
@@ -312,6 +357,14 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
// 4. buf
if((leading & BUF_SET) == BUF_SET) {
len=in.readInt();
+ if (65535 < len || len < 0) {
+ String sender = msg.getSrc() != null? msg.getSrc().toString() : "(unknown sender)";
+ log.getLogWriter().warning(ExternalStrings.CORRUPTED_MESSAGE_RECEIVED, sender);
+ throw new IOException(ExternalStrings.CORRUPTED_MESSAGE_RECEIVED.toLocalizedString(sender));
+ }
+ if (DEBUG_SERIALIZATION) {
+ log.info("reading " + len + " payload bytes");
+ }
try { // GemStoneAddition -- flag this as a problem
msg.setBuffer(new byte[len]);
}
@@ -330,12 +383,15 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
if (msg.getHeaders() == null) {
msg.setHeaders(msg.createHeaders(len));
}
+ if (DEBUG_SERIALIZATION) {
+ log.info("reading " + len + " headers");
+ }
for(int i=0; i < len; i++) {
hdr_name=in.readUTF();
int hlen = in.readInt();
-// if (log.isTraceEnabled()) {
-// log.trace("reading header " + hdr_name + " of length " + hlen);
-// }
+ if (DEBUG_SERIALIZATION) {
+ log.info("reading header " + hdr_name + " of length " + hlen);
+ }
byte[] headerBytes = new byte[hlen];
in.readFully(headerBytes);
ByteArrayInputStream bais = new ByteArrayInputStream(headerBytes);
@@ -351,12 +407,12 @@ public class GFJGBasicAdapter implements GFBasicAdapter {
}
try {
hdr=msg.readHeader(dis);
-// if (log.isTraceEnabled()) {
-// log.trace("read " + hdr.toString());
-// }
+ if (DEBUG_SERIALIZATION) {
+ log.info("read " + hdr.toString());
+ }
msg.getHeaders().put(hdr_name, hdr);
} catch (Exception e) {
- GemFireTracer log = GemFireTracer.getLog(getClass());
+ log = GemFireTracer.getLog(getClass());
if (log.isErrorEnabled()) {
log.error("Failed to deserialize a header " + hdr_name + " of length " + hlen, e);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/GFJGPeerAdapter.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/GFJGPeerAdapter.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/GFJGPeerAdapter.java
index 764658f..7fd921b 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/GFJGPeerAdapter.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/GFJGPeerAdapter.java
@@ -53,7 +53,7 @@ import com.gemstone.gemfire.security.AuthenticationFailedException;
import com.gemstone.org.jgroups.Address;
import com.gemstone.org.jgroups.Header;
import com.gemstone.org.jgroups.Message;
-import com.gemstone.org.jgroups.stack.GFPeerAdapter;
+import com.gemstone.org.jgroups.spi.GFPeerAdapter;
import com.gemstone.org.jgroups.stack.IpAddress;
import com.gemstone.org.jgroups.util.ExternalStrings;
import com.gemstone.org.jgroups.util.GFLogWriter;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/JGroupMembershipManager.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/JGroupMembershipManager.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/JGroupMembershipManager.java
index 1e153ff..9c2d9fd 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/JGroupMembershipManager.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/jgroup/JGroupMembershipManager.java
@@ -42,11 +42,13 @@ import com.gemstone.gemfire.SystemConnectException;
import com.gemstone.gemfire.SystemFailure;
import com.gemstone.gemfire.ToDataException;
import com.gemstone.gemfire.cache.Cache;
+import com.gemstone.gemfire.cache.server.CacheServer;
import com.gemstone.gemfire.cache.util.BoundedLinkedHashMap;
import com.gemstone.gemfire.distributed.DistributedMember;
import com.gemstone.gemfire.distributed.DistributedSystem;
import com.gemstone.gemfire.distributed.DistributedSystemDisconnectedException;
import com.gemstone.gemfire.distributed.Locator;
+import com.gemstone.gemfire.distributed.internal.DM;
import com.gemstone.gemfire.distributed.internal.DMStats;
import com.gemstone.gemfire.distributed.internal.DSClock;
import com.gemstone.gemfire.distributed.internal.DistributionConfig;
@@ -77,6 +79,7 @@ import com.gemstone.gemfire.internal.Version;
import com.gemstone.gemfire.internal.admin.remote.RemoteTransportConfig;
import com.gemstone.gemfire.internal.cache.DirectReplyMessage;
import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
+import com.gemstone.gemfire.internal.cache.xmlcache.BridgeServerCreation;
import com.gemstone.gemfire.internal.cache.xmlcache.CacheXmlGenerator;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
@@ -110,8 +113,8 @@ import com.gemstone.org.jgroups.protocols.UDP;
import com.gemstone.org.jgroups.protocols.VERIFY_SUSPECT;
import com.gemstone.org.jgroups.protocols.pbcast.GMS;
import com.gemstone.org.jgroups.protocols.pbcast.NAKACK;
-import com.gemstone.org.jgroups.stack.GFBasicAdapter;
-import com.gemstone.org.jgroups.stack.GFPeerAdapter;
+import com.gemstone.org.jgroups.spi.GFBasicAdapter;
+import com.gemstone.org.jgroups.spi.GFPeerAdapter;
import com.gemstone.org.jgroups.stack.GossipServer;
import com.gemstone.org.jgroups.stack.IpAddress;
import com.gemstone.org.jgroups.stack.ProtocolStack;
@@ -1175,7 +1178,9 @@ public class JGroupMembershipManager implements MembershipManager
if (JGroupMembershipManager.this.shutdownInProgress) {
return; // probably a jgroups race condition
}
- saveCacheXmlForReconnect();
+ if (!dconfig.getDisableAutoReconnect()) {
+ saveCacheXmlForReconnect(dconfig.getUseSharedConfiguration());
+ }
// make sure that we've received a connected channel and aren't responsible
// for the notification
if (JGroupMembershipManager.this.puller != null // not shutting down
@@ -1212,6 +1217,13 @@ public class JGroupMembershipManager implements MembershipManager
// }
}
}
+ InternalDistributedSystem system = InternalDistributedSystem.getAnyInstance();
+ if (system != null) {
+ DM dm = system.getDM();
+ if (dm != null) {
+ dm.setRootCause(e);
+ }
+ }
JGroupMembershipManager.this.uncleanShutdown(reason, e);
}
}
@@ -2565,6 +2577,8 @@ public class JGroupMembershipManager implements MembershipManager
// stubToMemberMap.clear();
// memberToStubMap.clear();
+ this.timer.cancel();
+
if (DEBUG) {
System.err.println("DEBUG: done closing JGroupMembershipManager");
}
@@ -2684,28 +2698,32 @@ public class JGroupMembershipManager implements MembershipManager
}
/** generate XML for the cache before shutting down due to forced disconnect */
- private void saveCacheXmlForReconnect() {
- if (!dconfig.getDisableAutoReconnect()) {
- // first save the current cache description so reconnect can rebuild the cache
- GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
- if (cache != null && (cache instanceof Cache)) {
- if (!Boolean.getBoolean("gemfire.autoReconnect-useCacheXMLFile")
- && !cache.isSqlfSystem() && !dconfig.getUseSharedConfiguration()) {
- try {
- if (logger.isDebugEnabled()) {
- logger.debug("generating XML to rebuild the cache after reconnect completes");
- }
- StringPrintWriter pw = new StringPrintWriter();
- CacheXmlGenerator.generate((Cache)cache, pw, true, false);
- String cacheXML = pw.toString();
- cache.getCacheConfig().setCacheXMLDescription(cacheXML);
- if (logger.isDebugEnabled()) {
- logger.debug("XML generation completed: {}", cacheXML);
- }
- } catch (CancelException e) {
- logger.info(LocalizedMessage.create(LocalizedStrings.JGroupMembershipManager_PROBLEM_GENERATING_CACHE_XML), e);
- }
+ public void saveCacheXmlForReconnect(boolean sharedConfigEnabled) {
+ // first save the current cache description so reconnect can rebuild the cache
+ GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
+ if (cache != null && (cache instanceof Cache)) {
+ if (!Boolean.getBoolean("gemfire.autoReconnect-useCacheXMLFile")
+ && !cache.isSqlfSystem() && !sharedConfigEnabled) {
+ try {
+ logger.info("generating XML to rebuild the cache after reconnect completes");
+ StringPrintWriter pw = new StringPrintWriter();
+ CacheXmlGenerator.generate((Cache)cache, pw, true, false);
+ String cacheXML = pw.toString();
+ cache.getCacheConfig().setCacheXMLDescription(cacheXML);
+ logger.info("XML generation completed: {}", cacheXML);
+ } catch (CancelException e) {
+ logger.info(LocalizedMessage.create(LocalizedStrings.JGroupMembershipManager_PROBLEM_GENERATING_CACHE_XML), e);
+ }
+ } else if (sharedConfigEnabled && !cache.getCacheServers().isEmpty()) {
+ // we need to retain a cache-server description if this JVM was started by gfsh
+ List<BridgeServerCreation> list = new ArrayList<BridgeServerCreation>(cache.getCacheServers().size());
+ for (Iterator it = cache.getCacheServers().iterator(); it.hasNext(); ) {
+ CacheServer cs = (CacheServer)it.next();
+ BridgeServerCreation bsc = new BridgeServerCreation(cache, cs);
+ list.add(bsc);
}
+ cache.getCacheConfig().setCacheServerCreation(list);
+ logger.info("CacheServer configuration saved");
}
}
}
@@ -2771,7 +2789,9 @@ public class JGroupMembershipManager implements MembershipManager
}
}
}
- saveCacheXmlForReconnect();
+ if (!dconfig.getDisableAutoReconnect()) {
+ saveCacheXmlForReconnect(dconfig.getUseSharedConfiguration());
+ }
listener.membershipFailure("Channel closed", problem);
throw new DistributedSystemDisconnectedException("Channel closed", problem);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/streaming/StreamingOperation.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/streaming/StreamingOperation.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/streaming/StreamingOperation.java
index f0ab757..8520819 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/streaming/StreamingOperation.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/streaming/StreamingOperation.java
@@ -26,8 +26,13 @@ import com.gemstone.gemfire.GemFireRethrowable;
import com.gemstone.gemfire.InternalGemFireError;
import com.gemstone.gemfire.InternalGemFireException;
import com.gemstone.gemfire.SystemFailure;
+import com.gemstone.gemfire.cache.query.Struct;
import com.gemstone.gemfire.cache.query.internal.DefaultQuery;
+import com.gemstone.gemfire.cache.query.internal.PRQueryTraceInfo;
import com.gemstone.gemfire.cache.query.internal.QueryMonitor;
+import com.gemstone.gemfire.cache.query.internal.StructImpl;
+import com.gemstone.gemfire.cache.query.internal.types.StructTypeImpl;
+import com.gemstone.gemfire.cache.query.types.ObjectType;
import com.gemstone.gemfire.distributed.internal.DM;
import com.gemstone.gemfire.distributed.internal.DistributionManager;
import com.gemstone.gemfire.distributed.internal.DistributionMessage;
@@ -39,6 +44,8 @@ import com.gemstone.gemfire.distributed.internal.ReplyMessage;
import com.gemstone.gemfire.distributed.internal.ReplyProcessor21;
import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
import com.gemstone.gemfire.internal.HeapDataOutputStream;
+import com.gemstone.gemfire.internal.InternalDataSerializer;
+import com.gemstone.gemfire.internal.Version;
import com.gemstone.gemfire.internal.cache.PartitionedRegionQueryEvaluator;
import com.gemstone.gemfire.internal.cache.Token;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
@@ -473,6 +480,7 @@ public abstract class StreamingOperation {
public int getDSFID() {
return STREAMING_REPLY_MESSAGE;
}
+
@Override
public void fromData(DataInput in) throws IOException, ClassNotFoundException {
@@ -482,6 +490,8 @@ public abstract class StreamingOperation {
this.msgNum = in.readInt();
this.lastMsg = in.readBoolean();
this.pdxReadSerialized = in.readBoolean();
+ Version senderVersion = InternalDataSerializer.getVersionForDataStream(in);
+ boolean isSenderAbove_8_1 = senderVersion.compareTo(Version.GFE_81) > 0;
if (n == -1) {
this.objectList = null;
}
@@ -496,6 +506,12 @@ public abstract class StreamingOperation {
try {
ReplyProcessor21 messageProcessor = ReplyProcessor21.getProcessor(processorId);
boolean isQueryMessageProcessor = messageProcessor instanceof PartitionedRegionQueryEvaluator.StreamingQueryPartitionResponse;
+ ObjectType elementType = null;
+ if(isQueryMessageProcessor) {
+ elementType = ((PartitionedRegionQueryEvaluator.
+ StreamingQueryPartitionResponse)messageProcessor).getResultType();
+ }
+
boolean lowMemoryDetected = false;
for (int i = 0; i < n; i++) {
//TestHook used in ResourceManagerWithQueryMonitorDUnitTest.
@@ -508,6 +524,15 @@ public abstract class StreamingOperation {
break;
}
Object o = DataSerializer.readObject(in);
+ if(isQueryMessageProcessor && elementType != null && elementType.isStructType()) {
+ boolean convertToStruct = isSenderAbove_8_1 ;
+ if(convertToStruct && i == 0) {
+ convertToStruct = !(o instanceof PRQueryTraceInfo);
+ }
+ if(convertToStruct) {
+ o = new StructImpl( (StructTypeImpl)elementType, (Object[])o);
+ }
+ }
this.objectList.add(o);
}
if (lowMemoryDetected) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/AvailablePort.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/AvailablePort.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/AvailablePort.java
index 0cb703d..29e1f36 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/AvailablePort.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/AvailablePort.java
@@ -343,7 +343,10 @@ public class AvailablePort {
while (true) {
int port = getRandomWildcardBindPortNumber();
if (isPortAvailable(port, protocol, addr)) {
- return port;
+ // don't return the products default multicast port
+ if ( !(protocol == JGROUPS && port == DistributionConfig.DEFAULT_MCAST_PORT) ){
+ return port;
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Banner.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Banner.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Banner.java
index 76e484f..d4c1e32 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Banner.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Banner.java
@@ -11,6 +11,7 @@ package com.gemstone.gemfire.internal;
import com.gemstone.gemfire.SystemFailure;
import com.gemstone.gemfire.distributed.internal.DistributionConfig;
import com.gemstone.gemfire.distributed.internal.DistributionConfigImpl;
+import com.gemstone.gemfire.internal.logging.LogService;
import java.io.*;
import java.lang.management.ManagementFactory;
@@ -134,6 +135,8 @@ public class Banner {
out.println(" " + key + " = " + "********");
}
}
+ out.println("Log4J 2 Configuration:");
+ out.println(" " + LogService.getConfigInformation());
}
out.println(SEPERATOR);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/ByteBufferWriter.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/ByteBufferWriter.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/ByteBufferWriter.java
new file mode 100644
index 0000000..05849c1
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/ByteBufferWriter.java
@@ -0,0 +1,19 @@
+package com.gemstone.gemfire.internal;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Used by a couple of our classes to say they can have
+ * a ByteBuffer written to them.
+ * @author dschneider
+ */
+public interface ByteBufferWriter {
+ /**
+ * Writes bb.position()..bb.limit() bytes to this writer.
+ * Note that some implementations of this interface will
+ * keep a reference to bb so callers should expect to give
+ * up ownership of bb and should not modify it after calling
+ * this method.
+ */
+ public void write(ByteBuffer bb);
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/DSFIDFactory.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/DSFIDFactory.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/DSFIDFactory.java
index 78afd51..3c33553 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/DSFIDFactory.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/DSFIDFactory.java
@@ -42,17 +42,13 @@ import com.gemstone.gemfire.cache.client.internal.locator.LocatorStatusRequest;
import com.gemstone.gemfire.cache.client.internal.locator.LocatorStatusResponse;
import com.gemstone.gemfire.cache.client.internal.locator.QueueConnectionRequest;
import com.gemstone.gemfire.cache.client.internal.locator.QueueConnectionResponse;
-import com.gemstone.gemfire.cache.client.internal.locator.wan.LocatorJoinMessage;
-import com.gemstone.gemfire.cache.client.internal.locator.wan.RemoteLocatorJoinRequest;
-import com.gemstone.gemfire.cache.client.internal.locator.wan.RemoteLocatorJoinResponse;
-import com.gemstone.gemfire.cache.client.internal.locator.wan.RemoteLocatorPingRequest;
-import com.gemstone.gemfire.cache.client.internal.locator.wan.RemoteLocatorPingResponse;
-import com.gemstone.gemfire.cache.client.internal.locator.wan.RemoteLocatorRequest;
-import com.gemstone.gemfire.cache.client.internal.locator.wan.RemoteLocatorResponse;
+import com.gemstone.gemfire.cache.hdfs.internal.HDFSGatewayEventImpl;
import com.gemstone.gemfire.cache.query.QueryService;
import com.gemstone.gemfire.cache.query.internal.CqEntry;
+import com.gemstone.gemfire.cache.query.internal.CumulativeNonDistinctResults;
import com.gemstone.gemfire.cache.query.internal.LinkedResultSet;
import com.gemstone.gemfire.cache.query.internal.LinkedStructSet;
+import com.gemstone.gemfire.cache.query.internal.NWayMergeResults;
import com.gemstone.gemfire.cache.query.internal.NullToken;
import com.gemstone.gemfire.cache.query.internal.PRQueryTraceInfo;
import com.gemstone.gemfire.cache.query.internal.ResultsBag;
@@ -166,6 +162,7 @@ import com.gemstone.gemfire.internal.admin.remote.RevokePersistentIDRequest;
import com.gemstone.gemfire.internal.admin.remote.RevokePersistentIDResponse;
import com.gemstone.gemfire.internal.admin.remote.RootRegionRequest;
import com.gemstone.gemfire.internal.admin.remote.RootRegionResponse;
+import com.gemstone.gemfire.internal.admin.remote.ShutdownAllGatewayHubsRequest;
import com.gemstone.gemfire.internal.admin.remote.ShutdownAllRequest;
import com.gemstone.gemfire.internal.admin.remote.ShutdownAllResponse;
import com.gemstone.gemfire.internal.admin.remote.SnapshotResultMessage;
@@ -193,6 +190,9 @@ import com.gemstone.gemfire.internal.cache.CreateRegionProcessor;
import com.gemstone.gemfire.internal.cache.DestroyOperation;
import com.gemstone.gemfire.internal.cache.DestroyPartitionedRegionMessage;
import com.gemstone.gemfire.internal.cache.DestroyRegionOperation;
+import com.gemstone.gemfire.internal.cache.DistTXPrecommitMessage;
+import com.gemstone.gemfire.internal.cache.DistTXCommitMessage;
+import com.gemstone.gemfire.internal.cache.DistTXRollbackMessage;
import com.gemstone.gemfire.internal.cache.DistributedClearOperation.ClearRegionMessage;
import com.gemstone.gemfire.internal.cache.DistributedClearOperation.ClearRegionWithContextMessage;
import com.gemstone.gemfire.internal.cache.DistributedPutAllOperation.EntryVersionsList;
@@ -241,6 +241,7 @@ import com.gemstone.gemfire.internal.cache.RemoteRemoveAllMessage;
import com.gemstone.gemfire.internal.cache.RoleEventImpl;
import com.gemstone.gemfire.internal.cache.SearchLoadAndWriteProcessor;
import com.gemstone.gemfire.internal.cache.SendQueueOperation.SendQueueMessage;
+import com.gemstone.gemfire.internal.cache.ServerPingMessage;
import com.gemstone.gemfire.internal.cache.StateFlushOperation.StateMarkerMessage;
import com.gemstone.gemfire.internal.cache.StateFlushOperation.StateStabilizationMessage;
import com.gemstone.gemfire.internal.cache.StateFlushOperation.StateStabilizedMessage;
@@ -250,6 +251,7 @@ import com.gemstone.gemfire.internal.cache.TXCommitMessage.CommitProcessForLockI
import com.gemstone.gemfire.internal.cache.TXCommitMessage.CommitProcessForTXIdMessage;
import com.gemstone.gemfire.internal.cache.TXCommitMessage.CommitProcessQueryMessage;
import com.gemstone.gemfire.internal.cache.TXCommitMessage.CommitProcessQueryReplyMessage;
+import com.gemstone.gemfire.internal.cache.TXEntryState;
import com.gemstone.gemfire.internal.cache.TXId;
import com.gemstone.gemfire.internal.cache.TXManagerImpl;
import com.gemstone.gemfire.internal.cache.TXRemoteCommitMessage;
@@ -285,6 +287,7 @@ import com.gemstone.gemfire.internal.cache.partitioned.CreateBucketMessage.Creat
import com.gemstone.gemfire.internal.cache.partitioned.DeposePrimaryBucketMessage;
import com.gemstone.gemfire.internal.cache.partitioned.DeposePrimaryBucketMessage.DeposePrimaryBucketReplyMessage;
import com.gemstone.gemfire.internal.cache.partitioned.DestroyMessage;
+import com.gemstone.gemfire.internal.cache.partitioned.DestroyRegionOnDataStoreMessage;
import com.gemstone.gemfire.internal.cache.partitioned.DumpAllPRConfigMessage;
import com.gemstone.gemfire.internal.cache.partitioned.DumpB2NRegion;
import com.gemstone.gemfire.internal.cache.partitioned.DumpB2NRegion.DumpB2NReplyMessage;
@@ -368,6 +371,7 @@ import com.gemstone.gemfire.internal.cache.tier.sockets.ObjectPartList651;
import com.gemstone.gemfire.internal.cache.tier.sockets.RemoveClientFromBlacklistMessage;
import com.gemstone.gemfire.internal.cache.tier.sockets.SerializedObjectPartList;
import com.gemstone.gemfire.internal.cache.tier.sockets.VersionedObjectList;
+import com.gemstone.gemfire.internal.cache.tx.DistTxEntryEvent;
import com.gemstone.gemfire.internal.cache.versions.DiskRegionVersionVector;
import com.gemstone.gemfire.internal.cache.versions.DiskVersionTag;
import com.gemstone.gemfire.internal.cache.versions.VMRegionVersionVector;
@@ -529,6 +533,12 @@ public final class DSFIDFactory implements DataSerializableFixedID {
registerDSFID(PR_REMOVE_ALL_REPLY_MESSAGE, RemoveAllPRMessage.RemoveAllReplyMessage.class);
registerDSFID(REMOTE_REMOVE_ALL_MESSAGE, RemoteRemoveAllMessage.class);
registerDSFID(REMOTE_REMOVE_ALL_REPLY_MESSAGE, RemoteRemoveAllMessage.RemoveAllReplyMessage.class);
+ registerDSFID(DISTTX_ROLLBACK_MESSAGE, DistTXRollbackMessage.class);
+ registerDSFID(DISTTX_COMMIT_MESSAGE, DistTXCommitMessage.class);
+ registerDSFID(DISTTX_PRE_COMMIT_MESSAGE, DistTXPrecommitMessage.class);
+ registerDSFID(DISTTX_ROLLBACK_REPLY_MESSAGE, DistTXRollbackMessage.DistTXRollbackReplyMessage.class);
+ registerDSFID(DISTTX_COMMIT_REPLY_MESSAGE, DistTXCommitMessage.DistTXCommitReplyMessage.class);
+ registerDSFID(DISTTX_PRE_COMMIT_REPLY_MESSAGE, DistTXPrecommitMessage.DistTXPrecommitReplyMessage.class);
registerDSFID(PR_PUT_MESSAGE, PutMessage.class);
registerDSFID(INVALIDATE_MESSAGE,
InvalidateOperation.InvalidateMessage.class);
@@ -613,6 +623,8 @@ public final class DSFIDFactory implements DataSerializableFixedID {
registerDSFID(RESULTS_SET, ResultsSet.class);
registerDSFID(SORTED_RESULT_SET, SortedResultSet.class);
registerDSFID(SORTED_STRUCT_SET, SortedStructSet.class);
+ registerDSFID(NWAY_MERGE_RESULTS, NWayMergeResults.class);
+ registerDSFID(CUMULATIVE_RESULTS, CumulativeNonDistinctResults.class);
registerDSFID(UNDEFINED, Undefined.class);
registerDSFID(STRUCT_IMPL, StructImpl.class);
registerDSFID(STRUCT_SET, StructSet.class);
@@ -985,6 +997,8 @@ public final class DSFIDFactory implements DataSerializableFixedID {
RemoteFetchVersionMessage.FetchVersionReplyMessage.class);
registerDSFID(RELEASE_CLEAR_LOCK_MESSAGE, ReleaseClearLockMessage.class);
registerDSFID(PR_TOMBSTONE_MESSAGE, PRTombstoneMessage.class);
+ registerDSFID(HDFS_GATEWAY_EVENT_IMPL, HDFSGatewayEventImpl.class);
+
registerDSFID(REQUEST_RVV_MESSAGE, InitialImageOperation.RequestRVVMessage.class);
registerDSFID(RVV_REPLY_MESSAGE, InitialImageOperation.RVVReplyMessage.class);
registerDSFID(SNAPPY_COMPRESSED_CACHED_DESERIALIZABLE, SnappyCompressedCachedDeserializable.class);
@@ -995,6 +1009,14 @@ public final class DSFIDFactory implements DataSerializableFixedID {
registerDSFID(PR_FETCH_BULK_ENTRIES_REPLY_MESSAGE, FetchBulkEntriesReplyMessage.class);
registerDSFID(PR_QUERY_TRACE_INFO, PRQueryTraceInfo.class);
registerDSFID(INDEX_CREATION_DATA, IndexCreationData.class);
+ registerDSFID(DIST_TX_OP, DistTxEntryEvent.class);
+ registerDSFID(DIST_TX_PRE_COMMIT_RESPONSE, DistTXPrecommitMessage.DistTxPrecommitResponse.class);
+ registerDSFID(DIST_TX_THIN_ENTRY_STATE, TXEntryState.DistTxThinEntryState.class);
+ registerDSFID(SERVER_PING_MESSAGE, ServerPingMessage.class);
+ registerDSFID(PR_DESTROY_ON_DATA_STORE_MESSAGE,
+ DestroyRegionOnDataStoreMessage.class);
+ registerDSFID(SHUTDOWN_ALL_GATEWAYHUBS_REQUEST,
+ ShutdownAllGatewayHubsRequest.class);
}
/**
@@ -1046,6 +1068,8 @@ public final class DSFIDFactory implements DataSerializableFixedID {
return readConfigurationRequest(in);
case CONFIGURATION_RESPONSE:
return readConfigurationResponse(in);
+ case PR_DESTROY_ON_DATA_STORE_MESSAGE:
+ return readDestroyOnDataStore(in);
default:
final Constructor<?> cons;
if (dsfid >= Byte.MIN_VALUE && dsfid <= Byte.MAX_VALUE) {
@@ -1121,6 +1145,12 @@ public final class DSFIDFactory implements DataSerializableFixedID {
return (InterestResultPolicyImpl)InterestResultPolicy.fromOrdinal(ordinal);
}
+ private static DataSerializableFixedID readDestroyOnDataStore(DataInput in)
+ throws IOException, ClassNotFoundException {
+ DataSerializableFixedID serializable = new DestroyRegionOnDataStoreMessage();
+ serializable.fromData(in);
+ return serializable;
+ }
/**
* Map for SQLFabric specific classIds to the {@link Class} of an
* implementation. We maintain this separate map for SQLFabric to allow
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/DataSerializableFixedID.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/DataSerializableFixedID.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/DataSerializableFixedID.java
index 9ad64b9..a8a1715 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/DataSerializableFixedID.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/DataSerializableFixedID.java
@@ -73,6 +73,7 @@ public interface DataSerializableFixedID extends SerializationVersions {
case FOO:
return new FOO(in);
*/
+ public static final short HDFS_GATEWAY_EVENT_IMPL = -141;
public static final short SNAPPY_COMPRESSED_CACHED_DESERIALIZABLE = -140;
public static final short GATEWAY_EVENT_IMPL = -136;
@@ -284,7 +285,10 @@ public interface DataSerializableFixedID extends SerializationVersions {
public static final byte PR_REMOVE_ALL_REPLY_MESSAGE = -7;
public static final byte REMOTE_REMOVE_ALL_MESSAGE = -6;
public static final byte REMOTE_REMOVE_ALL_REPLY_MESSAGE = -5;
- // -4..-1 unused
+ public static final byte DISTTX_COMMIT_MESSAGE = -4;
+ public static final byte DISTTX_PRE_COMMIT_MESSAGE = -3;
+ public static final byte DISTTX_COMMIT_REPLY_MESSAGE = -2;
+ public static final byte DISTTX_PRE_COMMIT_REPLY_MESSAGE = -1;
public static final byte ILLEGAL = 0;
@@ -579,7 +583,8 @@ public interface DataSerializableFixedID extends SerializationVersions {
public static final short TOKEN_REMOVED2 = 145;
public static final short STARTUP_RESPONSE_WITHVERSION_MESSAGE = 146;
- //147-148 unused
+ public static final short SHUTDOWN_ALL_GATEWAYHUBS_REQUEST = 147;
+
public static final short TOKEN_TOMBSTONE = 149;
public static final short PR_DESTROY_REPLY_MESSAGE = 150;
public static final short R_DESTROY_REPLY_MESSAGE = 151;
@@ -606,7 +611,11 @@ public interface DataSerializableFixedID extends SerializationVersions {
public static final short TX_COMMIT_MESSAGE_701 = 164;
public static final short PR_FETCH_BULK_ENTRIES_MESSAGE = 165;
public static final short PR_FETCH_BULK_ENTRIES_REPLY_MESSAGE = 166;
- // 167..999 unused
+ public static final short NWAY_MERGE_RESULTS = 167;
+ public static final short CUMULATIVE_RESULTS = 168;
+ public static final short DISTTX_ROLLBACK_MESSAGE = 169;
+ public static final short DISTTX_ROLLBACK_REPLY_MESSAGE = 170;
+ // 171..999 unused
public static final short ADD_HEALTH_LISTENER_REQUEST = 1000;
public static final short ADD_HEALTH_LISTENER_RESPONSE = 1001;
@@ -792,7 +801,13 @@ public interface DataSerializableFixedID extends SerializationVersions {
public static final short PR_QUERY_TRACE_INFO = 2162;
public static final short INDEX_CREATION_DATA = 2163;
+
+ public static final short SERVER_PING_MESSAGE = 2164;
+ public static final short PR_DESTROY_ON_DATA_STORE_MESSAGE = 2165;
+ public static final short DIST_TX_OP = 2166;
+ public static final short DIST_TX_PRE_COMMIT_RESPONSE = 2167;
+ public static final short DIST_TX_THIN_ENTRY_STATE = 2168;
// NOTE, codes > 65535 will take 4 bytes to serialize
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/HeapDataOutputStream.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/HeapDataOutputStream.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/HeapDataOutputStream.java
index bb00cbd..dd0c151 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/HeapDataOutputStream.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/HeapDataOutputStream.java
@@ -18,9 +18,13 @@ import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.LinkedList;
+import org.apache.logging.log4j.Logger;
+
import com.gemstone.gemfire.DataSerializer;
import com.gemstone.gemfire.internal.cache.BytesAndBitsForCompactor;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
+import com.gemstone.gemfire.internal.logging.LogService;
+import com.gemstone.gemfire.internal.tcp.ByteBufferInputStream.ByteSource;
/** HeapDataOutputStream is an OutputStream that also implements DataOutput
* and stores all data written to it in heap memory.
@@ -46,7 +50,8 @@ import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
* }
*/
public class HeapDataOutputStream extends OutputStream implements
- ObjToByteArraySerializer, VersionedDataStream {
+ ObjToByteArraySerializer, VersionedDataStream, ByteBufferWriter {
+ private static final Logger logger = LogService.getLogger();
ByteBuffer buffer;
protected LinkedList<ByteBuffer> chunks = null;
protected int size = 0;
@@ -62,6 +67,7 @@ public class HeapDataOutputStream extends OutputStream implements
private Error expansionException = null;
private int memoPosition;
private Version version;
+ private boolean doNotCopy;
private static final int INITIAL_CAPACITY = 1024;
@@ -82,10 +88,18 @@ public class HeapDataOutputStream extends OutputStream implements
}
this.MIN_CHUNK_SIZE = INITIAL_CAPACITY;
this.buffer = ByteBuffer.allocate(maxStrBytes);
+ this.doNotCopy = false;
writeUTFNoLength(s);
}
public HeapDataOutputStream(int allocSize, Version version) {
+ this(allocSize, version, false);
+ }
+ /**
+ * @param doNotCopy if true then byte arrays/buffers/sources will not
+ * be copied to this hdos but instead referenced.
+ */
+ public HeapDataOutputStream(int allocSize, Version version, boolean doNotCopy) {
if (allocSize < 32) {
this.MIN_CHUNK_SIZE = 32;
} else {
@@ -93,6 +107,26 @@ public class HeapDataOutputStream extends OutputStream implements
}
this.buffer = ByteBuffer.allocate(allocSize);
this.version = version;
+ this.doNotCopy = doNotCopy;
+ }
+
+ /**
+ * @param doNotCopy if true then byte arrays/buffers/sources will not
+ * be copied to this hdos but instead referenced.
+ */
+ public HeapDataOutputStream(ByteBuffer initialBuffer, Version version, boolean doNotCopy) {
+ if (initialBuffer.position() != 0) {
+ initialBuffer = initialBuffer.slice();
+ }
+ int allocSize = initialBuffer.capacity();
+ if (allocSize < 32) {
+ this.MIN_CHUNK_SIZE = 32;
+ } else {
+ this.MIN_CHUNK_SIZE = allocSize;
+ }
+ this.buffer = initialBuffer;
+ this.version = version;
+ this.doNotCopy = doNotCopy;
}
/**
@@ -112,9 +146,22 @@ public class HeapDataOutputStream extends OutputStream implements
this.MIN_CHUNK_SIZE = 32;
}
this.buffer = ByteBuffer.wrap(bytes);
+ this.doNotCopy = false;
}
/**
+ * Returns true if this HDOS currently does not copy byte arrays/buffers written to it.
+ * Instead of copying a reference is kept to the original array/buffer.
+ */
+ public boolean setDoNotCopy(boolean v) {
+ boolean result = this.doNotCopy;
+ if (result != v) {
+ this.doNotCopy = v;
+ }
+ return result;
+ }
+
+ /**
* {@inheritDoc}
*/
@Override
@@ -177,17 +224,85 @@ public class HeapDataOutputStream extends OutputStream implements
/** override OutputStream's write() */
@Override
- public void write(byte[] source, int offset, int len) {
+ public void write(byte[] source, int offset, int len) {
+ if (len == 0) return;
if (this.ignoreWrites) return;
checkIfWritable();
- int remainingSpace = this.buffer.capacity() - this.buffer.position();
- if (remainingSpace < len) {
- this.buffer.put(source, offset, remainingSpace);
- offset += remainingSpace;
- len -= remainingSpace;
- ensureCapacity(len);
+ if (this.doNotCopy && len > MIN_TO_COPY) {
+ moveBufferToChunks();
+ addToChunks(source, offset, len);
+ } else {
+ int remainingSpace = this.buffer.capacity() - this.buffer.position();
+ if (remainingSpace < len) {
+ this.buffer.put(source, offset, remainingSpace);
+ offset += remainingSpace;
+ len -= remainingSpace;
+ ensureCapacity(len);
+ }
+ this.buffer.put(source, offset, len);
+ }
+ }
+
+ private void addToChunks(byte[] source, int offset, int len) {
+ ByteBuffer bb = ByteBuffer.wrap(source, offset, len).slice();
+ bb = bb.slice();
+ this.size += bb.remaining();
+ this.chunks.add(bb);
+ }
+
+ private void addToChunks(ByteBuffer bb) {
+ int remaining = bb.remaining();
+ if (remaining > 0) {
+ this.size += remaining;
+ if (bb.position() != 0) {
+ bb = bb.slice();
+ }
+ this.chunks.add(bb);
+ }
+ }
+ public int getByteBufferCount() {
+ int result = 0;
+ if (this.chunks != null) {
+ result += this.chunks.size();
+ }
+ if (this.buffer.remaining() > 0) {
+ result++;
+ }
+ return result;
+ }
+
+ public void fillByteBufferArray(ByteBuffer[] bbArray, int offset) {
+ if (this.chunks != null) {
+ for (ByteBuffer bb: this.chunks) {
+ bbArray[offset++] = bb;
+ }
+ }
+ if (this.buffer.remaining() > 0) {
+ bbArray[offset] = this.buffer;
+ }
+ }
+
+ private void moveBufferToChunks() {
+ final ByteBuffer oldBuffer = this.buffer;
+ if (this.chunks == null) {
+ this.chunks = new LinkedList<ByteBuffer>();
+ }
+ if (oldBuffer.position() == 0) {
+ // if position is zero then nothing has been written (yet) to buffer so no need to move it to chunks
+ return;
+ }
+ oldBuffer.flip();
+ this.size += oldBuffer.remaining();
+ ByteBuffer bufToAdd = oldBuffer.slice();
+ this.chunks.add(bufToAdd);
+ int newPos = oldBuffer.limit();
+ if ((oldBuffer.capacity() - newPos) <= 0) {
+ this.buffer = ByteBuffer.allocate(MIN_CHUNK_SIZE);
+ } else {
+ oldBuffer.limit(oldBuffer.capacity());
+ oldBuffer.position(newPos);
+ this.buffer = oldBuffer.slice();
}
- this.buffer.put(source, offset, len);
}
public final int size() {
@@ -210,27 +325,7 @@ public class HeapDataOutputStream extends OutputStream implements
this.buffer = bb;
}
}
-
- public void writeWithByteArrayWrappedConditionally(byte[] source, int offset, int len) {
- if (this.ignoreWrites) return;
- checkIfWritable();
- this.expand(MIN_CHUNK_SIZE);
-
- //Asif:
- //let us expand first so that current byte buffer goes into the list
- //and a new current byte buffer is created. We than place the wrapped ByteBuffer
- // into the list
- ByteBuffer temp = ByteBuffer.wrap(source,offset,len);
- //Slicing is needed so that other functions like consolidateChunk etc work correctly
- temp = temp.slice();
- temp.limit(len);
- //Hide this buffer in the linked list so that it is not used for any further
- // writes as we want it to be immutable & it is possible that capacity is > limit
- // i.e len does not cover the end of the source.
- this.chunks.add(temp);
- this.size += len;
- }
-
+
private final void consolidateChunks() {
if (this.chunks != null) {
final int size = size();
@@ -288,13 +383,32 @@ public class HeapDataOutputStream extends OutputStream implements
this.size += this.buffer.remaining();
}
}
+ /**
+ * Returns a ByteBuffer of the unused buffer; returns null if the buffer was completely used.
+ */
+ public ByteBuffer finishWritingAndReturnUnusedBuffer() {
+ finishWriting();
+ ByteBuffer result = this.buffer.duplicate();
+ if (result.remaining() == 0) {
+ // buffer was never used.
+ result.limit(result.capacity());
+ return result;
+ }
+ int newPos = result.limit();
+ if ((result.capacity() - newPos) > 0) {
+ result.limit(result.capacity());
+ result.position(newPos);
+ return result.slice();
+ } else {
+ return null;
+ }
+ }
+
@Override
public void close() {
reset();
}
- // @todo darrel: add a method that returns a list of ByteBuffer
-
/** gets the contents of this stream as s ByteBuffer, ready for reading.
* The stream should not be written to past this point until it has been reset.
*/
@@ -380,12 +494,17 @@ public class HeapDataOutputStream extends OutputStream implements
if (size() == 0) {
return;
}
+ // TODO OFFHEAP: this code end up calling chan.write at least once for each DirectByteBuffer in this HDOS.
+ // It will combine consecutive heap ByteBuffers into a write call.
+ // It would be better to do one chan.write call with an array of ByteBuffers like sendTo(SocketChannel) does.
+ out.clear();
if (this.chunks != null) {
for (ByteBuffer bb: this.chunks) {
sendChunkTo(bb, chan, out);
}
}
sendChunkTo(this.buffer, chan, out);
+ flushBuffer(chan, out);
}
/**
@@ -394,29 +513,46 @@ public class HeapDataOutputStream extends OutputStream implements
*/
private final void sendChunkTo(ByteBuffer in, SocketChannel sc, ByteBuffer out) throws IOException {
int bytesSent = in.remaining();
- final int OUT_MAX = out.capacity();
- out.clear();
- final byte[] bytes = in.array();
- int off = in.arrayOffset() + in.position();
- int len = bytesSent;
- while (len > 0) {
- int bytesThisTime = len;
- if (bytesThisTime > OUT_MAX) {
- bytesThisTime = OUT_MAX;
+ if (in.isDirect()) {
+ flushBuffer(sc, out);
+ while (in.remaining() > 0) {
+ sc.write(in);
}
- out.put(bytes, off, bytesThisTime);
- off += bytesThisTime;
- len -= bytesThisTime;
- out.flip();
- while (out.remaining() > 0) {
- sc.write(out);
+ } else {
+ // copy in to out. If out fills flush it
+ int OUT_MAX = out.remaining();
+ if (bytesSent <= OUT_MAX) {
+ out.put(in);
+ } else {
+ final byte[] bytes = in.array();
+ int off = in.arrayOffset() + in.position();
+ int len = bytesSent;
+ while (len > 0) {
+ int bytesThisTime = len;
+ if (bytesThisTime > OUT_MAX) {
+ bytesThisTime = OUT_MAX;
+ }
+ out.put(bytes, off, bytesThisTime);
+ off += bytesThisTime;
+ len -= bytesThisTime;
+ flushBuffer(sc, out);
+ OUT_MAX = out.remaining();
+ }
+ in.position(in.limit());
}
- out.clear();
}
- in.position(in.limit());
this.size -= bytesSent;
}
+ private void flushBuffer(SocketChannel sc, ByteBuffer out) throws IOException {
+ if (out.position() == 0) return;
+ out.flip();
+ while (out.remaining() > 0) {
+ sc.write(out);
+ }
+ out.clear();
+ }
+
/**
* Write the contents of this stream to the byte buffer.
* @throws BufferOverflowException if out is not large enough to contain all of
@@ -447,41 +583,81 @@ public class HeapDataOutputStream extends OutputStream implements
}
/**
- * Write the contents of this stream to the specified stream.
+ * Write the contents of this stream to the specified stream using
+ * outBuf if a buffer is needed.
*/
- public final void sendTo(OutputStream out) throws IOException {
+ public final void sendTo(OutputStream out, ByteBuffer outBuf) throws IOException {
finishWriting();
if (this.chunks != null) {
for (ByteBuffer bb: this.chunks) {
- int bytesToWrite = bb.remaining();
- if (bytesToWrite > 0) {
- if (bb.hasArray()) {
- out.write(bb.array(), bb.arrayOffset()+bb.position(), bytesToWrite);
- bb.position(bb.limit());
- } else { // fix for bug 43007
- byte[] bytes = new byte[bytesToWrite];
- bb.get(bytes);
- out.write(bytes);
- }
- this.size -= bytesToWrite;
- }
+ sendTo(out, outBuf, bb);
}
}
- {
- ByteBuffer bb = this.buffer;
- int bytesToWrite = bb.remaining();
- if (bytesToWrite > 0) {
- if (bb.hasArray()) {
- out.write(bb.array(), bb.arrayOffset()+bb.position(), bytesToWrite);
- bb.position(bb.limit());
- } else {
- byte[] bytes = new byte[bytesToWrite];
- bb.get(bytes);
- out.write(bytes);
+ sendTo(out, outBuf, this.buffer);
+ flushStream(out, outBuf);
+ }
+
+ private void sendTo(OutputStream out, ByteBuffer outBuf, ByteBuffer inBuf) throws IOException {
+ this.size -= writeByteBufferToStream(out, outBuf, inBuf);
+ }
+
+ /**
+ * Returns the number of bytes written
+ */
+ public static int writeByteBufferToStream(OutputStream out, ByteBuffer outBuf, ByteBuffer inBuf) throws IOException {
+ int bytesToWrite = inBuf.remaining();
+ if (bytesToWrite > 0) {
+ if (inBuf.hasArray()) {
+ flushStream(out, outBuf);
+ out.write(inBuf.array(), inBuf.arrayOffset()+inBuf.position(), bytesToWrite);
+ inBuf.position(inBuf.limit());
+ } else { // fix for bug 43007
+ // copy direct inBuf to heap outBuf. If out fills flush it
+ int bytesToWriteThisTime = bytesToWrite;
+ int OUT_MAX = outBuf.remaining();
+ while (bytesToWriteThisTime > OUT_MAX) {
+ // copy only OUT_MAX bytes and flush outBuf
+ int oldLimit = inBuf.limit();
+ inBuf.limit(inBuf.position()+OUT_MAX);
+ outBuf.put(inBuf);
+ inBuf.limit(oldLimit);
+ flushStream(out, outBuf);
+ bytesToWriteThisTime -= OUT_MAX;
+ OUT_MAX = outBuf.remaining();
}
- this.size -= bytesToWrite;
+ outBuf.put(inBuf);
+ }
+ }
+ return bytesToWrite;
+ }
+
+ public static void flushStream(OutputStream out, ByteBuffer outBuf) throws IOException {
+ if (outBuf.position() == 0) return;
+ assert outBuf.hasArray();
+ outBuf.flip();
+ out.write(outBuf.array(), outBuf.arrayOffset(), outBuf.remaining());
+ outBuf.clear();
+ }
+
+ /**
+ * Write the contents of this stream to the specified stream.
+ */
+ public final void sendTo(ByteBufferWriter out) {
+ finishWriting();
+ if (this.chunks != null) {
+ for (ByteBuffer bb: this.chunks) {
+ basicSendTo(out, bb);
}
}
+ basicSendTo(out, this.buffer);
+ }
+
+ private void basicSendTo(ByteBufferWriter out, ByteBuffer bb) {
+ int bytesToWrite = bb.remaining();
+ if (bytesToWrite > 0) {
+ out.write(bb.duplicate());
+ this.size -= bytesToWrite;
+ }
}
/**
* Returns an input stream that can be used to read the contents that
@@ -1097,9 +1273,19 @@ public class HeapDataOutputStream extends OutputStream implements
ensureCapacity(5);
if (v instanceof HeapDataOutputStream) {
HeapDataOutputStream other = (HeapDataOutputStream)v;
+ other.finishWriting();
InternalDataSerializer.writeArrayLength(other.size(), this);
- other.sendTo((OutputStream)this);
- other.rewind();
+ if (this.doNotCopy) {
+ if (other.chunks != null) {
+ for (ByteBuffer bb: other.chunks) {
+ write(bb);
+ }
+ }
+ write(other.buffer);
+ } else {
+ other.sendTo((ByteBufferWriter)this);
+ other.rewind();
+ }
} else {
ByteBuffer sizeBuf = this.buffer;
int sizePos = sizeBuf.position();
@@ -1111,24 +1297,69 @@ public class HeapDataOutputStream extends OutputStream implements
sizeBuf.putInt(sizePos+1, arraySize);
}
}
-
+
+ /**
+ * We set "doNotCopy" to prevent wasting time
+ * by copying bytes. But to do this we create
+ * either a HeapByteBuffer to DirectByteBuffer
+ * to reference the byte array or off-heap memory.
+ * The ByteBuffer instance itself uses up memory
+ * that needs to be initialized and eventually
+ * gc'd so for smaller sizes it is better to just copy it.
+ * Public for unit test access.
+ */
+ public static final int MIN_TO_COPY = 128;
+
/**
* Write a byte buffer to this HeapDataOutputStream,
*
* the contents of the buffer between the position and the limit
* are copied to the output stream.
*/
- public void write(ByteBuffer source) {
+ @Override
+ public void write(ByteBuffer bb) {
if (this.ignoreWrites) return;
checkIfWritable();
- int remainingSpace = this.buffer.capacity() - this.buffer.position();
+ int remaining = bb.remaining();
+ if (remaining == 0) return;
+ if (this.doNotCopy && remaining > MIN_TO_COPY) {
+ moveBufferToChunks();
+ addToChunks(bb);
+ } else {
+ int remainingSpace = this.buffer.remaining();
+ if (remainingSpace < remaining) {
+ int oldLimit = bb.limit();
+ bb.limit(bb.position() + remainingSpace);
+ this.buffer.put(bb);
+ bb.limit(oldLimit);
+ ensureCapacity(bb.remaining());
+ }
+ this.buffer.put(bb);
+ }
+ }
+
+ /**
+ * Write a byte source to this HeapDataOutputStream,
+ *
+ * the contents of the buffer between the position and the limit
+ * are copied to the output stream.
+ */
+ public void write(ByteSource source) {
+ ByteBuffer bb = source.getBackingByteBuffer();
+ if (bb != null) {
+ write(bb);
+ return;
+ }
+ if (this.ignoreWrites) return;
+ checkIfWritable();
+ int remainingSpace = this.buffer.limit() - this.buffer.position();
if (remainingSpace < source.remaining()) {
int oldLimit = source.limit();
source.limit(source.position() + remainingSpace);
- this.buffer.put(source);
+ source.sendTo(this.buffer);
source.limit(oldLimit);
ensureCapacity(source.remaining());
}
- this.buffer.put(source);
+ source.sendTo(this.buffer);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/InsufficientDiskSpaceException.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/InsufficientDiskSpaceException.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/InsufficientDiskSpaceException.java
index 5d8b290..1f4b1ef 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/InsufficientDiskSpaceException.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/InsufficientDiskSpaceException.java
@@ -13,7 +13,7 @@ import java.io.IOException;
import com.gemstone.gemfire.cache.DiskAccessException;
import com.gemstone.gemfire.cache.DiskStore;
-import com.gemstone.gemfire.internal.InternalDataSerializer.Sendable;
+import com.gemstone.gemfire.internal.Sendable;
/**
* @author kneeraj
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/InternalDataSerializer.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/InternalDataSerializer.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/InternalDataSerializer.java
index 3eb7c5b..87ab28d 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/InternalDataSerializer.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/InternalDataSerializer.java
@@ -413,16 +413,38 @@ public abstract class InternalDataSerializer extends DataSerializer implements D
writeStringArray(array, out);
return true;
}});
- classesToSerializers.put("java.util.concurrent.TimeUnit",
- new WellKnownDS() {
- @Override
- public final boolean toData(Object o, DataOutput out)
- throws IOException {
- TimeUnit unit = (TimeUnit) o;
- out.writeByte(TIME_UNIT);
- writeTimeUnit(unit, out);
- return true;
- }});
+ classesToSerializers.put(TimeUnit.NANOSECONDS.getClass().getName(),
+ new WellKnownDS() {
+ @Override public final boolean toData(Object o, DataOutput out)
+ throws IOException {
+ out.writeByte(TIME_UNIT);
+ out.writeByte(TIME_UNIT_NANOSECONDS);
+ return true;
+ }});
+ classesToSerializers.put(TimeUnit.MICROSECONDS.getClass().getName(),
+ new WellKnownDS() {
+ @Override public final boolean toData(Object o, DataOutput out)
+ throws IOException {
+ out.writeByte(TIME_UNIT);
+ out.writeByte(TIME_UNIT_MICROSECONDS);
+ return true;
+ }});
+ classesToSerializers.put(TimeUnit.MILLISECONDS.getClass().getName(),
+ new WellKnownDS() {
+ @Override public final boolean toData(Object o, DataOutput out)
+ throws IOException {
+ out.writeByte(TIME_UNIT);
+ out.writeByte(TIME_UNIT_MILLISECONDS);
+ return true;
+ }});
+ classesToSerializers.put(TimeUnit.SECONDS.getClass().getName(),
+ new WellKnownDS() {
+ @Override public final boolean toData(Object o, DataOutput out)
+ throws IOException {
+ out.writeByte(TIME_UNIT);
+ out.writeByte(TIME_UNIT_SECONDS);
+ return true;
+ }});
classesToSerializers.put("java.util.Date",
new WellKnownPdxDS() {
@Override
@@ -1562,7 +1584,8 @@ public abstract class InternalDataSerializer extends DataSerializer implements D
out.writeByte(OBJECT_ARRAY);
writeObjectArray(array, out, ensurePdxCompatibility);
return true;
- } else if (is662SerializationEnabled() && o.getClass().isEnum()) {
+ } else if (is662SerializationEnabled() && (o.getClass().isEnum()
+ /* for bug 52271 */ || (o.getClass().getSuperclass() != null && o.getClass().getSuperclass().isEnum()))) {
if (isPdxSerializationInProgress()) {
writePdxEnum((Enum<?>)o, out);
} else {
@@ -1988,40 +2011,6 @@ public abstract class InternalDataSerializer extends DataSerializer implements D
private static final byte TIME_UNIT_SECONDS = -4;
/**
- * Writes a <code>TimeUnit</code> to a <code>DataOutput</code>.
- *
- * @throws IOException
- * A problem occurs while writing to <code>out</code>
- *
- * @see #readTimeUnit
- */
- public static void writeTimeUnit(TimeUnit unit, DataOutput out)
- throws IOException {
-
- InternalDataSerializer.checkOut(out);
-
- if (logger.isTraceEnabled(LogMarker.SERIALIZER)) {
- logger.trace(LogMarker.SERIALIZER, "Writing TimeUnit: {}", unit);
- }
-
- if (unit.equals(TimeUnit.NANOSECONDS)) {
- out.writeByte(TIME_UNIT_NANOSECONDS);
-
- } else if (unit.equals(TimeUnit.MICROSECONDS)) {
- out.writeByte(TIME_UNIT_MICROSECONDS);
-
- } else if (unit.equals(TimeUnit.MILLISECONDS)) {
- out.writeByte(TIME_UNIT_MILLISECONDS);
-
- } else if (unit.equals(TimeUnit.SECONDS)) {
- out.writeByte(TIME_UNIT_SECONDS);
-
- } else {
- throw new InternalGemFireException(LocalizedStrings.DataSerializer_UNSUPPORTED_TIMEUNIT_0.toLocalizedString(unit));
- }
- }
-
- /**
* Reads a <code>TimeUnit</code> from a <code>DataInput</code>.
*
* @throws IOException
@@ -2419,6 +2408,15 @@ public abstract class InternalDataSerializer extends DataSerializer implements D
// }
};
}
+ boolean wasDoNotCopy = false;
+ if (out instanceof HeapDataOutputStream) {
+ // To fix bug 52197 disable doNotCopy mode
+ // while serialize with an ObjectOutputStream.
+ // The problem is that ObjectOutputStream keeps
+ // an internal byte array that it reuses while serializing.
+ wasDoNotCopy = ((HeapDataOutputStream) out).setDoNotCopy(false);
+ }
+ try {
ObjectOutput oos = new ObjectOutputStream(stream);
if ( stream instanceof VersionedDataStream ) {
Version v = ((VersionedDataStream)stream).getVersion();
@@ -2430,6 +2428,11 @@ public abstract class InternalDataSerializer extends DataSerializer implements D
// To fix bug 35568 just call flush. We can't call close because
// it calls close on the wrapped OutputStream.
oos.flush();
+ } finally {
+ if (wasDoNotCopy) {
+ ((HeapDataOutputStream) out).setDoNotCopy(true);
+ }
+ }
}
}
@@ -2583,7 +2586,6 @@ public abstract class InternalDataSerializer extends DataSerializer implements D
} catch (CacheClosedException cce) {
throw cce;
} catch (Exception ex) {
- logger.fatal("exception in deserialization", ex);
SerializationException ex2 = new SerializationException(LocalizedStrings.DataSerializer_COULD_NOT_CREATE_AN_INSTANCE_OF_0.toLocalizedString(ds.getClass().getName()), ex);
throw ex2;
}
@@ -3860,27 +3862,6 @@ public abstract class InternalDataSerializer extends DataSerializer implements D
// subclasses need to implement toData
}
- /**
- * Interface to implement if your class supports sending itself directly to a DataOutput
- * during serialization.
- * Note that you are responsable for sending all the bytes that represent your instance,
- * even bytes describing your class name if those are required.
- *
- * @author darrel
- * @since 6.6
- */
- public static interface Sendable {
- /**
- * Take all the bytes in the object and write them to the data output. It needs
- * to be written in the GemFire wire format so that it will deserialize correctly.
- *
- * @param out
- * the data output to send this object to
- * @throws IOException
- */
- void sendTo(DataOutput out) throws IOException;
- }
-
public static void writeObjectArray(Object[] array, DataOutput out, boolean ensureCompatibility)
throws IOException {
InternalDataSerializer.checkOut(out);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/LinuxProcFsStatistics.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/LinuxProcFsStatistics.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/LinuxProcFsStatistics.java
index a54bc40..ef950a2 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/LinuxProcFsStatistics.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/LinuxProcFsStatistics.java
@@ -7,21 +7,20 @@
*/
package com.gemstone.gemfire.internal;
+import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
-import java.io.InputStreamReader;
-import java.io.BufferedReader;
import java.io.IOException;
+import java.io.InputStreamReader;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
-import java.util.ListIterator;
import java.util.NoSuchElementException;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
-import com.gemstone.gemfire.internal.LinuxProcessStats;
-import com.gemstone.gemfire.internal.LinuxSystemStats;
+import org.apache.logging.log4j.Logger;
-class LinuxProcFsStatistics {
+public class LinuxProcFsStatistics {
private enum CPU {
USER, NICE, SYSTEM, IDLE, IOWAIT, IRQ, SOFTIRQ,
/** stands for aggregation of all columns not present in the enum list*/
@@ -199,6 +198,38 @@ class LinuxProcFsStatistics {
}
}
+ /**
+ * Returns the available system memory (free + cached).
+ * @param logger the logger
+ * @return the available memory in bytes
+ */
+ public static long getAvailableMemory(Logger logger) {
+ try {
+ BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream("/proc/meminfo")));
+ try {
+ long free = 0;
+ Pattern p = Pattern.compile("(.*)?:\\s+(\\d+)( kB)?");
+
+ String line;
+ while ((line = br.readLine()) != null) {
+ Matcher m = p.matcher(line);
+ if (m.matches() && ("MemFree".equals(m.group(1)) || "Cached".equals(m.group(1)))) {
+ free += Long.parseLong(m.group(2));
+ }
+ }
+
+ // convert to bytes
+ return 1024 * free;
+
+ } finally {
+ br.close();
+ }
+ } catch (IOException e) {
+ logger.warn("Error determining free memory", e);
+ return Long.MAX_VALUE;
+ }
+ }
+
// Example of /proc/meminfo
// total: used: free: shared: buffers: cached:
//Mem: 4118380544 3816050688 302329856 0 109404160 3060326400
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/OSProcess.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/OSProcess.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/OSProcess.java
index 2893fbb..6bb07a1 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/OSProcess.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/OSProcess.java
@@ -50,8 +50,11 @@ import com.gemstone.gemfire.internal.logging.LoggingThreadGroup;
public class OSProcess {
private static final Logger logger = LogService.getLogger();
- private static final boolean DISABLE_OUTPUT_REDIRECTION = Boolean.getBoolean("gemfire.OSProcess.DISABLE_OUTPUT_REDIRECTION");
- private static final boolean ENABLE_OUTPUT_REDIRECTION = Boolean.getBoolean("gemfire.OSProcess.ENABLE_OUTPUT_REDIRECTION");
+ public static final String DISABLE_OUTPUT_REDIRECTION_PROPERTY = "gemfire.OSProcess.DISABLE_OUTPUT_REDIRECTION";
+ public static final String ENABLE_OUTPUT_REDIRECTION_PROPERTY = "gemfire.OSProcess.ENABLE_OUTPUT_REDIRECTION";
+
+ private static final boolean DISABLE_OUTPUT_REDIRECTION = Boolean.getBoolean(DISABLE_OUTPUT_REDIRECTION_PROPERTY);
+ private static final boolean ENABLE_OUTPUT_REDIRECTION = Boolean.getBoolean(ENABLE_OUTPUT_REDIRECTION_PROPERTY);
static final boolean pureMode = PureJavaMode.isPure();
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Sendable.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Sendable.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Sendable.java
new file mode 100644
index 0000000..2d797be
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Sendable.java
@@ -0,0 +1,26 @@
+package com.gemstone.gemfire.internal;
+
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ * Interface to implement if your class supports sending itself directly to a DataOutput
+ * during serialization.
+ * Note that you are responsible for sending all the bytes that represent your instance,
+ * even bytes describing your class name if those are required.
+ *
+ * @author darrel
+ * @since 6.6
+ */
+public interface Sendable {
+ /**
+ * Take all the bytes in the object and write them to the data output. It needs
+ * to be written in the GemFire wire format so that it will deserialize correctly
+ * if DataSerializer.readObject is called.
+ *
+ * @param out
+ * the data output to send this object to
+ * @throws IOException
+ */
+ void sendTo(DataOutput out) throws IOException;
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SharedLibrary.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SharedLibrary.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SharedLibrary.java
index c5fe993..59ab34e 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SharedLibrary.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SharedLibrary.java
@@ -55,7 +55,7 @@ public class SharedLibrary {
is64Bit = false;
} else {
if (unsafe != null) {
- is64Bit = (new UnsafeWrapper()).getAddressSize() == 8;
+ is64Bit = unsafe.getAddressSize() == 8;
} else {
is64Bit = false;
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SystemAdmin.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SystemAdmin.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SystemAdmin.java
index a308f83..6c4b01c 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SystemAdmin.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SystemAdmin.java
@@ -736,7 +736,7 @@ public class SystemAdmin {
DiskStoreImpl.modifyRegion(diskStoreName, dirs, regionOption,
lruOption, lruActionOption, lruLimitOption,
concurrencyLevelOption, initialCapacityOption, loadFactorOption,
- compressorClassNameOption, statisticsEnabledOption, true);
+ compressorClassNameOption, statisticsEnabledOption, null/*offHeap*/, true);
System.out.println("The region " + regionOption + " was successfully modified in the disk store " + diskStoreName);
} else if (remove) {
if (regionOption == null) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SystemTimer.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SystemTimer.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SystemTimer.java
index 38e73b3..42e5447 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SystemTimer.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SystemTimer.java
@@ -333,7 +333,7 @@ public final class SystemTimer {
checkCancelled();
if (logger.isTraceEnabled()) {
Date tilt = new Date(System.currentTimeMillis() + delay);
- SimpleDateFormat sdf = new SimpleDateFormat("yyyy-mm-DD HH:mm:ss.SSS");
+ SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
logger.trace("SystemTimer#schedule (long): {}: expect task {} to fire around {}", this, task, sdf.format(tilt));
}
timer.schedule(task, delay);
@@ -345,7 +345,7 @@ public final class SystemTimer {
public void schedule(SystemTimerTask task, Date time) {
checkCancelled();
if (logger.isTraceEnabled()) {
- SimpleDateFormat sdf = new SimpleDateFormat("yyyy-mm-DD HH:mm:ss.SSS");
+ SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
logger.trace("SystemTimer#schedule (Date): {}: expect task {} to fire around {}", this, task, sdf.format(time));
}
timer.schedule(task, time);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Version.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Version.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Version.java
index d953fd5..2ccdaeb 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Version.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/Version.java
@@ -55,7 +55,7 @@ public final class Version implements Comparable<Version> {
/** byte used as ordinal to represent this <code>Version</code> */
private final short ordinal;
- public static final int HIGHEST_VERSION = 38;
+ public static final int HIGHEST_VERSION = 43;
private static final Version[] VALUES = new Version[HIGHEST_VERSION+1];
@@ -171,15 +171,25 @@ public final class Version implements Comparable<Version> {
public static final Version GFE_81 = new Version("GFE", "8.1", (byte)8,
(byte)1, (byte)0, (byte)0, GFE_81_ORDINAL);
- private static final byte GFE_82_ORDINAL = 38;
+ // 36-39 available for 8.1.x variants
+
+ private static final byte GFE_82_ORDINAL = 40;
public static final Version GFE_82 = new Version("GFE", "8.2", (byte)8,
(byte)2, (byte)0, (byte)0, GFE_82_ORDINAL);
+ // 41-42 available for 8.2.x variants
+
+ private static final byte GFE_90_ORDINAL = 43;
+
+ public static final Version GFE_90 = new Version("GFE", "9.0", (byte)9,
+ (byte)0, (byte)0, (byte)0, GFE_90_ORDINAL);
+
/**
* This constant must be set to the most current version of GFE/SQLF.
+ * !!! NOTE: update HIGHEST_VERSION when changing CURRENT !!!
*/
- public static final Version CURRENT = GFE_82;
+ public static final Version CURRENT = GFE_90;
/**
* A lot of versioning code needs access to the current version's ordinal
@@ -606,4 +616,12 @@ public final class Version implements Comparable<Version> {
result = mult * result + this.ordinal;
return result;
}
+
+ public byte[] toBytes() {
+ byte[] bytes = new byte[2];
+ bytes[0] = (byte)(ordinal >> 8);
+ bytes[1] = (byte)ordinal;
+ return bytes;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/ClientStatsManager.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/ClientStatsManager.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/ClientStatsManager.java
index 5600628..7166024 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/ClientStatsManager.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/ClientStatsManager.java
@@ -78,8 +78,12 @@ public class ClientStatsManager {
EventID eventId = new EventID(ds);
EntryEventImpl event = new EntryEventImpl((Object)null);
+ try {
event.setEventId(eventId);
regionProxy.putForMetaRegion(ds.getMemberId(), stats, null, event, null, true);
+ } finally {
+ event.release();
+ }
}
catch (DistributedSystemDisconnectedException e) {
throw e;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/remote/RemoteRegionAttributes.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/remote/RemoteRegionAttributes.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/remote/RemoteRegionAttributes.java
index bdf4db2..0aa40a7 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/remote/RemoteRegionAttributes.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/remote/RemoteRegionAttributes.java
@@ -24,6 +24,7 @@ import com.gemstone.gemfire.cache.CacheLoader;
import com.gemstone.gemfire.cache.CacheLoaderException;
import com.gemstone.gemfire.cache.CacheWriter;
import com.gemstone.gemfire.cache.CacheWriterException;
+import com.gemstone.gemfire.cache.CustomEvictionAttributes;
import com.gemstone.gemfire.cache.CustomExpiry;
import com.gemstone.gemfire.cache.DataPolicy;
import com.gemstone.gemfire.cache.Declarable;
@@ -98,7 +99,10 @@ public class RemoteRegionAttributes implements RegionAttributes,
private String[] gatewaySendersDescs;
private boolean isGatewaySenderEnabled = false;
private String[] asyncEventQueueDescs;
+ private String hdfsStoreName;
+ private boolean hdfsWriteOnly;
private String compressorDesc;
+ private boolean offHeap;
/**
* constructs a new default RemoteRegionAttributes.
@@ -148,7 +152,10 @@ public class RemoteRegionAttributes implements RegionAttributes,
this.isDiskSynchronous = attr.isDiskSynchronous();
this.gatewaySendersDescs = getDescs(attr.getGatewaySenderIds().toArray());
this.asyncEventQueueDescs = getDescs(attr.getAsyncEventQueueIds().toArray());
+ this.hdfsStoreName = attr.getHDFSStoreName();
+ this.hdfsWriteOnly = attr.getHDFSWriteOnly();
this.compressorDesc = getDesc(attr.getCompressor());
+ this.offHeap = attr.getOffHeap();
}
/**
@@ -355,6 +362,10 @@ public class RemoteRegionAttributes implements RegionAttributes,
compressorDesc);
}
+ public boolean getOffHeap() {
+ return this.offHeap;
+ }
+
public void toDataPre_GFE_8_0_0_0(DataOutput out) throws IOException {
DataSerializer.writeString(this.cacheLoaderDesc, out);
DataSerializer.writeString(this.cacheWriterDesc, out);
@@ -400,7 +411,7 @@ public class RemoteRegionAttributes implements RegionAttributes,
out.writeBoolean(this.concurrencyChecksEnabled);
}
- public void toData(DataOutput out) throws IOException {
+ public void toDataPre_GFE_9_0_0_0(DataOutput out) throws IOException {
DataSerializer.writeString(this.cacheLoaderDesc, out);
DataSerializer.writeString(this.cacheWriterDesc, out);
@@ -451,7 +462,13 @@ public class RemoteRegionAttributes implements RegionAttributes,
DataSerializer.writeString(this.compressorDesc, out);
}
-
+
+ public void toData(DataOutput out) throws IOException {
+ toDataPre_GFE_9_0_0_0(out);
+ out.writeBoolean(this.offHeap);
+ DataSerializer.writeString(this.hdfsStoreName, out);
+ }
+
public void fromDataPre_GFE_8_0_0_0(DataInput in) throws IOException, ClassNotFoundException {
this.cacheLoaderDesc = DataSerializer.readString(in);
this.cacheWriterDesc = DataSerializer.readString(in);
@@ -499,7 +516,7 @@ public class RemoteRegionAttributes implements RegionAttributes,
this.concurrencyChecksEnabled = in.readBoolean();
}
- public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ public void fromDataPre_GFE_9_0_0_0(DataInput in) throws IOException, ClassNotFoundException {
this.cacheLoaderDesc = DataSerializer.readString(in);
this.cacheWriterDesc = DataSerializer.readString(in);
this.cacheListenerDescs = DataSerializer.readStringArray(in);
@@ -550,7 +567,13 @@ public class RemoteRegionAttributes implements RegionAttributes,
this.compressorDesc = DataSerializer.readString(in);
}
-
+
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ fromDataPre_GFE_9_0_0_0(in);
+ this.offHeap = in.readBoolean();
+ this.hdfsStoreName = DataSerializer.readString(in);
+ }
+
private String[] getDescs(Object[] l) {
if (l == null) {
return new String[0];
@@ -716,6 +739,15 @@ public class RemoteRegionAttributes implements RegionAttributes,
return this.evictionAttributes;
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public CustomEvictionAttributes getCustomEvictionAttributes() {
+ // TODO: HDFS: no support for custom eviction attributes from remote yet
+ return null;
+ }
+
public boolean getCloningEnabled() {
// TODO Auto-generated method stub
return this.cloningEnable;
@@ -724,7 +756,12 @@ public class RemoteRegionAttributes implements RegionAttributes,
public String getDiskStoreName() {
return this.diskStoreName;
}
-
+ public String getHDFSStoreName() {
+ return this.hdfsStoreName;
+ }
+ public boolean getHDFSWriteOnly() {
+ return this.hdfsWriteOnly;
+ }
public boolean isDiskSynchronous() {
return this.isDiskSynchronous;
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/31d1b20e/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/remote/ShutdownAllGatewayHubsRequest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/remote/ShutdownAllGatewayHubsRequest.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/remote/ShutdownAllGatewayHubsRequest.java
new file mode 100644
index 0000000..6d98d5d
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/admin/remote/ShutdownAllGatewayHubsRequest.java
@@ -0,0 +1,52 @@
+package com.gemstone.gemfire.internal.admin.remote;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import com.gemstone.gemfire.distributed.internal.DistributionManager;
+import com.gemstone.gemfire.distributed.internal.DistributionMessage;
+import com.gemstone.gemfire.distributed.internal.ReplyMessage;
+
+/**
+ *
+ * This ShutdownAllGatewayHubsRequest just reply with ignored bit true so that
+ * old version member's request will be ignored and no exception will be thrown.
+ *
+ * From 9.0 old wan support is removed. Ideally ShutdownAllGatewayHubsRequest
+ * should be removed but it it there for rolling upgrade support when request
+ * come from old version member to shut down hubs.
+ *
+ * @author kbachhav
+ * @since 9.0
+ *
+ */
+public class ShutdownAllGatewayHubsRequest extends DistributionMessage {
+
+ protected int rpid;
+
+ @Override
+ public int getDSFID() {
+ return SHUTDOWN_ALL_GATEWAYHUBS_REQUEST;
+ }
+
+ @Override
+ public int getProcessorType() {
+ return DistributionManager.STANDARD_EXECUTOR;
+ }
+
+ @Override
+ protected void process(DistributionManager dm) {
+ ReplyMessage.send(getSender(), this.rpid, null, dm, true /*ignored*/, false, false);
+ }
+
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ super.fromData(in);
+ this.rpid = in.readInt();
+ }
+
+ public void toData(DataOutput out) throws IOException {
+ super.toData(out);
+ out.writeInt(this.rpid);
+ }
+}