http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/GridCacheWriteSynchronizationMode.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/GridCacheWriteSynchronizationMode.java b/modules/core/src/main/java/org/apache/ignite/cache/GridCacheWriteSynchronizationMode.java new file mode 100644 index 0000000..389c980 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/GridCacheWriteSynchronizationMode.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache; + +import org.apache.ignite.transactions.*; +import org.jetbrains.annotations.*; + +/** + * Mode indicating how GridGain should wait for write replies from other nodes. Default + * value is {@link #FULL_ASYNC}}, which means that GridGain will not wait for responses from + * participating nodes. This means that by default remote nodes may get their state updated slightly after + * any of the cache write methods complete, or after {@link IgniteTx#commit()} method completes. + * <p> + * Note that regardless of write synchronization mode, cache data will always remain fully + * consistent across all participating nodes. + * <p> + * Write synchronization mode may be configured via {@link org.apache.ignite.cache.CacheConfiguration#getWriteSynchronizationMode()} + * configuration property. + */ +public enum GridCacheWriteSynchronizationMode { + /** + * Flag indicating that GridGain should wait for write or commit replies from all nodes. + * This behavior guarantees that whenever any of the atomic or transactional writes + * complete, all other participating nodes which cache the written data have been updated. + */ + FULL_SYNC, + + /** + * Flag indicating that GridGain will not wait for write or commit responses from participating nodes, + * which means that remote nodes may get their state updated a bit after any of the cache write methods + * complete, or after {@link IgniteTx#commit()} method completes. + */ + FULL_ASYNC, + + /** + * This flag only makes sense for {@link GridCacheMode#PARTITIONED} mode. When enabled, GridGain + * will wait for write or commit to complete on {@code primary} node, but will not wait for + * backups to be updated. + */ + PRIMARY_SYNC; + + /** Enumerated values. */ + private static final GridCacheWriteSynchronizationMode[] VALS = values(); + + /** + * Efficiently gets enumerated value from its ordinal. + * + * @param ord Ordinal value. + * @return Enumerated value or {@code null} if ordinal out of range. + */ + @Nullable public static GridCacheWriteSynchronizationMode fromOrdinal(int ord) { + return ord >= 0 && ord < VALS.length ? VALS[ord] : null; + } +}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinity.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinity.java b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinity.java new file mode 100644 index 0000000..e0d499f --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinity.java @@ -0,0 +1,262 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache.affinity; + +import org.apache.ignite.cache.*; +import org.apache.ignite.cluster.*; +import org.jetbrains.annotations.*; + +import java.util.*; + +/** + * Provides affinity information to detect which node is primary and which nodes are + * backups for a partitioned cache. You can get an instance of this interface by calling + * {@code GridCache.affinity()} method. + * <p> + * Mapping of a key to a node is a three-step operation. First step will get an affinity key for given key + * using {@link GridCacheAffinityKeyMapper}. If mapper is not specified, the original key will be used. Second step + * will map affinity key to partition using {@link GridCacheAffinityFunction#partition(Object)} method. Third step + * will map obtained partition to nodes for current grid topology version. + * <p> + * Interface provides various {@code 'mapKeysToNodes(..)'} methods which provide node affinity mapping for + * given keys. All {@code 'mapKeysToNodes(..)'} methods are not transactional and will not enlist + * keys into ongoing transaction. + */ +public interface GridCacheAffinity<K> { + /** + * Gets number of partitions in cache according to configured affinity function. + * + * @return Number of cache partitions. + * @see GridCacheAffinityFunction + * @see CacheConfiguration#getAffinity() + * @see CacheConfiguration#setAffinity(GridCacheAffinityFunction) + */ + public int partitions(); + + /** + * Gets partition id for the given key. + * + * @param key Key to get partition id for. + * @return Partition id. + * @see GridCacheAffinityFunction + * @see CacheConfiguration#getAffinity() + * @see CacheConfiguration#setAffinity(GridCacheAffinityFunction) + */ + public int partition(K key); + + /** + * Returns {@code true} if given node is the primary node for given key. + * To check if local node is primary for given key, pass + * {@link org.apache.ignite.Ignite#localNode()} as first parameter. + * + * @param n Node to check. + * @param key Key to check. + * @return {@code True} if local node is the primary node for given key. + */ + public boolean isPrimary(ClusterNode n, K key); + + /** + * Returns {@code true} if local node is one of the backup nodes for given key. + * To check if local node is primary for given key, pass {@link org.apache.ignite.Ignite#localNode()} + * as first parameter. + * + * @param n Node to check. + * @param key Key to check. + * @return {@code True} if local node is one of the backup nodes for given key. + */ + public boolean isBackup(ClusterNode n, K key); + + /** + * Returns {@code true} if local node is primary or one of the backup nodes + * for given key. To check if local node is primary or backup for given key, pass + * {@link org.apache.ignite.Ignite#localNode()} as first parameter. + * <p> + * This method is essentially equivalent to calling + * <i>"{@link #isPrimary(org.apache.ignite.cluster.ClusterNode, Object)} || {@link #isBackup(org.apache.ignite.cluster.ClusterNode, Object)})"</i>, + * however it is more efficient as it makes both checks at once. + * + * @param n Node to check. + * @param key Key to check. + * @return {@code True} if local node is primary or backup for given key. + */ + public boolean isPrimaryOrBackup(ClusterNode n, K key); + + /** + * Gets partition ids for which nodes of the given projection has primary + * ownership. + * <p> + * Note that since {@link org.apache.ignite.cluster.ClusterNode} implements {@link org.apache.ignite.cluster.ClusterGroup}, + * to find out primary partitions for a single node just pass + * a single node into this method. + * <p> + * This method may return an empty array if none of nodes in the projection + * have nearOnly disabled. + * + * @param n Grid node. + * @return Partition ids for which given projection has primary ownership. + * @see GridCacheAffinityFunction + * @see CacheConfiguration#getAffinity() + * @see CacheConfiguration#setAffinity(GridCacheAffinityFunction) + */ + public int[] primaryPartitions(ClusterNode n); + + /** + * Gets partition ids for which nodes of the given projection has backup + * ownership. Note that you can find a back up at a certain level, e.g. + * {@code first} backup or {@code third} backup by specifying the + * {@code 'levels} parameter. If no {@code 'level'} is specified then + * all backup partitions are returned. + * <p> + * Note that since {@link org.apache.ignite.cluster.ClusterNode} implements {@link org.apache.ignite.cluster.ClusterGroup}, + * to find out backup partitions for a single node, just pass that single + * node into this method. + * <p> + * This method may return an empty array if none of nodes in the projection + * have nearOnly disabled. + * + * @param n Grid node. + * @return Partition ids for which given projection has backup ownership. + * @see GridCacheAffinityFunction + * @see CacheConfiguration#getAffinity() + * @see CacheConfiguration#setAffinity(GridCacheAffinityFunction) + */ + public int[] backupPartitions(ClusterNode n); + + /** + * Gets partition ids for which nodes of the given projection has ownership + * (either primary or backup). + * <p> + * Note that since {@link org.apache.ignite.cluster.ClusterNode} implements {@link org.apache.ignite.cluster.ClusterGroup}, + * to find out all partitions for a single node, just pass that single + * node into this method. + * <p> + * This method may return an empty array if none of nodes in the projection + * have nearOnly disabled. + * + * @param n Grid node. + * @return Partition ids for which given projection has ownership. + * @see GridCacheAffinityFunction + * @see CacheConfiguration#getAffinity() + * @see CacheConfiguration#setAffinity(GridCacheAffinityFunction) + */ + public int[] allPartitions(ClusterNode n); + + /** + * Maps passed in key to a key which will be used for node affinity. The affinity + * key may be different from actual key if some field in the actual key was + * designated for affinity mapping via {@link GridCacheAffinityKeyMapped} annotation + * or if a custom {@link GridCacheAffinityKeyMapper} was configured. + * + * @param key Key to map. + * @return Key to be used for node-to-affinity mapping (may be the same + * key as passed in). + */ + public Object affinityKey(K key); + + /** + * This method provides ability to detect which keys are mapped to which nodes. + * Use it to determine which nodes are storing which keys prior to sending + * jobs that access these keys. + * <p> + * This method works as following: + * <ul> + * <li>For local caches it returns only local node mapped to all keys.</li> + * <li> + * For fully replicated caches {@link GridCacheAffinityFunction} is + * used to determine which keys are mapped to which nodes. + * </li> + * <li>For partitioned caches, the returned map represents node-to-key affinity.</li> + * </ul> + * + * @param keys Keys to map to nodes. + * @return Map of nodes to keys or empty map if there are no alive nodes for this cache. + */ + public Map<ClusterNode, Collection<K>> mapKeysToNodes(@Nullable Collection<? extends K> keys); + + /** + * This method provides ability to detect to which primary node the given key + * is mapped. Use it to determine which nodes are storing which keys prior to sending + * jobs that access these keys. + * <p> + * This method works as following: + * <ul> + * <li>For local caches it returns only local node ID.</li> + * <li> + * For fully replicated caches first node ID returned by {@link GridCacheAffinityFunction} + * is returned. + * </li> + * <li>For partitioned caches, primary node for the given key is returned.</li> + * </ul> + * + * @param key Keys to map to a node. + * @return Primary node for the key or {@code null} if there are no alive nodes for this cache. + */ + @Nullable public ClusterNode mapKeyToNode(K key); + + /** + * Gets primary and backup nodes for the key. Note that primary node is always + * first in the returned collection. + * <p> + * If there are only cache nodes in the projection with + * {@link CacheConfiguration#getDistributionMode()} property set to {@code NEAR_ONLY}, then this + * method will return an empty collection. + * + * @param key Key to get affinity nodes for. + * @return Collection of primary and backup nodes for the key with primary node + * always first, or an empty collection if this projection contains only nodes with + * {@link CacheConfiguration#getDistributionMode()} property set to {@code NEAR_ONLY}. + */ + public Collection<ClusterNode> mapKeyToPrimaryAndBackups(K key); + + /** + * Gets primary node for the given partition. + * + * @param part Partition id. + * @return Primary node for the given partition. + * @see GridCacheAffinityFunction + * @see CacheConfiguration#getAffinity() + * @see CacheConfiguration#setAffinity(GridCacheAffinityFunction) + */ + public ClusterNode mapPartitionToNode(int part); + + /** + * Gets primary nodes for the given partitions. + * + * @param parts Partition ids. + * @return Mapping of given partitions to their primary nodes. + * @see GridCacheAffinityFunction + * @see CacheConfiguration#getAffinity() + * @see CacheConfiguration#setAffinity(GridCacheAffinityFunction) + */ + public Map<Integer, ClusterNode> mapPartitionsToNodes(Collection<Integer> parts); + + /** + * Gets primary and backup nodes for partition. Note that primary node is always + * first in the returned collection. + * <p> + * If there are only cache nodes in the projection with + * {@link CacheConfiguration#getDistributionMode()} property set to {@code NEAR_ONLY}, then this + * method will return an empty collection. + * + * @param part Partition to get affinity nodes for. + * @return Collection of primary and backup nodes for partition with primary node + * always first, or an empty collection if this projection contains only nodes with + * {@link CacheConfiguration#getDistributionMode()} property set to {@code NEAR_ONLY}. + */ + public Collection<ClusterNode> mapPartitionToPrimaryAndBackups(int part); +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityFunction.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityFunction.java b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityFunction.java new file mode 100644 index 0000000..439c231 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityFunction.java @@ -0,0 +1,112 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache.affinity; + +import org.apache.ignite.cache.*; +import org.apache.ignite.cluster.*; +import org.gridgain.grid.cache.*; + +import java.io.*; +import java.util.*; + +/** + * Cache key affinity which maps keys to nodes. This interface is utilized for + * both, replicated and partitioned caches. Cache affinity can be configured + * for individual caches via {@link CacheConfiguration#getAffinity()} method. + * <p> + * Whenever a key is given to cache, it is first passed to a pluggable + * {@link GridCacheAffinityKeyMapper} which may potentially map this key to an alternate + * key which should be used for affinity. The key returned from + * {@link GridCacheAffinityKeyMapper#affinityKey(Object)} method is then passed to + * {@link #partition(Object) partition(Object)} method to find out the partition for the key. + * On each topology change, partition-to-node mapping is calculated using + * {@link #assignPartitions(GridCacheAffinityFunctionContext)} method, which assigns a collection + * of nodes to each partition. + * This collection of nodes is used for node affinity. In {@link org.apache.ignite.cache.GridCacheMode#REPLICATED REPLICATED} + * cache mode the key will be cached on all returned nodes; generally, all caching nodes + * participate in caching every key in replicated mode. In {@link org.apache.ignite.cache.GridCacheMode#PARTITIONED PARTITIONED} + * mode, only primary and backup nodes are returned with primary node always in the + * first position. So if there is {@code 1} backup node, then the returned collection will + * have {@code 2} nodes in it - {@code primary} node in first position, and {@code backup} + * node in second. + * <p> + * For more information about cache affinity and examples refer to {@link GridCacheAffinityKeyMapper} and + * {@link GridCacheAffinityKeyMapped @GridCacheAffinityKeyMapped} documentation. + * @see GridCacheAffinityKeyMapped + * @see GridCacheAffinityKeyMapper + */ +public interface GridCacheAffinityFunction extends Serializable { + /** + * Resets cache affinity to its initial state. This method will be called by + * the system any time the affinity has been sent to remote node where + * it has to be reinitialized. If your implementation of affinity function + * has no initialization logic, leave this method empty. + */ + public void reset(); + + /** + * Gets total number of partitions available. All caches should always provide + * correct partition count which should be the same on all participating nodes. + * Note that partitions should always be numbered from {@code 0} inclusively to + * {@code N} exclusively without any gaps. + * + * @return Total partition count. + */ + public int partitions(); + + /** + * Gets partition number for a given key starting from {@code 0}. Partitioned caches + * should make sure that keys are about evenly distributed across all partitions + * from {@code 0} to {@link #partitions() partition count} for best performance. + * <p> + * Note that for fully replicated caches it is possible to segment key sets among different + * grid node groups. In that case each node group should return a unique partition + * number. However, unlike partitioned cache, mappings of keys to nodes in + * replicated caches are constant and a node cannot migrate from one partition + * to another. + * + * @param key Key to get partition for. + * @return Partition number for a given key. + */ + public int partition(Object key); + + /** + * Gets affinity nodes for a partition. In case of replicated cache, all returned + * nodes are updated in the same manner. In case of partitioned cache, the returned + * list should contain only the primary and back up nodes with primary node being + * always first. + * <p> + * Note that partitioned affinity must obey the following contract: given that node + * <code>N</code> is primary for some key <code>K</code>, if any other node(s) leave + * grid and no node joins grid, node <code>N</code> will remain primary for key <code>K</code>. + * + * @param affCtx Affinity function context. Will provide all required information to calculate + * new partition assignments. + * @return Unmodifiable list indexed by partition number. Each element of array is a collection in which + * first node is a primary node and other nodes are backup nodes. + */ + public List<List<ClusterNode>> assignPartitions(GridCacheAffinityFunctionContext affCtx); + + /** + * Removes node from affinity. This method is called when it is safe to remove left node from + * affinity mapping. + * + * @param nodeId ID of node to remove. + */ + public void removeNode(UUID nodeId); +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityFunctionContext.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityFunctionContext.java b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityFunctionContext.java new file mode 100644 index 0000000..f742d69 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityFunctionContext.java @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache.affinity; + +import org.apache.ignite.cluster.*; +import org.apache.ignite.events.*; +import org.jetbrains.annotations.*; + +import java.util.*; + +/** + * Affinity function context. This context is passed to {@link GridCacheAffinityFunction} for + * partition reassignment on every topology change event. + */ +public interface GridCacheAffinityFunctionContext { + /** + * Gets affinity assignment for given partition on previous topology version. First node in returned list is + * a primary node, other nodes are backups. + * + * @param part Partition to get previous assignment for. + * @return List of nodes assigned to given partition on previous topology version or {@code null} + * if this information is not available. + */ + @Nullable public List<ClusterNode> previousAssignment(int part); + + /** + * Gets number of backups for new assignment. + * + * @return Number of backups for new assignment. + */ + public int backups(); + + /** + * Gets current topology snapshot. Snapshot will contain only nodes on which particular cache is configured. + * List of passed nodes is guaranteed to be sorted in a same order on all nodes on which partition assignment + * is performed. + * + * @return Cache topology snapshot. + */ + public List<ClusterNode> currentTopologySnapshot(); + + /** + * Gets current topology version number. + * + * @return Current topology version number. + */ + public long currentTopologyVersion(); + + /** + * Gets discovery event caused topology change. + * + * @return Discovery event caused latest topology change or {@code null} if this information is + * not available. + */ + @Nullable public IgniteDiscoveryEvent discoveryEvent(); +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKey.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKey.java b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKey.java new file mode 100644 index 0000000..b251764 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKey.java @@ -0,0 +1,189 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache.affinity; + +import org.gridgain.grid.kernal.processors.cache.*; +import org.gridgain.grid.util.typedef.internal.*; +import org.gridgain.grid.util.tostring.*; + +import java.io.*; + +/** + * Optional wrapper for cache keys to provide support + * for custom affinity mapping. The value returned by + * {@link #affinityKey(Object)} method will be used for key-to-node + * affinity. + * <p> + * Note that the {@link #equals(Object)} and {@link #hashCode()} methods + * delegate directly to the wrapped cache key provided by {@link #key()} + * method. + * <p> + * This class is optional and does not have to be used. It only provides + * extra convenience whenever custom affinity mapping is required. Here is + * an example of how {@code Person} objects can be collocated with + * {@code Company} objects they belong to: + * <pre name="code" class="java"> + * Object personKey = new GridCacheAffinityKey(myPersonId, myCompanyId); + * + * // Both, the company and the person objects will be cached on the same node. + * cache.put(myCompanyId, new Company(..)); + * cache.put(personKey, new Person(..)); + * </pre> + * <p> + * For more information and examples of cache affinity refer to + * {@link GridCacheAffinityKeyMapper} and {@link GridCacheAffinityKeyMapped @GridCacheAffinityKeyMapped} + * documentation. + * @see GridCacheAffinityKeyMapped + * @see GridCacheAffinityKeyMapper + * @see GridCacheAffinityFunction + */ +public class GridCacheAffinityKey<K> implements Externalizable { + /** */ + private static final long serialVersionUID = 0L; + + /** Key. */ + @GridToStringInclude + private K key; + + /** Affinity key. */ + @GridToStringInclude + private Object affKey; + + /** + * Empty constructor. + */ + public GridCacheAffinityKey() { + // No-op. + } + + /** + * Initializes key wrapper for a given key. If affinity key + * is not initialized, then this key will be used for affinity. + * + * @param key Key. + */ + public GridCacheAffinityKey(K key) { + A.notNull(key, "key"); + + this.key = key; + } + + /** + * Initializes key together with its affinity key counter-part. + * + * @param key Key. + * @param affKey Affinity key. + */ + public GridCacheAffinityKey(K key, Object affKey) { + A.notNull(key, "key"); + + this.key = key; + this.affKey = affKey; + } + + /** + * Gets wrapped key. + * + * @return Wrapped key. + */ + public K key() { + return key; + } + + /** + * Sets wrapped key. + * + * @param key Wrapped key. + */ + public void key(K key) { + this.key = key; + } + + /** + * Gets affinity key to use for affinity mapping. If affinity key is not provided, + * then {@code key} value will be returned. + * <p> + * This method is annotated with {@link GridCacheAffinityKeyMapped} and will be picked up + * by {@link GridCacheDefaultAffinityKeyMapper} automatically. + * + * @return Affinity key to use for affinity mapping. + */ + @GridCacheAffinityKeyMapped + @SuppressWarnings({"unchecked"}) + public <T> T affinityKey() { + A.notNull(key, "key"); + + return (T)(affKey == null ? key : affKey); + } + + /** + * Sets affinity key to use for affinity mapping. If affinity key is not provided, + * then {@code key} value will be returned. + * + * @param affKey Affinity key to use for affinity mapping. + */ + public void affinityKey(Object affKey) { + this.affKey = affKey; + } + + /** {@inheritDoc} */ + @Override public void writeExternal(ObjectOutput out) throws IOException { + out.writeObject(key); + out.writeObject(affKey); + } + + /** {@inheritDoc} */ + @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { + key = (K)in.readObject(); + affKey = in.readObject(); + } + + /** + * Hash code implementation which delegates to the underlying {@link #key()}. Note, however, + * that different subclasses of {@code GridCacheAffinityKey} will produce different hash codes. + * <p> + * Users should override this method if different behavior is desired. + * + * @return Hash code. + */ + @Override public int hashCode() { + A.notNull(key, "key"); + + return 31 * key.hashCode() + getClass().getName().hashCode(); + } + + /** + * Equality check which delegates to the underlying key equality. Note, however, that + * different subclasses of {@code GridCacheAffinityKey} will never be equal. + * <p> + * Users should override this method if different behavior is desired. + * + * @param obj Object to check for equality. + * @return {@code True} if objects are equal. + */ + @Override public boolean equals(Object obj) { + A.notNull(key, "key"); + + return obj != null && getClass() == obj.getClass() && key.equals(((GridCacheAffinityKey)obj).key); + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(GridCacheAffinityKey.class, this); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKeyMapped.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKeyMapped.java b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKeyMapped.java new file mode 100644 index 0000000..4b38e01 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKeyMapped.java @@ -0,0 +1,158 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache.affinity; + +import org.gridgain.grid.cache.*; + +import java.lang.annotation.*; +import java.util.concurrent.*; + +/** + * Optional annotation to specify custom key-to-node affinity. Affinity key is a key + * which will be used to determine a node on which given cache key will be stored. This + * annotation allows to mark a field or a method in the cache key object that will be + * used as an affinity key (instead of the entire cache key object that is used for + * affinity by default). Note that a class can have only one field or method annotated + * with {@code @GridCacheAffinityKeyMapped} annotation. + * <p> + * One of the major use cases for this annotation is the routing of grid computations + * to the nodes where the data for this computation is cached, the concept + * otherwise known as {@code Collocation Of Computations And Data}. + * <p> + * <h1 class="header">Mapping Cache Keys</h1> + * The default implementation of {@link GridCacheAffinityKeyMapper}, which will be used + * if no explicit affinity mapper is specified in cache configuration, will first look + * for any field or method annotated with {@code @GridCacheAffinityKeyMapped} annotation. + * If such field or method is not found, then the cache key itself will be used for + * key-to-node affinity (this means that all objects with the same cache key will always + * be routed to the same node). If such field or method is found, then the value of this + * field or method will be used for key-to-node affinity. This allows to specify alternate + * affinity key, other than the cache key itself, whenever needed. + * <p> + * For example, if a {@code Person} object is always accessed together with a {@code Company} object + * for which this person is an employee, then for better performance and scalability it makes sense to + * collocate {@code Person} objects together with their {@code Company} object when storing them in + * cache. To achieve that, cache key used to cache {@code Person} objects should have a field or method + * annotated with {@code @GridCacheAffinityKeyMapped} annotation, which will provide the value of + * the company key for which that person works, like so: + * <pre name="code" class="java"> + * public class PersonKey { + * // Person ID used to identify a person. + * private String personId; + * + * // Company ID which will be used for affinity. + * @GridCacheAffinityKeyMapped + * private String companyId; + * ... + * } + * ... + * // Instantiate person keys. + * Object personKey1 = new PersonKey("myPersonId1", "myCompanyId"); + * Object personKey2 = new PersonKey("myPersonId2", "myCompanyId"); + * + * // Both, the company and the person objects will be cached on the same node. + * cache.put("myCompanyId", new Company(..)); + * cache.put(personKey1, new Person(..)); + * cache.put(personKey2, new Person(..)); + * </pre> + * <p> + * <h2 class="header">GridCacheAffinityKey</h2> + * For convenience, you can also optionally use {@link GridCacheAffinityKey} class. Here is how a + * {@code PersonKey} defined above would look using {@link GridCacheAffinityKey}: + * <pre name="code" class="java"> + * Object personKey1 = new GridCacheAffinityKey("myPersonId1", "myCompanyId"); + * Object personKey2 = new GridCacheAffinityKey("myPersonId2", "myCompanyId"); + * + * // Both, the company and the person objects will be cached on the same node. + * cache.put(myCompanyId, new Company(..)); + * cache.put(personKey1, new Person(..)); + * cache.put(personKey2, new Person(..)); + * </pre> + * <p> + * <h1 class="header">Collocating Computations And Data</h1> + * It is also possible to route computations to the nodes where the data is cached. This concept + * is otherwise known as {@code Collocation Of Computations And Data}. In this case, + * {@code @GridCacheAffinityKeyMapped} annotation allows to specify a routing affinity key for a + * {@link org.apache.ignite.compute.ComputeJob} or any other grid computation, such as {@link Runnable}, {@link Callable}, or + * {@link org.apache.ignite.lang.IgniteClosure}. It should be attached to a method or field that provides affinity key + * for the computation. Only one annotation per class is allowed. Whenever such annotation is detected, + * then {@link org.apache.ignite.spi.loadbalancing.LoadBalancingSpi} will be bypassed, and computation will be routed to the grid node + * where the specified affinity key is cached. You can also use optional {@link org.apache.ignite.cache.GridCacheName @GridCacheName} + * annotation whenever non-default cache name needs to be specified. + * <p> + * Here is how this annotation can be used to route a job to a node where Person object + * is cached with ID "1234": + * <pre name="code" class="java"> + * G.grid().run(new Runnable() { + * // This annotation is optional. If omitted, then default + * // no-name cache will be used. + * @GridCacheName + * private String cacheName = "myCache"; + * + * // This annotation specifies that computation should be routed + * // precisely to the node where key '1234' is cached. + * @GridCacheAffinityKeyMapped + * private String personKey = "1234"; + * + * @Override public void run() { + * // Some computation logic here. + * ... + * } + * }; + * </pre> + * The same can be achieved by annotating method instead of field as follows: + * <pre name="code" class="java"> + * G.grid().run(new Runnable() { + * @Override public void run() { + * // Some computation logic here. + * ... + * } + * + * // This annotation is optional. If omitted, then default + * // no-name cache will be used. + * @GridCacheName + * public String cacheName() { + * return "myCache"; + * } + * + * // This annotation specifies that computation should be routed + * // precisely to the node where key '1234' is cached. + * @GridCacheAffinityKeyMapped + * public String personKey() { + * return "1234"; + * } + * }; + * </pre> + * <p> + * For more information about cache affinity also see {@link GridCacheAffinityKeyMapper} and + * {@link GridCacheAffinityFunction} documentation. + * Affinity for a key can be found from any node, regardless of whether it has cache started + * or not. If cache is not started, affinity function will be fetched from the remote node + * which does have the cache running. + * + * @see org.apache.ignite.cache.GridCacheName + * @see GridCacheAffinityFunction + * @see GridCacheAffinityKeyMapper + * @see GridCacheAffinityKey + */ +@Documented +@Retention(RetentionPolicy.RUNTIME) +@Target({ElementType.FIELD, ElementType.METHOD}) +public @interface GridCacheAffinityKeyMapped { + // No-op. +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKeyMapper.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKeyMapper.java b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKeyMapper.java new file mode 100644 index 0000000..f168e0e --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityKeyMapper.java @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache.affinity; + +import org.apache.ignite.cache.*; + +import java.io.*; + +/** + * Affinity mapper which maps cache key to an affinity key. Affinity key is a key which will be + * used to determine a node on which this key will be cached. Every cache key will first be passed + * through {@link #affinityKey(Object)} method, and the returned value of this method + * will be given to {@link GridCacheAffinityFunction} implementation to find out key-to-node affinity. + * <p> + * The default implementation, which will be used if no explicit affinity mapper is specified + * in cache configuration, will first look for any field or method annotated with + * {@link GridCacheAffinityKeyMapped @GridCacheAffinityKeyMapped} annotation. If such field or method + * is not found, then the cache key itself will be returned from {@link #affinityKey(Object) affinityKey(Object)} + * method (this means that all objects with the same cache key will always be routed to the same node). + * If such field or method is found, then the value of this field or method will be returned from + * {@link #affinityKey(Object) affinityKey(Object)} method. This allows to specify alternate affinity key, other + * than the cache key itself, whenever needed. + * <p> + * A custom (other than default) affinity mapper can be provided + * via {@link CacheConfiguration#getAffinityMapper()} configuration property. + * <p> + * For more information on affinity mapping and examples refer to {@link GridCacheAffinityFunction} and + * {@link GridCacheAffinityKeyMapped @GridCacheAffinityKeyMapped} documentation. + * @see GridCacheAffinityFunction + * @see GridCacheAffinityKeyMapped + */ +public interface GridCacheAffinityKeyMapper extends Serializable { + /** + * Maps passed in key to an alternate key which will be used for node affinity. + * + * @param key Key to map. + * @return Key to be used for node-to-affinity mapping (may be the same + * key as passed in). + */ + public Object affinityKey(Object key); + + /** + * Resets cache affinity mapper to its initial state. This method will be called by + * the system any time the affinity mapper has been sent to remote node where + * it has to be reinitialized. If your implementation of affinity mapper + * has no initialization logic, leave this method empty. + */ + public void reset(); +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeAddressHashResolver.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeAddressHashResolver.java b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeAddressHashResolver.java new file mode 100644 index 0000000..02d34b1 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeAddressHashResolver.java @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache.affinity; + +import org.apache.ignite.cluster.*; +import org.gridgain.grid.util.typedef.internal.*; + +/** + * Node hash resolver which uses {@link org.apache.ignite.cluster.ClusterNode#consistentId()} as alternate hash value. + */ +public class GridCacheAffinityNodeAddressHashResolver implements GridCacheAffinityNodeHashResolver { + /** */ + private static final long serialVersionUID = 0L; + + /** {@inheritDoc} */ + @Override public Object resolve(ClusterNode node) { + return node.consistentId(); + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(GridCacheAffinityNodeAddressHashResolver.class, this); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeHashResolver.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeHashResolver.java b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeHashResolver.java new file mode 100644 index 0000000..4e066c3 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeHashResolver.java @@ -0,0 +1,43 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache.affinity; + +import org.apache.ignite.cluster.*; + +import java.io.*; + +/** + * Resolver which is used to provide node hash value for affinity function. + * <p> + * Node IDs constantly change when nodes get restarted, which causes affinity mapping to change between restarts, + * and hence causing redundant repartitioning. Providing an alternate node hash value, which survives node restarts, + * will help to map keys to the same nodes whenever possible. + * <p> + * Note that on case clients exist they will query this object from the server and use it for affinity calculation. + * Therefore you must ensure that server and clients can marshal and unmarshal this object in portable format, + * i.e. all parties have object class(es) configured as portable. + */ +public interface GridCacheAffinityNodeHashResolver extends Serializable { + /** + * Resolve alternate hash value for the given Grid node. + * + * @param node Grid node. + * @return Resolved hash ID. + */ + public Object resolve(ClusterNode node); +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeIdHashResolver.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeIdHashResolver.java b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeIdHashResolver.java new file mode 100644 index 0000000..18c7510 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheAffinityNodeIdHashResolver.java @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache.affinity; + +import org.apache.ignite.cluster.*; +import org.gridgain.grid.util.typedef.internal.*; + +/** + * Node hash resolver which uses generated node ID as node hash value. As new node ID is generated + * on each node start, this resolver do not provide ability to map keys to the same nodes after restart. + */ +public class GridCacheAffinityNodeIdHashResolver implements GridCacheAffinityNodeHashResolver { + /** */ + private static final long serialVersionUID = 0L; + + /** {@inheritDoc} */ + @Override public Object resolve(ClusterNode node) { + return node.id(); + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(GridCacheAffinityNodeIdHashResolver.class, this); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheCentralizedAffinityFunction.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheCentralizedAffinityFunction.java b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheCentralizedAffinityFunction.java new file mode 100644 index 0000000..a8ffa40 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/GridCacheCentralizedAffinityFunction.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache.affinity; + +import java.lang.annotation.*; + +/** + * Annotation marker which identifies affinity function that must be calculated on one centralized node + * instead of independently on each node. In many cases it happens because it requires previous affinity state + * in order to calculate new one. + */ +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.TYPE) +public @interface GridCacheCentralizedAffinityFunction { + +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/consistenthash/GridCacheConsistentHashAffinityFunction.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/consistenthash/GridCacheConsistentHashAffinityFunction.java b/modules/core/src/main/java/org/apache/ignite/cache/affinity/consistenthash/GridCacheConsistentHashAffinityFunction.java new file mode 100644 index 0000000..628e852 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/consistenthash/GridCacheConsistentHashAffinityFunction.java @@ -0,0 +1,704 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.cache.affinity.consistenthash; + +import org.apache.ignite.*; +import org.apache.ignite.cache.*; +import org.apache.ignite.cache.affinity.*; +import org.apache.ignite.cluster.*; +import org.apache.ignite.lang.*; +import org.apache.ignite.resources.*; +import org.gridgain.grid.*; +import org.gridgain.grid.cache.affinity.*; +import org.gridgain.grid.util.*; +import org.gridgain.grid.util.tostring.*; +import org.gridgain.grid.util.typedef.*; +import org.gridgain.grid.util.typedef.internal.*; +import org.jdk8.backport.*; +import org.jetbrains.annotations.*; + +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +/** + * Affinity function for partitioned cache. This function supports the following + * configuration: + * <ul> + * <li> + * {@code backups} - Use this flag to control how many back up nodes will be + * assigned to every key. The default value is {@code 0}. + * </li> + * <li> + * {@code replicas} - Generally the more replicas a node gets, the more key assignments + * it will receive. You can configure different number of replicas for a node by + * setting user attribute with name {@link #getReplicaCountAttributeName()} to some + * number. Default value is {@code 512} defined by {@link #DFLT_REPLICA_COUNT} constant. + * </li> + * <li> + * {@code backupFilter} - Optional filter for back up nodes. If provided, then only + * nodes that pass this filter will be selected as backup nodes. If not provided, then + * primary and backup nodes will be selected out of all nodes available for this cache. + * </li> + * </ul> + * <p> + * Cache affinity can be configured for individual caches via {@link CacheConfiguration#getAffinity()} method. + */ +public class GridCacheConsistentHashAffinityFunction implements GridCacheAffinityFunction { + /** */ + private static final long serialVersionUID = 0L; + + /** Flag to enable/disable consistency check (for internal use only). */ + private static final boolean AFFINITY_CONSISTENCY_CHECK = Boolean.getBoolean("GRIDGAIN_AFFINITY_CONSISTENCY_CHECK"); + + /** Default number of partitions. */ + public static final int DFLT_PARTITION_COUNT = 10000; + + /** Default replica count for partitioned caches. */ + public static final int DFLT_REPLICA_COUNT = 128; + + /** + * Name of node attribute to specify number of replicas for a node. + * Default value is {@code gg:affinity:node:replicas}. + */ + public static final String DFLT_REPLICA_COUNT_ATTR_NAME = "gg:affinity:node:replicas"; + + /** Node hash. */ + private transient GridConsistentHash<NodeInfo> nodeHash; + + /** Total number of partitions. */ + private int parts = DFLT_PARTITION_COUNT; + + /** */ + private int replicas = DFLT_REPLICA_COUNT; + + /** */ + private String attrName = DFLT_REPLICA_COUNT_ATTR_NAME; + + /** */ + private boolean exclNeighbors; + + /** + * Optional backup filter. First node passed to this filter is primary node, + * and second node is a node being tested. + */ + private IgniteBiPredicate<ClusterNode, ClusterNode> backupFilter; + + /** */ + private GridCacheAffinityNodeHashResolver hashIdRslvr = new GridCacheAffinityNodeAddressHashResolver(); + + /** Injected grid. */ + @IgniteInstanceResource + private Ignite ignite; + + /** Injected cache name. */ + @IgniteCacheNameResource + private String cacheName; + + /** Injected logger. */ + @IgniteLoggerResource + private IgniteLogger log; + + /** Initialization flag. */ + @SuppressWarnings("TransientFieldNotInitialized") + private transient AtomicBoolean init = new AtomicBoolean(); + + /** Latch for initializing. */ + @SuppressWarnings({"TransientFieldNotInitialized"}) + private transient CountDownLatch initLatch = new CountDownLatch(1); + + /** Nodes IDs. */ + @GridToStringInclude + @SuppressWarnings({"TransientFieldNotInitialized"}) + private transient ConcurrentMap<UUID, NodeInfo> addedNodes = new ConcurrentHashMap<>(); + + /** Optional backup filter. */ + @GridToStringExclude + private final IgniteBiPredicate<NodeInfo, NodeInfo> backupIdFilter = new IgniteBiPredicate<NodeInfo, NodeInfo>() { + @Override public boolean apply(NodeInfo primaryNodeInfo, NodeInfo nodeInfo) { + return backupFilter == null || backupFilter.apply(primaryNodeInfo.node(), nodeInfo.node()); + } + }; + + /** Map of neighbors. */ + @SuppressWarnings("TransientFieldNotInitialized") + private transient ConcurrentMap<UUID, Collection<UUID>> neighbors = + new ConcurrentHashMap8<>(); + + /** + * Empty constructor with all defaults. + */ + public GridCacheConsistentHashAffinityFunction() { + // No-op. + } + + /** + * Initializes affinity with flag to exclude same-host-neighbors from being backups of each other + * and specified number of backups. + * <p> + * Note that {@code excludeNeighbors} parameter is ignored if {@code #getBackupFilter()} is set. + * + * @param exclNeighbors {@code True} if nodes residing on the same host may not act as backups + * of each other. + */ + public GridCacheConsistentHashAffinityFunction(boolean exclNeighbors) { + this.exclNeighbors = exclNeighbors; + } + + /** + * Initializes affinity with flag to exclude same-host-neighbors from being backups of each other, + * and specified number of backups and partitions. + * <p> + * Note that {@code excludeNeighbors} parameter is ignored if {@code #getBackupFilter()} is set. + * + * @param exclNeighbors {@code True} if nodes residing on the same host may not act as backups + * of each other. + * @param parts Total number of partitions. + */ + public GridCacheConsistentHashAffinityFunction(boolean exclNeighbors, int parts) { + A.ensure(parts != 0, "parts != 0"); + + this.exclNeighbors = exclNeighbors; + this.parts = parts; + } + + /** + * Initializes optional counts for replicas and backups. + * <p> + * Note that {@code excludeNeighbors} parameter is ignored if {@code backupFilter} is set. + * + * @param parts Total number of partitions. + * @param backupFilter Optional back up filter for nodes. If provided, backups will be selected + * from all nodes that pass this filter. First argument for this filter is primary node, and second + * argument is node being tested. + * <p> + * Note that {@code excludeNeighbors} parameter is ignored if {@code backupFilter} is set. + */ + public GridCacheConsistentHashAffinityFunction(int parts, + @Nullable IgniteBiPredicate<ClusterNode, ClusterNode> backupFilter) { + A.ensure(parts != 0, "parts != 0"); + + this.parts = parts; + this.backupFilter = backupFilter; + } + + /** + * Gets default count of virtual replicas in consistent hash ring. + * <p> + * To determine node replicas, node attribute with {@link #getReplicaCountAttributeName()} + * name will be checked first. If it is absent, then this value will be used. + * + * @return Count of virtual replicas in consistent hash ring. + */ + public int getDefaultReplicas() { + return replicas; + } + + /** + * Sets default count of virtual replicas in consistent hash ring. + * <p> + * To determine node replicas, node attribute with {@link #getReplicaCountAttributeName} name + * will be checked first. If it is absent, then this value will be used. + * + * @param replicas Count of virtual replicas in consistent hash ring.s + */ + public void setDefaultReplicas(int replicas) { + this.replicas = replicas; + } + + /** + * Gets total number of key partitions. To ensure that all partitions are + * equally distributed across all nodes, please make sure that this + * number is significantly larger than a number of nodes. Also, partition + * size should be relatively small. Try to avoid having partitions with more + * than quarter million keys. + * <p> + * Note that for fully replicated caches this method should always + * return {@code 1}. + * + * @return Total partition count. + */ + public int getPartitions() { + return parts; + } + + /** + * Sets total number of partitions. + * + * @param parts Total number of partitions. + */ + public void setPartitions(int parts) { + this.parts = parts; + } + + /** + * Gets hash ID resolver for nodes. This resolver is used to provide + * alternate hash ID, other than node ID. + * <p> + * Node IDs constantly change when nodes get restarted, which causes them to + * be placed on different locations in the hash ring, and hence causing + * repartitioning. Providing an alternate hash ID, which survives node restarts, + * puts node on the same location on the hash ring, hence minimizing required + * repartitioning. + * + * @return Hash ID resolver. + */ + public GridCacheAffinityNodeHashResolver getHashIdResolver() { + return hashIdRslvr; + } + + /** + * Sets hash ID resolver for nodes. This resolver is used to provide + * alternate hash ID, other than node ID. + * <p> + * Node IDs constantly change when nodes get restarted, which causes them to + * be placed on different locations in the hash ring, and hence causing + * repartitioning. Providing an alternate hash ID, which survives node restarts, + * puts node on the same location on the hash ring, hence minimizing required + * repartitioning. + * + * @param hashIdRslvr Hash ID resolver. + */ + public void setHashIdResolver(GridCacheAffinityNodeHashResolver hashIdRslvr) { + this.hashIdRslvr = hashIdRslvr; + } + + /** + * Gets optional backup filter. If not {@code null}, backups will be selected + * from all nodes that pass this filter. First node passed to this filter is primary node, + * and second node is a node being tested. + * <p> + * Note that {@code excludeNeighbors} parameter is ignored if {@code backupFilter} is set. + * + * @return Optional backup filter. + */ + @Nullable public IgniteBiPredicate<ClusterNode, ClusterNode> getBackupFilter() { + return backupFilter; + } + + /** + * Sets optional backup filter. If provided, then backups will be selected from all + * nodes that pass this filter. First node being passed to this filter is primary node, + * and second node is a node being tested. + * <p> + * Note that {@code excludeNeighbors} parameter is ignored if {@code backupFilter} is set. + * + * @param backupFilter Optional backup filter. + */ + public void setBackupFilter(@Nullable IgniteBiPredicate<ClusterNode, ClusterNode> backupFilter) { + this.backupFilter = backupFilter; + } + + /** + * Gets optional attribute name for replica count. If not provided, the + * default is {@link #DFLT_REPLICA_COUNT_ATTR_NAME}. + * + * @return User attribute name for replica count for a node. + */ + public String getReplicaCountAttributeName() { + return attrName; + } + + /** + * Sets optional attribute name for replica count. If not provided, the + * default is {@link #DFLT_REPLICA_COUNT_ATTR_NAME}. + * + * @param attrName User attribute name for replica count for a node. + */ + public void setReplicaCountAttributeName(String attrName) { + this.attrName = attrName; + } + + /** + * Checks flag to exclude same-host-neighbors from being backups of each other (default is {@code false}). + * <p> + * Note that {@code excludeNeighbors} parameter is ignored if {@code #getBackupFilter()} is set. + * + * @return {@code True} if nodes residing on the same host may not act as backups of each other. + */ + public boolean isExcludeNeighbors() { + return exclNeighbors; + } + + /** + * Sets flag to exclude same-host-neighbors from being backups of each other (default is {@code false}). + * <p> + * Note that {@code excludeNeighbors} parameter is ignored if {@code #getBackupFilter()} is set. + * + * @param exclNeighbors {@code True} if nodes residing on the same host may not act as backups of each other. + */ + public void setExcludeNeighbors(boolean exclNeighbors) { + this.exclNeighbors = exclNeighbors; + } + + /** + * Gets neighbors for a node. + * + * @param node Node. + * @return Neighbors. + */ + private Collection<UUID> neighbors(final ClusterNode node) { + Collection<UUID> ns = neighbors.get(node.id()); + + if (ns == null) { + Collection<ClusterNode> nodes = ignite.cluster().forHost(node).nodes(); + + ns = F.addIfAbsent(neighbors, node.id(), new ArrayList<>(F.nodeIds(nodes))); + } + + return ns; + } + + /** {@inheritDoc} */ + @SuppressWarnings("unchecked") + @Override public List<List<ClusterNode>> assignPartitions(GridCacheAffinityFunctionContext ctx) { + List<List<ClusterNode>> res = new ArrayList<>(parts); + + Collection<ClusterNode> topSnapshot = ctx.currentTopologySnapshot(); + + for (int part = 0; part < parts; part++) { + res.add(F.isEmpty(topSnapshot) ? + Collections.<ClusterNode>emptyList() : + // Wrap affinity nodes with unmodifiable list since unmodifiable generic collection + // doesn't provide equals and hashCode implementations. + U.sealList(nodes(part, topSnapshot, ctx.backups()))); + } + + return res; + } + + /** + * Assigns nodes to one partition. + * + * @param part Partition to assign nodes for. + * @param nodes Cache topology nodes. + * @return Assigned nodes, first node is primary, others are backups. + */ + public Collection<ClusterNode> nodes(int part, Collection<ClusterNode> nodes, int backups) { + if (nodes == null) + return Collections.emptyList(); + + int nodesSize = nodes.size(); + + if (nodesSize == 0) + return Collections.emptyList(); + + if (nodesSize == 1) // Minor optimization. + return nodes; + + initialize(); + + final Map<NodeInfo, ClusterNode> lookup = new GridLeanMap<>(nodesSize); + + // Store nodes in map for fast lookup. + for (ClusterNode n : nodes) + // Add nodes into hash circle, if absent. + lookup.put(resolveNodeInfo(n), n); + + Collection<NodeInfo> selected; + + if (backupFilter != null) { + final IgnitePredicate<NodeInfo> p = new P1<NodeInfo>() { + @Override public boolean apply(NodeInfo id) { + return lookup.containsKey(id); + } + }; + + final NodeInfo primaryId = nodeHash.node(part, p); + + IgnitePredicate<NodeInfo> backupPrimaryIdFilter = new IgnitePredicate<NodeInfo>() { + @Override public boolean apply(NodeInfo node) { + return backupIdFilter.apply(primaryId, node); + } + }; + + Collection<NodeInfo> backupIds = nodeHash.nodes(part, backups, p, backupPrimaryIdFilter); + + if (F.isEmpty(backupIds) && primaryId != null) { + ClusterNode n = lookup.get(primaryId); + + assert n != null; + + return Collections.singletonList(n); + } + + selected = primaryId != null ? F.concat(false, primaryId, backupIds) : backupIds; + } + else { + if (!exclNeighbors) { + selected = nodeHash.nodes(part, backups == Integer.MAX_VALUE ? backups : backups + 1, new P1<NodeInfo>() { + @Override public boolean apply(NodeInfo id) { + return lookup.containsKey(id); + } + }); + + if (selected.size() == 1) { + NodeInfo id = F.first(selected); + + assert id != null : "Node ID cannot be null in affinity node ID collection: " + selected; + + ClusterNode n = lookup.get(id); + + assert n != null; + + return Collections.singletonList(n); + } + } + else { + int primaryAndBackups = backups + 1; + + selected = new ArrayList<>(primaryAndBackups); + + final Collection<NodeInfo> selected0 = selected; + + List<NodeInfo> ids = nodeHash.nodes(part, primaryAndBackups, new P1<NodeInfo>() { + @Override public boolean apply(NodeInfo id) { + ClusterNode n = lookup.get(id); + + if (n == null) + return false; + + Collection<UUID> neighbors = neighbors(n); + + for (NodeInfo id0 : selected0) { + ClusterNode n0 = lookup.get(id0); + + if (n0 == null) + return false; + + Collection<UUID> neighbors0 = neighbors(n0); + + if (F.containsAny(neighbors0, neighbors)) + return false; + } + + selected0.add(id); + + return true; + } + }); + + if (AFFINITY_CONSISTENCY_CHECK) + assert F.eqOrdered(ids, selected); + } + } + + Collection<ClusterNode> ret = new ArrayList<>(selected.size()); + + for (NodeInfo id : selected) { + ClusterNode n = lookup.get(id); + + assert n != null; + + ret.add(n); + } + + return ret; + } + + /** {@inheritDoc} */ + @Override public int partition(Object key) { + initialize(); + + return U.safeAbs(key.hashCode() % parts); + } + + /** {@inheritDoc} */ + @Override public int partitions() { + initialize(); + + return parts; + } + + /** {@inheritDoc} */ + @Override public void reset() { + addedNodes = new ConcurrentHashMap<>(); + neighbors = new ConcurrentHashMap8<>(); + + initLatch = new CountDownLatch(1); + + init = new AtomicBoolean(); + } + + /** {@inheritDoc} */ + @Override public void removeNode(UUID nodeId) { + NodeInfo info = addedNodes.remove(nodeId); + + if (info == null) + return; + + nodeHash.removeNode(info); + + neighbors.clear(); + } + + /** + * Resolve node info for specified node. + * Add node to hash circle if this is the first node invocation. + * + * @param n Node to get info for. + * @return Node info. + */ + private NodeInfo resolveNodeInfo(ClusterNode n) { + UUID nodeId = n.id(); + NodeInfo nodeInfo = addedNodes.get(nodeId); + + if (nodeInfo != null) + return nodeInfo; + + assert hashIdRslvr != null; + + nodeInfo = new NodeInfo(nodeId, hashIdRslvr.resolve(n), n); + + neighbors.clear(); + + nodeHash.addNode(nodeInfo, replicas(n)); + + addedNodes.put(nodeId, nodeInfo); + + return nodeInfo; + } + + /** {@inheritDoc} */ + private void initialize() { + if (!init.get() && init.compareAndSet(false, true)) { + if (log.isInfoEnabled()) + log.info("Consistent hash configuration [cacheName=" + cacheName + ", partitions=" + parts + + ", excludeNeighbors=" + exclNeighbors + ", replicas=" + replicas + + ", backupFilter=" + backupFilter + ", hashIdRslvr=" + hashIdRslvr + ']'); + + nodeHash = new GridConsistentHash<>(); + + initLatch.countDown(); + } + else { + if (initLatch.getCount() > 0) { + try { + U.await(initLatch); + } + catch (GridInterruptedException ignored) { + // Recover interrupted state flag. + Thread.currentThread().interrupt(); + } + } + } + } + + /** + * @param n Node. + * @return Replicas. + */ + private int replicas(ClusterNode n) { + Integer nodeReplicas = n.attribute(attrName); + + if (nodeReplicas == null) + nodeReplicas = replicas; + + return nodeReplicas; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(GridCacheConsistentHashAffinityFunction.class, this); + } + + /** + * Node hash ID. + */ + private static final class NodeInfo implements Comparable<NodeInfo> { + /** Node ID. */ + private UUID nodeId; + + /** Hash ID. */ + private Object hashId; + + /** Grid node. */ + private ClusterNode node; + + /** + * @param nodeId Node ID. + * @param hashId Hash ID. + * @param node Rich node. + */ + private NodeInfo(UUID nodeId, Object hashId, ClusterNode node) { + assert nodeId != null; + assert hashId != null; + + this.hashId = hashId; + this.nodeId = nodeId; + this.node = node; + } + + /** + * @return Node ID. + */ + public UUID nodeId() { + return nodeId; + } + + /** + * @return Hash ID. + */ + public Object hashId() { + return hashId; + } + + /** + * @return Node. + */ + public ClusterNode node() { + return node; + } + + /** {@inheritDoc} */ + @Override public int hashCode() { + return hashId.hashCode(); + } + + /** {@inheritDoc} */ + @Override public boolean equals(Object obj) { + if (!(obj instanceof NodeInfo)) + return false; + + NodeInfo that = (NodeInfo)obj; + + // If objects are equal, hash codes should be the same. + // Cannot use that.hashId.equals(hashId) due to Comparable<N> interface restrictions. + return that.nodeId.equals(nodeId) && that.hashCode() == hashCode(); + } + + /** {@inheritDoc} */ + @Override public int compareTo(NodeInfo o) { + int diff = nodeId.compareTo(o.nodeId); + + if (diff == 0) { + int h1 = hashCode(); + int h2 = o.hashCode(); + + diff = h1 == h2 ? 0 : (h1 < h2 ? -1 : 1); + } + + return diff; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(NodeInfo.class, this); + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b1a738d/modules/core/src/main/java/org/apache/ignite/cache/affinity/consistenthash/package.html ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/affinity/consistenthash/package.html b/modules/core/src/main/java/org/apache/ignite/cache/affinity/consistenthash/package.html new file mode 100644 index 0000000..f5d5e93 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/cache/affinity/consistenthash/package.html @@ -0,0 +1,24 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + --> + +<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd";> +<html> +<body> + <!-- Package description. --> + Contains consistent hash based cache affinity for partitioned cache. +</body> +</html>
