You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by rv...@apache.org on 2015/07/03 21:21:33 UTC
[32/51] [partial] incubator-geode git commit: SGA #2
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Cache.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Cache.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Cache.java
index 4389fda..72a5555 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Cache.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Cache.java
@@ -16,6 +16,7 @@ import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue;
import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueueFactory;
import com.gemstone.gemfire.cache.client.ClientCache;
import com.gemstone.gemfire.cache.client.Pool;
+import com.gemstone.gemfire.cache.hdfs.HDFSStoreFactory;
import com.gemstone.gemfire.cache.server.CacheServer;
import com.gemstone.gemfire.cache.snapshot.CacheSnapshotService;
import com.gemstone.gemfire.cache.util.BridgeServer;
@@ -27,6 +28,7 @@ import com.gemstone.gemfire.cache.wan.GatewaySenderFactory;
import com.gemstone.gemfire.distributed.DistributedMember;
import com.gemstone.gemfire.distributed.DistributedSystem;
import com.gemstone.gemfire.i18n.LogWriterI18n;
+import com.gemstone.gemfire.cache.hdfs.HDFSStore;
/**
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CacheFactory.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CacheFactory.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CacheFactory.java
index 11536df..609bf97 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CacheFactory.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CacheFactory.java
@@ -166,20 +166,8 @@ public class CacheFactory {
GatewayException,
RegionExistsException
{
- GemFireCacheImpl instance = GemFireCacheImpl.getInstance();
-
- if (instance != null && !instance.isClosed()) {
- if (existingOk) {
- // Check if cache configuration matches.
- cacheConfig.validateCacheConfig(instance);
-
- return instance;
- } else {
- // instance.creationStack argument is for debugging...
- throw new CacheExistsException(instance, LocalizedStrings.CacheFactory_0_AN_OPEN_CACHE_ALREADY_EXISTS.toLocalizedString(instance), instance.creationStack);
- }
- }
- return GemFireCacheImpl.create(system, cacheConfig);
+ // Moved code in this method to GemFireCacheImpl.create
+ return GemFireCacheImpl.create(system, existingOk, cacheConfig);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CacheTransactionManager.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CacheTransactionManager.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CacheTransactionManager.java
index 0c666fa..41d0162 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CacheTransactionManager.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CacheTransactionManager.java
@@ -319,4 +319,23 @@ public interface CacheTransactionManager {
* @since 6.5
*/
public TransactionWriter getWriter();
+
+ /**
+ * Sets whether transactions should be executed in distributed or
+ * non-distributed mode. Once set this mode should not be changed during
+ * the course of transactions.
+ *
+ * @throws IllegalStateException if a transaction is already in progress
+ * and this method sets the distributed mode to a different value.
+ * @since 9.0
+ */
+ public void setDistributed(boolean distributed);
+
+ /**
+ * Returns the execution mode of transactions
+ * @return true if distributed,
+ * false otherwise.
+ * @since 9.0
+ */
+ public boolean isDistributed();
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CustomEvictionAttributes.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CustomEvictionAttributes.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CustomEvictionAttributes.java
new file mode 100644
index 0000000..fc2727e
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/CustomEvictionAttributes.java
@@ -0,0 +1,71 @@
+/*=========================================================================
+ * Copyright (c) 2009-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. GoPivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+
+package com.gemstone.gemfire.cache;
+
+/**
+ * Custom eviction attributes including {@link EvictionCriteria} and evictor
+ * start time and frequency, if any.
+ *
+ * @author swale
+ * @since gfxd 1.0
+ */
+public abstract class CustomEvictionAttributes {
+
+ private final EvictionCriteria<?, ?> criteria;
+
+ private final long evictorStartTime;
+ private final long evictorInterval;
+
+ private final boolean evictIncoming;
+
+ protected CustomEvictionAttributes(EvictionCriteria<?, ?> criteria,
+ long startTime, long interval, boolean evictIncoming) {
+ this.criteria = criteria;
+ this.evictorStartTime = startTime;
+ this.evictorInterval = interval;
+ this.evictIncoming = evictIncoming;
+ }
+
+ /**
+ * Get the {@link EvictionCriteria} for this custom eviction. The criteria
+ * will be applied to the region entries either periodically as per
+ * {@link #getEvictorStartTime()} and {@link #getEvictorInterval()}, or on
+ * incoming puts if {@link #isEvictIncoming()} is true.
+ */
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ public <K, V> EvictionCriteria<K, V> getCriteria() {
+ return (EvictionCriteria)this.criteria;
+ }
+
+ /**
+ * The absolute start time in milliseconds (as returned by
+ * {@link System#currentTimeMillis()}) when the evictor will be first fired.
+ * Thereafter the evictor will be fired periodically every
+ * {@link #getEvictorInterval()} milliseconds.
+ */
+ public final long getEvictorStartTime() {
+ return this.evictorStartTime;
+ }
+
+ /**
+ * The intervals at which the periodic evictor task is fired and
+ * {@link EvictionCriteria} evaluated to evict entries.
+ */
+ public final long getEvictorInterval() {
+ return this.evictorInterval;
+ }
+
+ /**
+ * If this returns true, then the criteria should always be applied to
+ * incoming entries and never as a periodic task.
+ */
+ public final boolean isEvictIncoming() {
+ return this.evictIncoming;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/DataPolicy.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/DataPolicy.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/DataPolicy.java
index fdf9a12..4ffeaba 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/DataPolicy.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/DataPolicy.java
@@ -22,6 +22,7 @@ import java.io.*;
* <li><code>PERSISTENT_PARTITION</code> in addition to <code>PARTITION</code> also causes data to be stored to disk. The region initialization uses the data stored on disk.
* <li><code>REPLICATE</code> causes data that this region is interested in to be stored in local memory. A distributed region will be initialized with the data from other caches. On distributed region operations that would cause the contents to differ with other caches are not allowed. This policy is allowed on local scope region but it behaves the same as <code>NORMAL</code>.
* <li><code>PERSISTENT_REPLICATE</code> in addition to <code>REPLICATE</code> also causes data to be stored to disk. The region initialization uses the data stored on disk. Note that the persistence applies to both local scope and distributed scope.
+ * <li><code>HDFS_PARTITION</code> in addition to <code>PARTITION</code> also causes data to be stored to HDFS. The region initialization may use the data stored on HDFS.
* </ol>
*
* @author Darrel Schneider
@@ -35,7 +36,7 @@ import java.io.*;
public class DataPolicy implements java.io.Serializable {
private static final long serialVersionUID = 2095573273889467233L;
- private static final DataPolicy[] VALUES = new DataPolicy[10];
+ private static final DataPolicy[] VALUES = new DataPolicy[11];
/**
* Data is never stored in local memory.
@@ -87,6 +88,18 @@ public class DataPolicy implements java.io.Serializable {
*/
public static final DataPolicy PERSISTENT_PARTITION = new DataPolicy(6, "PERSISTENT_PARTITION");
+ /**
+ * In addition to <code>PARTITION</code> also causes data to be stored to
+ * HDFS. The region initialization may use the data stored on HDFS.
+ */
+ public static final DataPolicy HDFS_PARTITION = new DataPolicy(7, "HDFS_PARTITION");
+
+ /**
+ * In addition to <code>HDFS_PARTITION</code> also causes data to be stored on local
+ * disk. The data can be evicted from the local disk and still be read
+ * from HDFS.
+ */
+ public static final DataPolicy HDFS_PERSISTENT_PARTITION = new DataPolicy(10, "HDFS_PERSISTENT_PARTITION");
/**
* The data policy used by default; it is {@link #NORMAL}.
*/
@@ -156,7 +169,7 @@ public class DataPolicy implements java.io.Serializable {
* @since 6.5
*/
public boolean withPersistence() {
- return this == PERSISTENT_PARTITION || this == PERSISTENT_REPLICATE;
+ return this == PERSISTENT_PARTITION || this == PERSISTENT_REPLICATE || this == HDFS_PERSISTENT_PARTITION;
}
/** Return whether this policy does partitioning.
@@ -166,7 +179,7 @@ public class DataPolicy implements java.io.Serializable {
* @since 6.5
*/
public boolean withPartitioning() {
- return this == PARTITION || this == PERSISTENT_PARTITION;
+ return this == PARTITION || this == PERSISTENT_PARTITION || this == HDFS_PARTITION || this==HDFS_PERSISTENT_PARTITION;
}
/** Return whether this policy does preloaded.
@@ -236,6 +249,16 @@ public class DataPolicy implements java.io.Serializable {
return this == PARTITION;
}
+ /** Return whether this policy does persistence on HDFS.
+ * @return true if this policy does persistence on HDFS.
+ * @see #HDFS_PARTITION
+ */
+ public boolean withHDFS() {
+ return this == HDFS_PARTITION || this == HDFS_PERSISTENT_PARTITION;
+ }
+
+
+
/** Returns a string representation for this data policy.
* @return the name of this data policy.
*/
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/EntryEvent.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/EntryEvent.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/EntryEvent.java
index 2b69ac7..dd8cf38 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/EntryEvent.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/EntryEvent.java
@@ -12,6 +12,14 @@ package com.gemstone.gemfire.cache;
/** Contains information about an event affecting an entry, including
* its identity and the the circumstances of the event.
* It is passed in to <code>CacheListener</code>, <code>CapacityController</code>, and <code>CacheWriter</code>.
+ * <p>
+ * If this event originated from a region stored off heap then this event can
+ * only be used as long as the notification method that obtained it has not returned.
+ * For example in your implementation of {@link CacheListener#afterUpdate(EntryEvent)} the event parameter
+ * is only valid until your afterUpdate method returns. It is not safe to store instances of this
+ * class and use them later when using off heap storage.
+ * Attempts to access off-heap data from this event after it has expired will result in an
+ * IllegalStateException.
*
* @author Eric Zoerner
*
@@ -41,6 +49,7 @@ public interface EntryEvent<K,V> extends CacheEvent<K,V> {
* @return the old value in the cache prior to this event.
* If the entry did not exist, was invalid, or was not available,
* then null is returned.
+ * @throws IllegalStateException if off-heap and called after the method that was passed this EntryEvent returns.
*/
public V getOldValue();
@@ -48,6 +57,7 @@ public interface EntryEvent<K,V> extends CacheEvent<K,V> {
* Returns the serialized form of the value in the cache before this event.
*
* @return the serialized form of the value in the cache before this event
+ * @throws IllegalStateException if off-heap and called after the method that was passed this EntryEvent returns.
*
* @since 5.5
*/
@@ -57,6 +67,7 @@ public interface EntryEvent<K,V> extends CacheEvent<K,V> {
* Returns the value in the cache after this event.
*
* @return the value in the cache after this event
+ * @throws IllegalStateException if off-heap and called after the method that was passed this EntryEvent returns.
*/
public V getNewValue();
@@ -64,6 +75,7 @@ public interface EntryEvent<K,V> extends CacheEvent<K,V> {
* Returns the serialized form of the value in the cache after this event.
*
* @return the serialized form of the value in the cache after this event
+ * @throws IllegalStateException if off-heap and called after the method that was passed this EntryEvent returns.
*
* @since 5.5
*/
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/EvictionCriteria.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/EvictionCriteria.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/EvictionCriteria.java
new file mode 100644
index 0000000..994edba
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/EvictionCriteria.java
@@ -0,0 +1,50 @@
+/*=========================================================================
+ * Copyright (c) 2009-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. GoPivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+
+package com.gemstone.gemfire.cache;
+
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * Interface implemented by an EVICTION BY CRITERIA of
+ * {@link CustomEvictionAttributes}. This will be invoked by periodic evictor
+ * task that will get the keys to be evicted using this and then destroy from
+ * the region to which this is attached.
+ *
+ * @author swale
+ * @since gfxd 1.0
+ */
+public interface EvictionCriteria<K, V> {
+
+ /**
+ * Get the (key, routing object) of the entries to be evicted from region
+ * satisfying EVICTION BY CRITERIA at this point of time.
+ * <p>
+ * The returned Map.Entry object by the Iterator may be reused internally so
+ * caller must extract the key, routing object from the entry on each
+ * iteration.
+ */
+ Iterator<Map.Entry<K, Object>> getKeysToBeEvicted(long currentMillis,
+ Region<K, V> region);
+
+ /**
+ * Last moment check if an entry should be evicted or not applying the
+ * EVICTION BY CRITERIA again under the region entry lock in case the entry
+ * has changed after the check in {@link #getKeysToBeEvicted}.
+ */
+ boolean doEvict(EntryEvent<K, V> event);
+
+ /**
+ * Return true if this eviction criteria is equivalent to the other one. This
+ * is used to ensure that custom eviction is configured identically on all the
+ * nodes of a cluster hosting the region to which this eviction criteria has
+ * been attached.
+ */
+ boolean isEquivalent(EvictionCriteria<K, V> other);
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/GemFireCache.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/GemFireCache.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/GemFireCache.java
index 4ed4dfb..d81d25d 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/GemFireCache.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/GemFireCache.java
@@ -18,6 +18,9 @@ import com.gemstone.gemfire.LogWriter;
import com.gemstone.gemfire.cache.client.ClientCache;
import com.gemstone.gemfire.cache.client.ClientCacheFactory;
import com.gemstone.gemfire.cache.control.ResourceManager;
+import com.gemstone.gemfire.cache.hdfs.HDFSStore;
+import com.gemstone.gemfire.cache.hdfs.HDFSStoreFactory;
+import com.gemstone.gemfire.cache.lucene.LuceneService;
import com.gemstone.gemfire.cache.wan.GatewaySenderFactory;
import com.gemstone.gemfire.distributed.DistributedSystem;
import com.gemstone.gemfire.pdx.PdxSerializer;
@@ -251,4 +254,25 @@ public interface GemFireCache extends RegionService {
*/
public Properties getInitializerProps();
+ /**
+ * Returns the LuceneService singleton instance.
+ * @since 8.5
+ */
+ public LuceneService getLuceneService();
+
+ /**
+ * Returns the HDFSStore by name or <code>null</code> if no HDFSStore is
+ * found.
+ *
+ * @param name the name of the HDFSStore to find.
+ */
+ public HDFSStore findHDFSStore(String name);
+
+ /**
+ * Creates a {@link HDFSStoreFactory} for creating a {@link HDFSStore}
+ *
+ * @return the HDFS store factory
+ */
+ public HDFSStoreFactory createHDFSStoreFactory();
+
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Operation.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Operation.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Operation.java
index c2571fe..1ca3272 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Operation.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Operation.java
@@ -322,7 +322,8 @@ public final class Operation implements java.io.Serializable {
OP_TYPE_DESTROY,
OP_DETAILS_EVICT
);
-
+
+
/**
* A region load snapshot.
* @see Region#loadSnapshot
@@ -663,6 +664,19 @@ public final class Operation implements java.io.Serializable {
OP_DETAILS_REMOVEALL
);
+ /**
+ * An entry local destroy caused by an eviction.
+ * @see Region#localDestroy(Object)
+ */
+ public static final Operation CUSTOM_EVICT_DESTROY
+ = new Operation("EVICT_DESTROY",
+ false, // isLocal
+ false, // isRegion
+ OP_TYPE_DESTROY,
+ OP_DETAILS_EVICT
+ );
+
+
/** The name of this mirror type. */
private final transient String name;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/PartitionAttributesFactory.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/PartitionAttributesFactory.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/PartitionAttributesFactory.java
index 9dc4563..9deeb12 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/PartitionAttributesFactory.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/PartitionAttributesFactory.java
@@ -81,7 +81,7 @@ public class PartitionAttributesFactory<K,V>
* Default local max memory value in megabytes. By default each partitioned
* Region can contribute 90% of the maximum memory allocated to a VM.
*/
- private static int computeMaxMem()
+ static int computeMaxMem()
{
final long maxMemInMegabytes = Runtime.getRuntime().maxMemory() / (1024 * 1024);
final long maxMemoryToUse = (long) (maxMemInMegabytes * 0.90);
@@ -100,7 +100,11 @@ public class PartitionAttributesFactory<K,V>
/**
* The default maximum amount of memory to be used by this region in this
* process, in megabytes.
+ *
+ * @deprecated Use {@link
+ * PartitionAttributesImpl#getLocalMaxMemoryDefault()} instead.
*/
+ @Deprecated
public static final int LOCAL_MAX_MEMORY_DEFAULT = computeMaxMem();
/**
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Region.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Region.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Region.java
index 9c02030..05bebe7 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Region.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/Region.java
@@ -709,6 +709,7 @@ public interface Region<K,V> extends ConcurrentMap<K, V> {
* In some cases <code>null</code> may be returned even
* if a previous value exists.
* If the region is a client proxy then <code>null</code> is returned.
+ * If the region is off-heap and the old value was stored in off-heap memory then <code>null</code> is returned.
* If the region is partitioned and the put is done on a non-primary then <code>null</code> is returned.
* If the value is not currently stored in memory but is on disk
* and if the region does not have cqs
@@ -905,6 +906,7 @@ public interface Region<K,V> extends ConcurrentMap<K, V> {
* In some cases <code>null</code> may be returned even
* if a previous value exists.
* If the region is a client proxy then <code>null</code> is returned.
+ * If the region is off-heap and the old value was stored in off-heap memory then <code>null</code> is returned.
* If the region is partitioned and the destroy is done on a non-primary then <code>null</code> is returned.
* If the value is not currently stored in memory but is on disk
* and if the region does not have cqs
@@ -940,6 +942,7 @@ public interface Region<K,V> extends ConcurrentMap<K, V> {
* In some cases <code>null</code> may be returned even
* if a previous value exists.
* If the region is a client proxy then <code>null</code> is returned.
+ * If the region is off-heap and the old value was stored in off-heap memory then <code>null</code> is returned.
* If the region is partitioned and the destroy is done on a non-primary then <code>null</code> is returned.
* If the value is not currently stored in memory but is on disk
* and if the region does not have cqs
@@ -1554,6 +1557,7 @@ public interface Region<K,V> extends ConcurrentMap<K, V> {
* In some cases <code>null</code> may be returned even
* if a previous value exists.
* If the region is a client proxy then <code>null</code> is returned.
+ * If the region is off-heap and the old value was stored in off-heap memory then <code>null</code> is returned.
* If the region is partitioned and the remove is done on a non-primary then <code>null</code> is returned.
* If the value is not currently stored in memory but is on disk
* and if the region does not have cqs
@@ -2199,6 +2203,8 @@ public boolean containsKeyOnServer(Object key);
* return true;
* } else return false;</pre>
* except that the action is performed atomically.
+ *<p>As of 8.5, if value is an Array then Arrays.equals or Arrays.deepEquals is used
+ *instead of Object.equals.
*<p>ConcurrentMap operations are supported on partitioned and replicated regions
*and in client caches. They are also supported on non-empty local regions.<p>
*<p>Please read the notes on ConcurrentMap operations in the javadoc for Region.</p>
@@ -2230,6 +2236,8 @@ public boolean containsKeyOnServer(Object key);
* return true;
* } else return false;</pre>
* except that the action is performed atomically.
+ *<p>As of 8.5, if value is an Array then Arrays.equals or Arrays.deepEquals is used
+ *instead of Object.equals.
*<p>ConcurrentMap operations are supported on partitioned and replicated regions
*and in client caches. They are also supported on non-empty local regions.</p>
*<p>Please read the notes on ConcurrentMap operations in the javadoc for Region.</p>
@@ -2373,6 +2381,7 @@ public boolean containsKeyOnServer(Object key);
* In some cases <code>null</code> may be returned even
* if a previous value exists.
* If the region is a client proxy then <code>null</code> is returned.
+ * If the region is off-heap and the old value was stored in off-heap memory then <code>null</code> is returned.
* If the region is partitioned and the setValue is done on a non-primary then <code>null</code> is returned.
* If the value is not currently stored in memory but is on disk
* and if the region does not have cqs
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionAttributes.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionAttributes.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionAttributes.java
index b97e744..dcb7ff0 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionAttributes.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionAttributes.java
@@ -11,6 +11,7 @@ import java.io.File;
import java.util.Set;
import com.gemstone.gemfire.cache.client.Pool;
+import com.gemstone.gemfire.cache.hdfs.HDFSStore;
import com.gemstone.gemfire.compression.Compressor;
/** Defines attributes for configuring a region.
@@ -139,6 +140,11 @@ public interface RegionAttributes<K,V> {
*/
public EvictionAttributes getEvictionAttributes();
+ /**
+ * Return the {@link CustomEvictionAttributes}, if any, set for the region.
+ */
+ public CustomEvictionAttributes getCustomEvictionAttributes();
+
/** Returns the cache listener for the region.
* @throws IllegalStateException if more than one cache listener exists on this attributes
* @return the region's <code>CacheListener</code>
@@ -439,11 +445,36 @@ public interface RegionAttributes<K,V> {
public boolean getConcurrencyChecksEnabled();
/**
+ * Returns the name of the {@link HDFSStore} that this region belongs
+ * to, if any.
+ * @return the name of the {@link HDFSStore} of this region;
+ * <code>null</code> is returned if this region has no
+ * {@link HDFSStore}.
+ * @since 9.0
+ */
+ public String getHDFSStoreName();
+
+ /**
+ * Returns true if this region is configured to
+ * be write-only to HDFS.
+ * @since 9.0
+ */
+ public boolean getHDFSWriteOnly();
+
+ /**
* Returns the compressor used by this region's entry values.
* @since 8.0
* @return null if the region does not have compression enabled.
*/
public Compressor getCompressor();
+
+ /**
+ * Returns whether or not this region uses off-heap memory.
+ * @return True if a usage of off-heap memory is enabled;
+ * false if usage of off-heap memory is disabled (default).
+ * @since 9.0
+ */
+ public boolean getOffHeap();
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionFactory.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionFactory.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionFactory.java
index 7de43a7..40041cb 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionFactory.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionFactory.java
@@ -393,6 +393,30 @@ public class RegionFactory<K,V>
}
/**
+ * Set custom {@link EvictionCriteria} for the region with start time and
+ * interval of evictor task to be run in milliseconds, or evict incoming rows
+ * in case both start and frequency are specified as zero.
+ *
+ * @param criteria
+ * an {@link EvictionCriteria} to be used for eviction for HDFS
+ * persistent regions
+ * @param start
+ * the start time at which periodic evictor task should be first
+ * fired to apply the provided {@link EvictionCriteria}; if this is
+ * zero then current time is used for the first invocation of evictor
+ * @param interval
+ * the periodic frequency at which to run the evictor task after the
+ * initial start; if this is if both start and frequency are zero
+ * then {@link EvictionCriteria} is applied on incoming insert/update
+ * to determine whether it is to be retained
+ */
+ public RegionFactory<K, V> setCustomEvictionAttributes(
+ EvictionCriteria<K, V> criteria, long start, long interval) {
+ this.attrsFactory.setCustomEvictionAttributes(criteria, start, interval);
+ return this;
+ }
+
+ /**
* Sets the scope for the next <code>RegionAttributes</code> created.
*
* @param scopeType
@@ -878,6 +902,31 @@ public class RegionFactory<K,V>
this.attrsFactory.addAsyncEventQueueId(asyncEventQueueId);
return this;
}
+ /**
+ * Sets the HDFSStore name attribute.
+ * This causes the region to belong to the HDFSStore.
+ * @param name the name of the hdfsstore
+ * @return a reference to this RegionFactory object
+ *
+ * @see AttributesFactory#setHDFSStoreName
+ * @since 9.0
+ */
+ public RegionFactory<K,V> setHDFSStoreName(String name) {
+ this.attrsFactory.setHDFSStoreName(name);
+ return this;
+ }
+
+ /**
+ * Sets the HDFS write only attribute. if the region
+ * is configured to be write only to HDFS, events that have
+ * been evicted from memory cannot be read back from HDFS.
+ * Events are written to HDFS in the order in which they occurred.
+ * @since 9.0
+ */
+ public RegionFactory<K,V> setHDFSWriteOnly(boolean writeOnly) {
+ this.attrsFactory.setHDFSWriteOnly(writeOnly);
+ return this;
+ }
/**
* Set the compressor to be used by this region for compressing
@@ -890,4 +939,14 @@ public class RegionFactory<K,V>
this.attrsFactory.setCompressor(compressor);
return this;
}
+
+ /**
+ * Enables this region's usage of off-heap memory if true.
+ * @param offHeap boolean flag to enable off-heap memory
+ * @since 9.0
+ */
+ public RegionFactory<K,V> setOffHeap(boolean offHeap) {
+ this.attrsFactory.setOffHeap(offHeap);
+ return this;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionShortcut.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionShortcut.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionShortcut.java
index 5000032..ae3cbdb 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionShortcut.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/RegionShortcut.java
@@ -226,4 +226,54 @@ public enum RegionShortcut {
* The actual RegionAttributes for a REPLICATE_PROXY region set the {@link DataPolicy} to {@link DataPolicy#EMPTY} and {@link Scope} to {@link Scope#DISTRIBUTED_ACK}.
*/
REPLICATE_PROXY,
+
+ /**
+ * A PARTITION_HDFS has local state that is partitioned across each peer member
+ * that created the region.
+ * In addition its state is written to HDFS.
+ * The random access to the data in HDFS is also enabled.
+ * The actual RegionAttributes for a PARTITION_HDFS region set the {@link DataPolicy} to {@link DataPolicy#HDFS_PARTITION}.
+ * The HDFS event queue's property random-access is set to true.
+ * The {@link EvictionAttributes} are set to {@link EvictionAlgorithm#LRU_HEAP}
+ * with {@link EvictionAction#OVERFLOW_TO_DISK}.
+ */
+ PARTITION_HDFS,
+
+ /**
+ * A PARTITION_REDUNDANT_HDFS has local state that is partitioned across each peer member
+ * that created the region.
+ * In addition its state is written to HDFS and recovered from HDFS when the region is
+ * created. The random access to the data in HDFS is also enabled.
+ * In addition an extra copy of the data is kept in memory.
+ * The actual RegionAttributes for a PARTITION_REDUNDANT_HDFS region set the {@link DataPolicy} to {@link DataPolicy#HDFS_PARTITION}
+ * and the redundant-copies to 1. The HDFS event queue's property random-access is set to true.
+ * The {@link EvictionAttributes} are set to {@link EvictionAlgorithm#LRU_HEAP}
+ * with {@link EvictionAction#OVERFLOW_TO_DISK}.
+ */
+ PARTITION_REDUNDANT_HDFS,
+
+ /**
+ * A PARTITION_WRITEONLY_HDFS_STORE has local state that is partitioned across each peer member
+ * that created the region.
+ * In addition its state is written to HDFS and recovered from HDFS when the region is
+ * created. The random access to the data in HDFS is disabled.
+ * The actual RegionAttributes for a PARTITION_WRITEONLY_HDFS_STORE region set the {@link DataPolicy} to {@link DataPolicy#HDFS_PARTITION}.
+ * The HDFS event queue's property write only is set as true.
+ * The {@link EvictionAttributes} are set to {@link EvictionAlgorithm#LRU_HEAP}
+ * with {@link EvictionAction#OVERFLOW_TO_DISK}.
+ */
+ PARTITION_WRITEONLY_HDFS_STORE,
+
+ /**
+ * A PARTITION_REDUNDANT_WRITEONLY_HDFS_STORE has local state that is partitioned across each peer member
+ * that created the region.
+ * In addition its state is written to HDFS and recovered from HDFS when the region is
+ * created. The random access to the data in HDFS is disabled.
+ * In addition an extra copy of the data is kept in memory.
+ * The actual RegionAttributes for a PARTITION_REDUNDANT_WRITEONLY_HDFS_STORE region set the {@link DataPolicy} to {@link DataPolicy#HDFS_PARTITION}
+ * and the redundant-copies to 1. The HDFS event queue's property write only is set as true.
+ * The {@link EvictionAttributes} are set to {@link EvictionAlgorithm#LRU_HEAP}
+ * with {@link EvictionAction#OVERFLOW_TO_DISK}.
+ */
+ PARTITION_REDUNDANT_WRITEONLY_HDFS_STORE
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/SerializedCacheValue.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/SerializedCacheValue.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/SerializedCacheValue.java
index 33f60f3..a78660d 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/SerializedCacheValue.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/SerializedCacheValue.java
@@ -20,6 +20,8 @@ package com.gemstone.gemfire.cache;
* by calling {@link EntryEvent#getSerializedOldValue()} then the SerializedCacheValue returned
* is only valid until your afterUpdate method returns. It is not safe to store instances of this
* class and use them later when using off heap storage.
+ * Attempts to access off-heap data from this event after it has expired will result in an
+ * IllegalStateException.
*
* @author Barry Oglesby
* @since 5.5
@@ -29,12 +31,14 @@ public interface SerializedCacheValue<V> {
/**
* Returns the raw byte[] that represents this cache value.
* @return the raw byte[] that represents this cache value
+ * @throws IllegalStateException if off-heap and called after the method that was passed the EntryEvent that this serialize value was obtained from returns and this value does not already have the serialized value.
*/
public byte[] getSerializedValue();
/**
* Returns the deserialized object for this cache value.
* @return the deserialized object for this cache value
+ * @throws IllegalStateException if off-heap and called after the method that was passed the EntryEvent that this serialize value was obtained from returns.
*/
public V getDeserializedValue();
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/asyncqueue/internal/AsyncEventQueueFactoryImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/asyncqueue/internal/AsyncEventQueueFactoryImpl.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/asyncqueue/internal/AsyncEventQueueFactoryImpl.java
index 203c444..f413218 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/asyncqueue/internal/AsyncEventQueueFactoryImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/asyncqueue/internal/AsyncEventQueueFactoryImpl.java
@@ -262,10 +262,12 @@ public class AsyncEventQueueFactoryImpl implements AsyncEventQueueFactory {
this.attrs.maximumQueueMemory = asyncQueueCreation.getMaximumQueueMemory();
this.attrs.isParallel = asyncQueueCreation.isParallel();
this.attrs.isBucketSorted = ((AsyncEventQueueCreation)asyncQueueCreation).isBucketSorted();
+ this.attrs.isHDFSQueue = ((AsyncEventQueueCreation)asyncQueueCreation).isHDFSQueue();
this.attrs.dispatcherThreads = asyncQueueCreation.getDispatcherThreads();
this.attrs.policy = asyncQueueCreation.getOrderPolicy();
this.attrs.eventFilters = asyncQueueCreation.getGatewayEventFilters();
this.attrs.eventSubstitutionFilter = asyncQueueCreation.getGatewayEventSubstitutionFilter();
+
}
public AsyncEventQueueFactory setParallel(boolean isParallel) {
@@ -276,4 +278,8 @@ public class AsyncEventQueueFactoryImpl implements AsyncEventQueueFactory {
this.attrs.isBucketSorted = isbucketSorted;
return this;
}
+ public AsyncEventQueueFactory setIsHDFSQueue(boolean isHDFSQueue) {
+ this.attrs.isHDFSQueue = isHDFSQueue;
+ return this;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/asyncqueue/internal/AsyncEventQueueImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/asyncqueue/internal/AsyncEventQueueImpl.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/asyncqueue/internal/AsyncEventQueueImpl.java
index 5e621f1..71e8d2a 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/asyncqueue/internal/AsyncEventQueueImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/asyncqueue/internal/AsyncEventQueueImpl.java
@@ -180,6 +180,11 @@ public class AsyncEventQueueImpl implements AsyncEventQueue {
return sender.isParallel();
}
+
+ public void destroy() {
+ ((AbstractGatewaySender)this.sender).destroy();
+ }
+
public boolean isBucketSorted() {
// TODO Auto-generated method stub
return false;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ClientMetadataService.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ClientMetadataService.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ClientMetadataService.java
index f3c17b5..629a9b6 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ClientMetadataService.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ClientMetadataService.java
@@ -66,6 +66,8 @@ public final class ClientMetadataService {
/** random number generator used in pruning */
private final Random rand = new Random();
+ private volatile boolean isMetadataStable = true;
+
public ClientMetadataService(Cache cache) {
this.cache = cache;
}
@@ -264,13 +266,14 @@ public final class ClientMetadataService {
ClientPartitionAdvisor prAdvisor, Set<Integer> bucketSet,
boolean primaryOnly) {
if (primaryOnly) {
- Set<Integer> bucketsWithoutServer = new HashSet<Integer>();
HashMap<ServerLocation, HashSet<Integer>> serverToBucketsMap = new HashMap<ServerLocation, HashSet<Integer>>();
for (Integer bucketId : bucketSet) {
ServerLocation server = prAdvisor.advisePrimaryServerLocation(bucketId);
if (server == null) {
- bucketsWithoutServer.add(bucketId);
- continue;
+ //If we don't have the metadata for some buckets, return
+ //null, indicating that we don't have any metadata. This
+ //will cause us to use the non-single hop path.
+ return null;
}
HashSet<Integer> buckets = serverToBucketsMap.get(server);
if (buckets == null) {
@@ -280,11 +283,6 @@ public final class ClientMetadataService {
buckets.add(bucketId);
}
- if (!serverToBucketsMap.isEmpty() ) {
- serverToBucketsMap.entrySet().iterator().next().getValue().addAll(
- bucketsWithoutServer);
- }
-
if (logger.isDebugEnabled()) {
logger.debug("ClientMetadataService: The server to bucket map is : {}", serverToBucketsMap);
}
@@ -304,7 +302,6 @@ public final class ClientMetadataService {
if (isDebugEnabled) {
logger.debug("ClientMetadataService: The buckets to be pruned are: {}", buckets);
}
- Set<Integer> bucketSetWithoutServer = new HashSet<Integer>();
HashMap<ServerLocation, HashSet<Integer>> serverToBucketsMap = new HashMap<ServerLocation, HashSet<Integer>>();
HashMap<ServerLocation, HashSet<Integer>> prunedServerToBucketsMap = new HashMap<ServerLocation, HashSet<Integer>>();
@@ -314,9 +311,11 @@ public final class ClientMetadataService {
if (isDebugEnabled) {
logger.debug("ClientMetadataService: For bucketId {} the server list is {}", bucketId, serversList);
}
- if ((serversList == null || serversList.size() == 0) ) {
- bucketSetWithoutServer.add(bucketId);
- continue;
+ if (serversList == null || serversList.size() == 0) {
+ //If we don't have the metadata for some buckets, return
+ //null, indicating that we don't have any metadata. This
+ //will cause us to use the non-single hop path.
+ return null;
}
if (isDebugEnabled) {
@@ -385,9 +384,6 @@ public final class ClientMetadataService {
}
serverToBucketsMap.remove(server);
}
- prunedServerToBucketsMap.entrySet().iterator().next().getValue().addAll(
- bucketSetWithoutServer);
-
if (isDebugEnabled) {
logger.debug("ClientMetadataService: The final prunedServerToBucket calculated is : {}", prunedServerToBucketsMap);
@@ -508,6 +504,7 @@ public final class ClientMetadataService {
if(this.nonPRs.contains(region.getFullPath())){
return;
}
+ this.setMetadataStable(false);
region.getCachePerfStats().incNonSingleHopsCount();
if (isRecursive) {
try {
@@ -830,6 +827,14 @@ public final class ClientMetadataService {
return HONOUR_SERVER_GROUP_IN_PR_SINGLE_HOP;
}
+ public boolean isMetadataStable() {
+ return isMetadataStable;
+ }
+
+ public void setMetadataStable(boolean isMetadataStable) {
+ this.isMetadataStable = isMetadataStable;
+ }
+
private boolean isMetadataRefreshed_TEST_ONLY = false;
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ClientRegionFactoryImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ClientRegionFactoryImpl.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ClientRegionFactoryImpl.java
index 7526ff7..3daf4ec 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ClientRegionFactoryImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ClientRegionFactoryImpl.java
@@ -54,7 +54,7 @@ public class ClientRegionFactoryImpl<K,V> implements ClientRegionFactory<K,V>
RegionAttributes ra = cache.getRegionAttributes(pra.toString());
if (ra == null) {
throw new IllegalStateException("The region shortcut " + pra
- + " has been removed.");
+ + " has been removed from " + cache.listRegionAttributes());
}
this.attrsFactory = new AttributesFactory<K,V>(ra);
initAttributeFactoryDefaults();
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/Connection.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/Connection.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/Connection.java
index 99bf77f..4018a40 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/Connection.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/Connection.java
@@ -10,6 +10,7 @@ package com.gemstone.gemfire.cache.client.internal;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
+import java.net.SocketException;
import java.nio.ByteBuffer;
import com.gemstone.gemfire.distributed.internal.ServerLocation;
@@ -25,7 +26,7 @@ public interface Connection {
public static final long DEFAULT_CONNECTION_ID = 26739;
public Socket getSocket();
- public ByteBuffer getCommBuffer();
+ public ByteBuffer getCommBuffer() throws SocketException;
public ConnectionStats getStats();
/**
* Forcefully close the resources used by this connection.
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ConnectionFactoryImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ConnectionFactoryImpl.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ConnectionFactoryImpl.java
index ae4b851..7538b04 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ConnectionFactoryImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ConnectionFactoryImpl.java
@@ -159,10 +159,11 @@ public class ConnectionFactoryImpl implements ConnectionFactory {
private void authenticateIfRequired(Connection conn) {
cancelCriterion.checkCancelInProgress(null);
if (!pool.isUsedByGateway() && !pool.getMultiuserAuthentication()) {
- if (conn.getServer().getRequiresCredentials()) {
- if (conn.getServer().getUserId() == -1) {
+ ServerLocation server = conn.getServer();
+ if (server.getRequiresCredentials()) {
+ if (server.getUserId() == -1) {
Long uniqueID = (Long)AuthenticateUserOp.executeOn(conn, pool);
- conn.getServer().setUserId(uniqueID);
+ server.setUserId(uniqueID);
if (logger.isDebugEnabled()) {
logger.debug("CFI.authenticateIfRequired() Completed authentication on {}", conn);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ConnectionImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ConnectionImpl.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ConnectionImpl.java
index eca73fc..43cff54 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ConnectionImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ConnectionImpl.java
@@ -13,6 +13,7 @@ import java.io.OutputStream;
import java.net.InetAddress;
import java.net.NoRouteToHostException;
import java.net.Socket;
+import java.net.SocketException;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -124,10 +125,10 @@ public class ConnectionImpl implements Connection {
out = SocketUtils.getOutputStream(theSocket);//theSocket.getOutputStream();
in = SocketUtils.getInputStream(theSocket);//theSocket.getInputStream();
this.status = handShake.greet(this, location, communicationMode);
- commBuffer = ServerConnection.allocateCommBuffer(socketBufferSize);
+ commBuffer = ServerConnection.allocateCommBuffer(socketBufferSize, theSocket);
if (sender != null) {
commBufferForAsyncRead = ServerConnection
- .allocateCommBuffer(socketBufferSize);
+ .allocateCommBuffer(socketBufferSize, theSocket);
}
theSocket.setSoTimeout(readTimeout);
endpoint = endpointManager.referenceEndpoint(location, this.status.getMemberId());
@@ -208,9 +209,29 @@ public class ConnectionImpl implements Connection {
logger.debug(e.getMessage(), e);
}
}
+ releaseCommBuffers();
+ }
+
+ private void releaseCommBuffers() {
+ ByteBuffer bb = this.commBuffer;
+ if (bb != null) {
+ this.commBuffer = null;
+ ServerConnection.releaseCommBuffer(bb);
+ }
+ bb = this.commBufferForAsyncRead;
+ if (bb != null) {
+ this.commBufferForAsyncRead = null;
+ ServerConnection.releaseCommBuffer(bb);
+ }
}
- public ByteBuffer getCommBuffer() {
+ public ByteBuffer getCommBuffer() throws SocketException {
+ if (isDestroyed()) {
+ // see bug 52193. Since the code used to see this
+ // as an attempt to use a close socket just throw
+ // a SocketException.
+ throw new SocketException("socket was closed");
+ }
return commBuffer;
}
@@ -313,7 +334,13 @@ public class ConnectionImpl implements Connection {
TEST_DURABLE_CLIENT_CRASH = v;
}
- public ByteBuffer getCommBufferForAsyncRead() {
+ public ByteBuffer getCommBufferForAsyncRead() throws SocketException {
+ if (isDestroyed()) {
+ // see bug 52193. Since the code used to see this
+ // as an attempt to use a close socket just throw
+ // a SocketException.
+ throw new SocketException("socket was closed");
+ }
return commBufferForAsyncRead;
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java
index 5ea03f8..5a03f47 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java
@@ -72,7 +72,7 @@ public class ExecuteRegionFunctionSingleHopOp {
hasResult, resultCollector, cms, allBuckets);
reexecute = SingleHopClientExecutor.submitAllHA(callableTasks,
- (LocalRegion)region, resultCollector, failedNodes);
+ (LocalRegion)region, function.isHA(), resultCollector, failedNodes);
if (isDebugEnabled) {
logger.debug("ExecuteRegionFunctionSingleHopOp#execute : The size of callableTask is : {}", callableTasks.size());
@@ -85,11 +85,11 @@ public class ExecuteRegionFunctionSingleHopOp {
serverRegionExecutor, resultCollector, hasResult, failedNodes,
maxRetryAttempts - 1);
}
- else {
- ExecuteRegionFunctionOp.execute(pool, region.getFullPath(), function,
- serverRegionExecutor, resultCollector, hasResult,
- maxRetryAttempts - 1);
- }
+// else {
+// ExecuteRegionFunctionOp.execute(pool, region.getFullPath(), function,
+// serverRegionExecutor, resultCollector, hasResult,
+// maxRetryAttempts - 1);
+// }
}
resultCollector.endResults();
@@ -120,7 +120,7 @@ public class ExecuteRegionFunctionSingleHopOp {
hasResult, resultCollector, cms, allBuckets, isHA,optimizeForWrite);
reexecute = SingleHopClientExecutor.submitAllHA(callableTasks,
- (LocalRegion)region, resultCollector, failedNodes);
+ (LocalRegion)region, isHA, resultCollector, failedNodes);
if (isDebugEnabled) {
logger.debug("ExecuteRegionFunctionSingleHopOp#execute : The size of callableTask is: {}, reexecute={}", callableTasks.size(), reexecute);
@@ -133,11 +133,11 @@ public class ExecuteRegionFunctionSingleHopOp {
functionId, serverRegionExecutor, resultCollector, hasResult,
failedNodes, maxRetryAttempts - 1, isHA, optimizeForWrite);
}
- else {
- ExecuteRegionFunctionOp.execute(pool, region.getFullPath(), functionId,
- serverRegionExecutor, resultCollector, hasResult,
- maxRetryAttempts - 1, isHA, optimizeForWrite);
- }
+// else {
+// ExecuteRegionFunctionOp.execute(pool, region.getFullPath(), functionId,
+// serverRegionExecutor, resultCollector, hasResult,
+// maxRetryAttempts - 1, isHA, optimizeForWrite);
+// }
}
resultCollector.endResults();
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPRMetaDataOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPRMetaDataOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPRMetaDataOp.java
index b334ea0..a13b15e 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPRMetaDataOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/GetClientPRMetaDataOp.java
@@ -117,6 +117,7 @@ public class GetClientPRMetaDataOp {
if (isDebugEnabled) {
logger.debug("GetClientPRMetaDataOpImpl#processResponse: received ClientPRMetadata from server successfully.");
}
+ cms.setMetadataStable(true);
return null;
case MessageType.EXCEPTION:
if (logger.isDebugEnabled()) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PoolImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PoolImpl.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PoolImpl.java
index 1c22037..851c5c8 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PoolImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PoolImpl.java
@@ -1216,7 +1216,7 @@ public class PoolImpl implements InternalPool {
/**
* A debug flag used for testing used in BridgeObserver
*/
- public static volatile boolean BEFORE_RECOVER_INTERST_CALLBACK_FLAG = false;
+ public static volatile boolean BEFORE_RECOVER_INTEREST_CALLBACK_FLAG = false;
/**
* A debug flag used for testing used in BridgeObserver
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ProxyCache.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ProxyCache.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ProxyCache.java
index 1fb9cf0..f4abb89 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ProxyCache.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ProxyCache.java
@@ -68,12 +68,12 @@ public class ProxyCache implements RegionService {
if (this.proxyQueryService != null) {
this.proxyQueryService.closeCqs(keepAlive);
}
- UserAttributes.userAttributes.set(this.getUserAttributes());
+ UserAttributes.userAttributes.set(this.userAttributes);
Iterator<ServerLocation> iter = this.userAttributes.getServerToId()
.keySet().iterator();
while (iter.hasNext()) {
ProxyCacheCloseOp.executeOn(iter.next(), (PoolImpl)this.userAttributes.getPool(),
- getProperties(), keepAlive);
+ this.userAttributes.getCredentials(), keepAlive);
}
ArrayList<ProxyCache> proxyCache = ((PoolImpl)this.userAttributes.getPool()).getProxyCacheList();
synchronized (proxyCache) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PutAllOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PutAllOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PutAllOp.java
index a112bda..782cad9 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PutAllOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PutAllOp.java
@@ -36,6 +36,7 @@ import com.gemstone.gemfire.internal.cache.tier.sockets.Message;
import com.gemstone.gemfire.internal.cache.tier.sockets.Part;
import com.gemstone.gemfire.internal.cache.tier.sockets.VersionedObjectList;
import com.gemstone.gemfire.internal.logging.LogService;
+import com.gemstone.gemfire.internal.offheap.StoredObject;
/**
* Does a region putAll on a server
@@ -297,7 +298,10 @@ public class PutAllOp {
getMessage().addStringOrObjPart(key);
Object value = mapEntry.getValue();
if (value instanceof CachedDeserializable) {
- {
+ if (value instanceof StoredObject && !((StoredObject) value).isSerialized()) {
+ // it is a byte[]
+ getMessage().addObjPart(((StoredObject) value).getDeserializedForReading());
+ } else {
Object cdValue = ((CachedDeserializable)value).getValue();
if (cdValue instanceof byte[]) {
getMessage().addRawPart((byte[])cdValue, true);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PutOp.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PutOp.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PutOp.java
index 56181e2..35822ed 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PutOp.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PutOp.java
@@ -26,6 +26,7 @@ import com.gemstone.gemfire.internal.cache.tier.sockets.Message;
import com.gemstone.gemfire.internal.cache.tier.sockets.Part;
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.logging.LogService;
+import com.gemstone.gemfire.internal.offheap.StoredObject;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
@@ -216,7 +217,11 @@ public class PutOp {
}
}
else if (value instanceof CachedDeserializable) {
- {
+ if (value instanceof StoredObject && !((StoredObject) value).isSerialized()) {
+ // it is a byte[]
+ getMessage().addObjPart(Boolean.FALSE);
+ getMessage().addObjPart(((StoredObject) value).getDeserializedForReading());
+ } else {
getMessage().addObjPart(Boolean.FALSE);
Object cdValue = ((CachedDeserializable)value).getValue();
if (cdValue instanceof byte[]) {
@@ -274,7 +279,11 @@ public class PutOp {
}
}
else if (value instanceof CachedDeserializable) {
- {
+ if (value instanceof StoredObject && !((StoredObject) value).isSerialized()) {
+ // it is a byte[]
+ getMessage().addObjPart(Boolean.FALSE);
+ getMessage().addObjPart(((StoredObject) value).getDeserializedForReading());
+ } else {
getMessage().addObjPart(Boolean.FALSE);
Object cdValue = ((CachedDeserializable)value).getValue();
if (cdValue instanceof byte[]) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueConnectionImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueConnectionImpl.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueConnectionImpl.java
index 7a4f2ff..d618bed 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueConnectionImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueConnectionImpl.java
@@ -10,6 +10,7 @@ package com.gemstone.gemfire.cache.client.internal;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
+import java.net.SocketException;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -122,7 +123,7 @@ public class QueueConnectionImpl implements Connection {
return shouldDestroy;
}
- public ByteBuffer getCommBuffer() {
+ public ByteBuffer getCommBuffer() throws SocketException {
return getConnection().getCommBuffer();
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueManagerImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueManagerImpl.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueManagerImpl.java
index 71357e3..f0bf9e9 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueManagerImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueManagerImpl.java
@@ -1206,7 +1206,7 @@ public class QueueManagerImpl implements QueueManager {
protected void recoverAllInterestTypes(final Connection recoveredConnection,
boolean isFirstNewConnection) {
- if (PoolImpl.BEFORE_RECOVER_INTERST_CALLBACK_FLAG) {
+ if (PoolImpl.BEFORE_RECOVER_INTEREST_CALLBACK_FLAG) {
BridgeObserver bo = BridgeObserverHolder.getInstance();
bo.beforeInterestRecovery();
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueStateImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueStateImpl.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueStateImpl.java
index 3dda60b..9a6b3db 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueStateImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/QueueStateImpl.java
@@ -110,7 +110,8 @@ public class QueueStateImpl implements QueueState {
return this.invalidateCount.get();
}
- /** test hook
+ /** test hook - access to this map should be synchronized on the
+ * map to avoid concurrent modification exceptions
*/
public Map getThreadIdToSequenceIdMap() {
return this.threadIdToSequenceId;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ServerRegionProxy.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ServerRegionProxy.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ServerRegionProxy.java
index a1cf4ba..7670ef9 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ServerRegionProxy.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/ServerRegionProxy.java
@@ -738,42 +738,42 @@ public class ServerRegionProxy extends ServerProxy implements ServerRegionDataAc
int retryAttempts = pool.getRetryAttempts();
if (this.pool.getPRSingleHopEnabled()) {
- ClientMetadataService cms = region.getCache()
- .getClientMetadataService();
- if (serverRegionExecutor.getFilter().isEmpty()) {
- HashMap<ServerLocation, HashSet<Integer>> serverToBuckets = cms
- .groupByServerToAllBuckets(this.region, function.optimizeForWrite());
- if (serverToBuckets == null || serverToBuckets.isEmpty()) {
- ExecuteRegionFunctionOp.execute(this.pool, rgnName, function,
- serverRegionExecutor, resultCollector, hasResult, retryAttempts);
- cms.scheduleGetPRMetaData(region, false);
- }
- else {
- ExecuteRegionFunctionSingleHopOp.execute(this.pool, this.region,
- function, serverRegionExecutor, resultCollector, hasResult,
- serverToBuckets, retryAttempts, true);
- }
- }
- else {
- boolean isBucketFilter = serverRegionExecutor.getExecuteOnBucketSetFlag();
- Map<ServerLocation, HashSet> serverToFilterMap = cms
- .getServerToFilterMap(serverRegionExecutor.getFilter(), region,
- function.optimizeForWrite(), isBucketFilter);
- if (serverToFilterMap == null || serverToFilterMap.isEmpty()) {
- ExecuteRegionFunctionOp.execute(this.pool, rgnName, function,
- serverRegionExecutor, resultCollector, hasResult, retryAttempts);
- cms.scheduleGetPRMetaData(region, false);
- }
- else {
- //Asif: In case of withBucketFilter , the serverToFilterMap is nothing but
- // serverToBucketsMap, so allBuckets flag should be true in that case
- // so allBuckets flag is governed by isBucketFilter flag
- ExecuteRegionFunctionSingleHopOp.execute(this.pool, this.region,
- function, serverRegionExecutor, resultCollector, hasResult,
- serverToFilterMap, retryAttempts, isBucketFilter);
-
-
+ ClientMetadataService cms = region.getCache().getClientMetadataService();
+ if (cms.isMetadataStable()) {
+ if (serverRegionExecutor.getFilter().isEmpty()) {
+ HashMap<ServerLocation, HashSet<Integer>> serverToBuckets = cms
+ .groupByServerToAllBuckets(this.region,
+ function.optimizeForWrite());
+ if (serverToBuckets == null || serverToBuckets.isEmpty()) {
+ ExecuteRegionFunctionOp
+ .execute(this.pool, rgnName, function, serverRegionExecutor,
+ resultCollector, hasResult, retryAttempts);
+ cms.scheduleGetPRMetaData(region, false);
+ } else {
+ ExecuteRegionFunctionSingleHopOp.execute(this.pool, this.region,
+ function, serverRegionExecutor, resultCollector, hasResult,
+ serverToBuckets, retryAttempts, true);
+ }
+ } else {
+ boolean isBucketFilter = serverRegionExecutor.getExecuteOnBucketSetFlag();
+ Map<ServerLocation, HashSet> serverToFilterMap = cms
+ .getServerToFilterMap(serverRegionExecutor.getFilter(), region,
+ function.optimizeForWrite(), isBucketFilter);
+ if (serverToFilterMap == null || serverToFilterMap.isEmpty()) {
+ ExecuteRegionFunctionOp
+ .execute(this.pool, rgnName, function, serverRegionExecutor,
+ resultCollector, hasResult, retryAttempts);
+ cms.scheduleGetPRMetaData(region, false);
+ } else {
+ ExecuteRegionFunctionSingleHopOp.execute(this.pool, this.region,
+ function, serverRegionExecutor, resultCollector, hasResult,
+ serverToFilterMap, retryAttempts, isBucketFilter);
+ }
}
+ } else {
+ cms.scheduleGetPRMetaData(region, false);
+ ExecuteRegionFunctionOp.execute(this.pool, rgnName, function,
+ serverRegionExecutor, resultCollector, hasResult, retryAttempts);
}
}
else {
@@ -793,39 +793,46 @@ public class ServerRegionProxy extends ServerProxy implements ServerRegionDataAc
Boolean.valueOf(isHA), Boolean.valueOf(optimizeForWrite));
int retryAttempts = pool.getRetryAttempts();
-
if (this.pool.getPRSingleHopEnabled()) {
ClientMetadataService cms = this.region.getCache()
- .getClientMetadataService();
- if (serverRegionExecutor.getFilter().isEmpty()) {
- HashMap<ServerLocation, HashSet<Integer>> serverToBuckets = cms
- .groupByServerToAllBuckets(this.region, optimizeForWrite);
- if (serverToBuckets == null || serverToBuckets.isEmpty()) {
- ExecuteRegionFunctionOp.execute(this.pool, rgnName, functionId,
- serverRegionExecutor, resultCollector, hasResult, retryAttempts, isHA, optimizeForWrite);
- cms.scheduleGetPRMetaData(this.region, false);
- }
- else {
- ExecuteRegionFunctionSingleHopOp.execute(this.pool, this.region,
- functionId, serverRegionExecutor, resultCollector, hasResult,
- serverToBuckets, retryAttempts, true, isHA, optimizeForWrite);
- }
- }
- else {
- boolean isBucketsAsFilter = serverRegionExecutor.getExecuteOnBucketSetFlag();
- Map<ServerLocation, HashSet> serverToFilterMap = cms
- .getServerToFilterMap(serverRegionExecutor.getFilter(), region,
- optimizeForWrite, isBucketsAsFilter);
- if (serverToFilterMap == null || serverToFilterMap.isEmpty()) {
- ExecuteRegionFunctionOp.execute(this.pool, rgnName, functionId,
- serverRegionExecutor, resultCollector, hasResult, retryAttempts, isHA, optimizeForWrite);
- cms.scheduleGetPRMetaData(region, false);
- }
- else {
- ExecuteRegionFunctionSingleHopOp.execute(this.pool, this.region,
- functionId, serverRegionExecutor, resultCollector, hasResult,
- serverToFilterMap, retryAttempts, false, isHA, optimizeForWrite);
+ .getClientMetadataService();
+ if (cms.isMetadataStable()) {
+ if (serverRegionExecutor.getFilter().isEmpty()) {
+ HashMap<ServerLocation, HashSet<Integer>> serverToBuckets = cms
+ .groupByServerToAllBuckets(this.region, optimizeForWrite);
+ if (serverToBuckets == null || serverToBuckets.isEmpty()) {
+ ExecuteRegionFunctionOp.execute(this.pool, rgnName, functionId,
+ serverRegionExecutor, resultCollector, hasResult,
+ retryAttempts, isHA, optimizeForWrite);
+ cms.scheduleGetPRMetaData(this.region, false);
+ } else {
+ ExecuteRegionFunctionSingleHopOp.execute(this.pool, this.region,
+ functionId, serverRegionExecutor, resultCollector, hasResult,
+ serverToBuckets, retryAttempts, true, isHA, optimizeForWrite);
+ }
+ } else {
+ boolean isBucketsAsFilter = serverRegionExecutor.getExecuteOnBucketSetFlag();
+ Map<ServerLocation, HashSet> serverToFilterMap = cms
+ .getServerToFilterMap(serverRegionExecutor.getFilter(), region,
+ optimizeForWrite, isBucketsAsFilter);
+ if (serverToFilterMap == null || serverToFilterMap.isEmpty()) {
+ ExecuteRegionFunctionOp.execute(this.pool, rgnName, functionId,
+ serverRegionExecutor, resultCollector, hasResult,
+ retryAttempts, isHA, optimizeForWrite);
+ cms.scheduleGetPRMetaData(region, false);
+ } else {
+ ExecuteRegionFunctionSingleHopOp
+ .execute(this.pool, this.region, functionId,
+ serverRegionExecutor, resultCollector, hasResult,
+ serverToFilterMap, retryAttempts, false, isHA,
+ optimizeForWrite);
+ }
}
+ } else {
+ cms.scheduleGetPRMetaData(region, false);
+ ExecuteRegionFunctionOp.execute(this.pool, rgnName, functionId,
+ serverRegionExecutor, resultCollector, hasResult, retryAttempts,
+ isHA, optimizeForWrite);
}
}
else {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/SingleHopClientExecutor.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/SingleHopClientExecutor.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/SingleHopClientExecutor.java
index ada3ab2..53da823 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/SingleHopClientExecutor.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/SingleHopClientExecutor.java
@@ -28,11 +28,14 @@ import com.gemstone.gemfire.cache.CacheClosedException;
import com.gemstone.gemfire.cache.client.ServerConnectivityException;
import com.gemstone.gemfire.cache.client.ServerOperationException;
import com.gemstone.gemfire.cache.client.internal.GetAllOp.GetAllOpImpl;
+import com.gemstone.gemfire.cache.execute.Function;
import com.gemstone.gemfire.cache.execute.FunctionException;
+import com.gemstone.gemfire.cache.execute.FunctionInvocationTargetException;
import com.gemstone.gemfire.cache.execute.ResultCollector;
import com.gemstone.gemfire.distributed.internal.ServerLocation;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.PutAllPartialResultException;
+import com.gemstone.gemfire.internal.cache.execute.BucketMovedException;
import com.gemstone.gemfire.internal.cache.execute.InternalFunctionInvocationTargetException;
import com.gemstone.gemfire.internal.cache.tier.sockets.VersionedObjectList;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
@@ -97,7 +100,7 @@ public class SingleHopClientExecutor {
}
}
- static boolean submitAllHA(List callableTasks, LocalRegion region,
+ static boolean submitAllHA(List callableTasks, LocalRegion region, boolean isHA,
ResultCollector rc, Set<String> failedNodes) {
ClientMetadataService cms = region.getCache()
@@ -144,12 +147,24 @@ public class SingleHopClientExecutor {
catch (CacheClosedException e) {
return false;
}
- cms.scheduleGetPRMetaData(region, false);
cms.removeBucketServerLocation(server);
+ cms.scheduleGetPRMetaData(region, false);
reexecute = true;
failedNodes.addAll(((InternalFunctionInvocationTargetException)ee
.getCause()).getFailedNodeSet());
rc.clearResults();
+ if (!isHA) {
+ if (ee.getCause().getCause() != null) {
+ throw new FunctionInvocationTargetException(ee.getCause()
+ .getCause());
+ } else {
+ throw new FunctionInvocationTargetException(
+ new BucketMovedException(
+ LocalizedStrings.FunctionService_BUCKET_MIGRATED_TO_ANOTHER_NODE
+ .toLocalizedString()));
+ }
+ }
+
}
else if (ee.getCause() instanceof FunctionException) {
if (isDebugEnabled) {
@@ -177,6 +192,10 @@ public class SingleHopClientExecutor {
cms.scheduleGetPRMetaData(region, false);
reexecute = true;
rc.clearResults();
+ if (!isHA) {
+ reexecute = false;
+ throw (ServerConnectivityException) ee.getCause();
+ }
}
else {
throw executionThrowable(ee.getCause());
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/pooling/PooledConnection.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/pooling/PooledConnection.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/pooling/PooledConnection.java
index db38b8b..3fcb2ab 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/pooling/PooledConnection.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/pooling/PooledConnection.java
@@ -10,6 +10,7 @@ package com.gemstone.gemfire.cache.client.internal.pooling;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
+import java.net.SocketException;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -278,7 +279,7 @@ class PooledConnection implements Connection {
}
}
- public ByteBuffer getCommBuffer() {
+ public ByteBuffer getCommBuffer() throws SocketException {
return getConnection().getCommBuffer();
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/control/ResourceManager.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/control/ResourceManager.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/control/ResourceManager.java
index d7ec83c..8627db9 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/control/ResourceManager.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/control/ResourceManager.java
@@ -16,7 +16,6 @@ import com.gemstone.gemfire.cache.LowMemoryException;
import com.gemstone.gemfire.cache.Region;
import com.gemstone.gemfire.cache.execute.Function;
import com.gemstone.gemfire.cache.query.QueryService;
-import com.gemstone.gemfire.cache.server.CacheServer;
/**
* Provides support for managing resources used by the local
@@ -42,7 +41,7 @@ public interface ResourceManager {
* @see ResourceManager#setCriticalHeapPercentage(float)
* @see ResourceManager#getCriticalHeapPercentage()
*/
- public static final float DEFAULT_CRITICAL_HEAP_PERCENTAGE = 0.0f;
+ public static final float DEFAULT_CRITICAL_PERCENTAGE = 0.0f;
/**
* The default percent of heap memory at which the VM should begin evicting
@@ -55,8 +54,8 @@ public interface ResourceManager {
* @see ResourceManager#setEvictionHeapPercentage(float)
* @see ResourceManager#getEvictionHeapPercentage()
*/
- public static final float DEFAULT_EVICTION_HEAP_PERCENTAGE = 0.0f;
-
+ public static final float DEFAULT_EVICTION_PERCENTAGE = 0.0f;
+
/**
* Creates a factory for defining and starting {@link RebalanceOperation
* RebalanceOperations}.
@@ -125,7 +124,7 @@ public interface ResourceManager {
* @since 6.0
*/
public void setCriticalHeapPercentage(float heapPercentage);
-
+
/**
* Get the percentage of heap at or above which the cache is considered in
* danger of becoming inoperable.
@@ -136,6 +135,48 @@ public interface ResourceManager {
* @since 6.0
*/
public float getCriticalHeapPercentage();
+
+ /**
+ * Set the percentage of off-heap at or above which the cache is considered in
+ * danger of becoming inoperable due to out of memory exceptions.
+ *
+ * <p>
+ * Changing this value can cause {@link LowMemoryException} to be thrown from
+ * the following {@link Cache} operations:
+ * <ul>
+ * <li>{@link Region#put(Object, Object)}
+ * <li>{@link Region#put(Object, Object, Object)}
+ * <li>{@link Region#create(Object, Object)}
+ * <li>{@link Region#create(Object, Object, Object)}
+ * <li>{@link Region#putAll(java.util.Map)}
+ * <li>{@linkplain QueryService#createIndex(String, com.gemstone.gemfire.cache.query.IndexType, String, String) index creation}
+ * <li>Execution of {@link Function}s whose {@link Function#optimizeForWrite()} returns true.
+ * </ul>
+ *
+ * <p>
+ * Only one change to this attribute or the eviction off-heap percentage will be
+ * allowed at any given time and its effect will be fully realized before the
+ * next change is allowed.
+ *
+ * @param offHeapPercentage a percentage of the maximum off-heap memory available
+ * @throws IllegalStateException if the ofHeapPercentage value is not >= 0 or
+ * <= 100 or when less than the current eviction off-heap percentage
+ * @see #getCriticalOffHeapPercentage()
+ * @see #getEvictionOffHeapPercentage()
+ * @since 9.0
+ */
+ public void setCriticalOffHeapPercentage(float offHeapPercentage);
+
+ /**
+ * Get the percentage of off-heap at or above which the cache is considered in
+ * danger of becoming inoperable.
+ *
+ * @return either the current or recently used percentage of the maximum
+ * off-heap memory
+ * @see #setCriticalOffHeapPercentage(float)
+ * @since 9.0
+ */
+ public float getCriticalOffHeapPercentage();
/**
* Set the percentage of heap at or above which the eviction should begin on
@@ -161,7 +202,7 @@ public interface ResourceManager {
* @since 6.0
*/
public void setEvictionHeapPercentage(float heapPercentage);
-
+
/**
* Get the percentage of heap at or above which the eviction should begin on
* Regions configured for {@linkplain
@@ -173,4 +214,38 @@ public interface ResourceManager {
* @since 6.0
*/
public float getEvictionHeapPercentage();
+
+ /**
+ * Set the percentage of off-heap at or above which the eviction should begin on
+ * Regions configured for {@linkplain
+ * EvictionAttributes#createLRUHeapAttributes() HeapLRU eviction}.
+ *
+ * <p>
+ * Changing this value may cause eviction to begin immediately.
+ *
+ * <p>
+ * Only one change to this attribute or critical off-heap percentage will be
+ * allowed at any given time and its effect will be fully realized before the
+ * next change is allowed.
+ *
+ * @param offHeapPercentage a percentage of the maximum off-heap memory available
+ * @throws IllegalStateException if the offHeapPercentage value is not >= 0 or
+ * <= 100 or when greater than the current critical off-heap percentage.
+ * @see #getEvictionOffHeapPercentage()
+ * @see #getCriticalOffHeapPercentage()
+ * @since 9.0
+ */
+ public void setEvictionOffHeapPercentage(float offHeapPercentage);
+
+ /**
+ * Get the percentage of off-heap at or above which the eviction should begin on
+ * Regions configured for {@linkplain
+ * EvictionAttributes#createLRUHeapAttributes() HeapLRU eviction}.
+ *
+ * @return either the current or recently used percentage of the maximum
+ * off-heap memory
+ * @see #setEvictionOffHeapPercentage(float)
+ * @since 9.0
+ */
+ public float getEvictionOffHeapPercentage();
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/1709e627/gemfire-core/src/main/java/com/gemstone/gemfire/cache/hdfs/HDFSEventQueueAttributes.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/hdfs/HDFSEventQueueAttributes.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/hdfs/HDFSEventQueueAttributes.java
new file mode 100644
index 0000000..ef7e863
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/hdfs/HDFSEventQueueAttributes.java
@@ -0,0 +1,72 @@
+/*=========================================================================
+ * Copyright (c) 2002-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *========================================================================
+ */
+
+package com.gemstone.gemfire.cache.hdfs;
+
+/**
+ * {@link HDFSEventQueueAttributes} represents the attributes of the buffer where events are
+ * accumulated before they are persisted to HDFS
+ *
+ * @author Hemant Bhanawat
+ */
+public interface HDFSEventQueueAttributes {
+
+ /**
+ * The Disk store that is required for overflow and persistence
+ * @return String
+ */
+ public String getDiskStoreName();
+
+ /**
+ * The maximum memory after which the data needs to be overflowed to disk
+ * @return int
+ */
+ public int getMaximumQueueMemory();
+
+ /**
+ * Represents the size of a batch per bucket that gets delivered
+ * from the HDFS Event Queue to HDFS. A higher value means that
+ * less number of bigger batches are persisted to HDFS and hence
+ * big files are created on HDFS. But, bigger batches consume memory.
+ *
+ * This value is an indication. Batches per bucket with size less than the specified
+ * are sent to HDFS if interval specified by {@link #getBatchTimeInterval()}
+ * has elapsed.
+ * @return int
+ */
+ public int getBatchSizeMB();
+
+ /**
+ * Represents the batch time interval for a HDFS queue. This is the maximum time interval
+ * that can elapse before a batch of data from a bucket is sent to HDFS.
+ *
+ * @return int
+ */
+ public int getBatchTimeInterval();
+
+ /**
+ * Represents whether the HDFS Event Queue is configured to be persistent or non-persistent
+ * @return boolean
+ */
+ public boolean isPersistent();
+
+ /**
+ * Represents whether or not the writing to the disk is synchronous.
+ *
+ * @return boolean
+ */
+ public boolean isDiskSynchronous();
+
+ /**
+ * Number of threads in VM consuming the events.
+ * default is one.
+ *
+ * @return int
+ */
+ public int getDispatcherThreads();
+}