[infinispan-commits] Infinispan SVN: r2621 - in branches/4.2.x: core/src/main/java/org/infinispan/commands and 11 other directories.
infinispan-commits at lists.jboss.org
infinispan-commits at lists.jboss.org
Wed Oct 27 16:44:12 EDT 2010
Author: mircea.markus
Date: 2010-10-27 16:44:10 -0400 (Wed, 27 Oct 2010)
New Revision: 2621
Added:
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/AbstractConsistentHash.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/AbstractWheelConsistentHash.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ConsistentHash.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ConsistentHashHelper.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/DefaultConsistentHash.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ExperimentalDefaultConsistentHash.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/NodeTopologyInfo.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/TopologyAwareConsistentHash.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/TopologyInfo.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/UnionConsistentHash.java
branches/4.2.x/core/src/test/java/org/infinispan/distribution/DistributionManagerImplTest.java
branches/4.2.x/core/src/test/java/org/infinispan/distribution/TopologyAwareConsistentHashTest.java
Removed:
branches/4.2.x/core/src/main/java/org/infinispan/distribution/AbstractConsistentHash.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ConsistentHash.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ConsistentHashHelper.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/DefaultConsistentHash.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/ExperimentalDefaultConsistentHash.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/UnionConsistentHash.java
Modified:
branches/4.2.x/core/src/main/java/org/infinispan/affinity/KeyAffinityServiceFactory.java
branches/4.2.x/core/src/main/java/org/infinispan/affinity/KeyAffinityServiceImpl.java
branches/4.2.x/core/src/main/java/org/infinispan/commands/CommandsFactory.java
branches/4.2.x/core/src/main/java/org/infinispan/commands/CommandsFactoryImpl.java
branches/4.2.x/core/src/main/java/org/infinispan/commands/control/RehashControlCommand.java
branches/4.2.x/core/src/main/java/org/infinispan/commands/tx/CommitCommand.java
branches/4.2.x/core/src/main/java/org/infinispan/config/Configuration.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/DistributionManager.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/DistributionManagerImpl.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/InvertedLeaveTask.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/JoinTask.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/RehashTask.java
branches/4.2.x/core/src/main/java/org/infinispan/distribution/TransactionLoggerImpl.java
branches/4.2.x/core/src/main/java/org/infinispan/marshall/jboss/ConstantObjectTable.java
branches/4.2.x/core/src/test/java/org/infinispan/affinity/BaseKeyAffinityServiceTest.java
branches/4.2.x/core/src/test/java/org/infinispan/affinity/KeyAffinityServiceTest.java
branches/4.2.x/core/src/test/java/org/infinispan/config/parsing/XmlFileParsingTest.java
branches/4.2.x/core/src/test/java/org/infinispan/distribution/BaseDistFunctionalTest.java
branches/4.2.x/core/src/test/java/org/infinispan/distribution/ConsistentHashPerfTest.java
branches/4.2.x/core/src/test/java/org/infinispan/distribution/DefaultConsistentHashTest.java
branches/4.2.x/core/src/test/java/org/infinispan/distribution/rehash/RehashTestBase.java
branches/4.2.x/core/src/test/java/org/infinispan/distribution/rehash/WorkDuringJoinTest.java
branches/4.2.x/server/hotrod/src/test/scala/org/infinispan/server/hotrod/HotRodDistributionTest.scala
Log:
[ISPN-180]-(Colocated nodes should be handled in DIST) - ongoing work
Modified: branches/4.2.x/core/src/main/java/org/infinispan/affinity/KeyAffinityServiceFactory.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/affinity/KeyAffinityServiceFactory.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/affinity/KeyAffinityServiceFactory.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -12,7 +12,7 @@
* Services build by this factory have the following characteristics:
* <ul>
* <li>are run asynchronously by a thread that can be plugged through an {@link org.infinispan.executors.ExecutorFactory} </li>
- * <li>for key generation, the {@link org.infinispan.distribution.ConsistentHash} function of a distributed cache is used. Service does not make sense for replicated caches.</li>
+ * <li>for key generation, the {@link org.infinispan.distribution.ch.ConsistentHash} function of a distributed cache is used. Service does not make sense for replicated caches.</li>
* <li>for each address cluster member (identified by an {@link org.infinispan.remoting.transport.Address} member, a fixed number of keys is generated</li>
* </ul>
*
Modified: branches/4.2.x/core/src/main/java/org/infinispan/affinity/KeyAffinityServiceImpl.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/affinity/KeyAffinityServiceImpl.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/affinity/KeyAffinityServiceImpl.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -3,7 +3,7 @@
import net.jcip.annotations.GuardedBy;
import net.jcip.annotations.ThreadSafe;
import org.infinispan.Cache;
-import org.infinispan.distribution.ConsistentHash;
+import org.infinispan.distribution.ch.ConsistentHash;
import org.infinispan.distribution.DistributionManager;
import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.notifications.cachemanagerlistener.event.CacheStoppedEvent;
Modified: branches/4.2.x/core/src/main/java/org/infinispan/commands/CommandsFactory.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/commands/CommandsFactory.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/commands/CommandsFactory.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -44,7 +44,7 @@
import org.infinispan.commands.write.ReplaceCommand;
import org.infinispan.commands.write.WriteCommand;
import org.infinispan.container.entries.InternalCacheValue;
-import org.infinispan.distribution.ConsistentHash;
+import org.infinispan.distribution.ch.ConsistentHash;
import org.infinispan.factories.scopes.Scope;
import org.infinispan.factories.scopes.Scopes;
import org.infinispan.remoting.transport.Address;
@@ -53,7 +53,6 @@
import java.util.Collection;
import java.util.List;
import java.util.Map;
-import java.util.Set;
/**
* A factory to build commands, initializing and injecting dependencies accordingly. Commands built for a specific,
Modified: branches/4.2.x/core/src/main/java/org/infinispan/commands/CommandsFactoryImpl.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/commands/CommandsFactoryImpl.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/commands/CommandsFactoryImpl.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -52,7 +52,7 @@
import org.infinispan.container.DataContainer;
import org.infinispan.container.entries.InternalCacheValue;
import org.infinispan.context.InvocationContextContainer;
-import org.infinispan.distribution.ConsistentHash;
+import org.infinispan.distribution.ch.ConsistentHash;
import org.infinispan.distribution.DistributionManager;
import org.infinispan.factories.annotations.Inject;
import org.infinispan.factories.annotations.Start;
Modified: branches/4.2.x/core/src/main/java/org/infinispan/commands/control/RehashControlCommand.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/commands/control/RehashControlCommand.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/commands/control/RehashControlCommand.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -11,7 +11,7 @@
import org.infinispan.container.entries.InternalCacheEntry;
import org.infinispan.container.entries.InternalCacheValue;
import org.infinispan.context.InvocationContext;
-import org.infinispan.distribution.ConsistentHash;
+import org.infinispan.distribution.ch.ConsistentHash;
import org.infinispan.distribution.DistributionManager;
import org.infinispan.loaders.CacheLoaderException;
import org.infinispan.loaders.CacheStore;
Modified: branches/4.2.x/core/src/main/java/org/infinispan/commands/tx/CommitCommand.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/commands/tx/CommitCommand.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/commands/tx/CommitCommand.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -53,4 +53,9 @@
public byte getCommandId() {
return COMMAND_ID;
}
+
+ @Override
+ public String toString() {
+ return "CommitCommand{" + super.toString();
+ }
}
Modified: branches/4.2.x/core/src/main/java/org/infinispan/config/Configuration.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/config/Configuration.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/config/Configuration.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -21,7 +21,7 @@
*/
package org.infinispan.config;
-import org.infinispan.distribution.DefaultConsistentHash;
+import org.infinispan.distribution.ch.DefaultConsistentHash;
import org.infinispan.eviction.EvictionStrategy;
import org.infinispan.eviction.EvictionThreadPolicy;
import org.infinispan.factories.ComponentRegistry;
Deleted: branches/4.2.x/core/src/main/java/org/infinispan/distribution/AbstractConsistentHash.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/AbstractConsistentHash.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/AbstractConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -1,32 +0,0 @@
-package org.infinispan.distribution;
-
-import org.infinispan.remoting.transport.Address;
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * An abstract consistent hash implementation that handles common implementations of certain methods. In particular,
- * default implementations of {@link #locateAll(java.util.Collection, int)} and {@link #isKeyLocalToAddress(org.infinispan.remoting.transport.Address, Object, int)}.
- * <p />
- * The versions provided here are relatively inefficient in that they call {@link #locate(Object, int)} first (and
- * sometimes in a loop). Depending on the algorithm used, there may be more efficient ways to achieve the same results
- * and in such cases the methods provided here should be overridden.
- * <p />
- * @author Manik Surtani
- * @since 4.0
- */
-public abstract class AbstractConsistentHash implements ConsistentHash {
- public Map<Object, List<Address>> locateAll(Collection<Object> keys, int replCount) {
- Map<Object, List<Address>> locations = new HashMap<Object, List<Address>>();
- for (Object k : keys) locations.put(k, locate(k, replCount));
- return locations;
- }
-
- public boolean isKeyLocalToAddress(Address a, Object key, int replCount) {
- // simple, brute-force impl
- return locate(key, replCount).contains(a);
- }
-}
Deleted: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ConsistentHash.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ConsistentHash.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -1,100 +0,0 @@
-package org.infinispan.distribution;
-
-import org.infinispan.remoting.transport.Address;
-
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-
-/**
- * A consistent hash algorithm implementation. Implementations would typically be constructed via reflection so should
- * implement a public, no-arg constructor.
- *
- * @author Manik Surtani
- * @since 4.0
- */
-public interface ConsistentHash {
-
- /**
- * Sets the collection of cache addresses in the cluster. The implementation should store these internally and use
- * these to locate keys.
- *
- * @param caches caches in cluster.
- */
- void setCaches(List<Address> caches);
-
- /**
- * Should return a collection of cache addresses in the cluster.
- *
- * @return collection of cache addresses
- */
- List<Address> getCaches();
-
- /**
- * Locates a key, given a replication count (number of copies).
- *
- * @param key key to locate
- * @param replCount replication count (number of copies)
- * @return a list of addresses where the key resides, where this list is a subset of the addresses set in {@link
- * #setCaches(java.util.List)}. Should never be null, and should contain replCount elements or the max
- * number of caches available, whichever is smaller.
- */
- List<Address> locate(Object key, int replCount);
-
- /**
- * The logical equivalent of calling {@link #locate(Object, int)} multiple times for each key in the collection of
- * keys. Implementations may be optimised for such a bulk lookup, or may just repeatedly call {@link #locate(Object,
- * int)}.
- *
- * @param keys keys to locate
- * @param replCount replication count (number of copies) for each key
- * @return Map of locations, keyed on key.
- */
- Map<Object, List<Address>> locateAll(Collection<Object> keys, int replCount);
-
- /**
- * Calculates the logical distance between two addresses. This distance is based on where the addresses lie in the
- * hash space.
- *
- * @param a1 address to test
- * @param a2 address to test
- * @return the distance between the 2 nodes. Always a positive number, where the distance between a1 and itself is
- * 0. The distance between a1 and the next adjacent node is 1 and teh distance between a1 and the previous
- * adjacent node is caches.size() - 1. A -1 may be returned if either of the addresses do not exist.
- */
- int getDistance(Address a1, Address a2);
-
- /**
- * Tests whether two addresses are logically next to each other in the hash space.
- *
- * @param a1 address to test
- * @param a2 address to test
- * @return true if adjacent, false if not
- */
- boolean isAdjacent(Address a1, Address a2);
-
- /**
- * Test to see whether a key is mapped to a given address.
- * @param a address to test
- * @param key key to test
- * @param replCount repl count
- * @return true if the key is mapped to the address; false otherwise
- */
- boolean isKeyLocalToAddress(Address a, Object key, int replCount);
-
- /**
- * Returns the value between 0 and the hash space limit, or hash id, for a particular address. If there's no such
- * value for an address, this method will return -1.
- *
- * @return An int between 0 and hash space if the address is present in the hash wheel, otherwise it returns -1.
- */
- int getHashId(Address a);
-
- /**
- * Returns the hash space constant for this consistent hash algorithm class. This integer is often used as modulus
- * for arithmetic operations within the algorithm, for example, limiting the range of possible hash values.
- *
- * @return A positive integer containing the hash space constant or 0 is not supported by implementation.
- */
- int getHashSpace();
-}
Deleted: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ConsistentHashHelper.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ConsistentHashHelper.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ConsistentHashHelper.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -1,144 +0,0 @@
-package org.infinispan.distribution;
-
-import org.infinispan.config.Configuration;
-import org.infinispan.remoting.transport.Address;
-import org.infinispan.util.Util;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.LinkedList;
-import java.util.List;
-
-/**
- * A helper class that handles the construction of consistent hash instances based on configuration.
- *
- * @author Manik Surtani
- * @since 4.0
- */
-public class ConsistentHashHelper {
-
- /**
- * Returns a new consistent hash of the same type with the given address removed.
- *
- * @param ch consistent hash to start with
- * @param toRemove address to remove
- * @param c configuration
- * @return a new consistent hash instance of the same type
- */
- public static ConsistentHash removeAddress(ConsistentHash ch, Address toRemove, Configuration c) {
- if (ch instanceof UnionConsistentHash)
- return removeAddressFromUnionConsistentHash((UnionConsistentHash) ch, toRemove, c);
- else {
- ConsistentHash newCH = (ConsistentHash) Util.getInstance(c.getConsistentHashClass());
- List<Address> caches = new ArrayList<Address>(ch.getCaches());
- caches.remove(toRemove);
- newCH.setCaches(caches);
- return newCH;
- }
- }
-
- /**
- * Creates a new UnionConsistentHash instance based on the old instance, removing the provided address from both
- * target consistent hash instances in the union.
- *
- * @param uch union consistent hash instance
- * @param toRemove address to remove
- * @param c configuration
- * @return a new UnionConsistentHash instance
- */
- public static UnionConsistentHash removeAddressFromUnionConsistentHash(UnionConsistentHash uch, Address toRemove, Configuration c) {
- ConsistentHash newFirstCH = removeAddress(uch.getOldConsistentHash(), toRemove, c);
- ConsistentHash newSecondCH = removeAddress(uch.getNewConsistentHash(), toRemove, c);
- return new UnionConsistentHash(newFirstCH, newSecondCH);
- }
-
- /**
- * Creates a new consistent hash instance based on the user's configuration, and populates the consistent hash
- * with the collection of addresses passed in.
- *
- * @param c configuration
- * @param addresses with which to populate the consistent hash
- * @return a new consistent hash instance
- */
- public static ConsistentHash createConsistentHash(Configuration c, List<Address> addresses) {
- ConsistentHash ch = (ConsistentHash) Util.getInstance(c.getConsistentHashClass());
- ch.setCaches(addresses);
- return ch;
- }
-
- /**
- * Creates a new consistent hash instance based on the user's configuration, and populates the consistent hash
- * with the collection of addresses passed in.
- *
- * @param c configuration
- * @param addresses with which to populate the consistent hash
- * @param moreAddresses to add to the list of addresses
- * @return a new consistent hash instance
- */
- public static ConsistentHash createConsistentHash(Configuration c, List<Address> addresses, Address... moreAddresses) {
- List<Address> list = new LinkedList<Address>(addresses);
- list.addAll(Arrays.asList(moreAddresses));
- return createConsistentHash(c, list);
- }
-
- /**
- * Creates a new consistent hash instance based on the user's configuration, and populates the consistent hash
- * with the collection of addresses passed in.
- *
- * @param c configuration
- * @param addresses with which to populate the consistent hash
- * @param moreAddresses to add to the list of addresses
- * @return a new consistent hash instance
- */
- public static ConsistentHash createConsistentHash(Configuration c, List<Address> addresses, Collection<Address> moreAddresses) {
- List<Address> list = new LinkedList<Address>(addresses);
- list.addAll(moreAddresses);
- return createConsistentHash(c, list);
- }
-
- /**
- * Creates a new consistent hash instance based on the type specified, and populates the consistent hash
- * with the collection of addresses passed in.
- *
- * @param clazz type of the consistent hash to create
- * @param addresses with which to populate the consistent hash
- * @return a new consistent hash instance
- */
- public static ConsistentHash createConsistentHash(Class<? extends ConsistentHash> clazz, List<Address> addresses) {
- ConsistentHash ch;
- ch = Util.getInstance(clazz);
- if (addresses != null && !addresses.isEmpty()) ch.setCaches(addresses);
- return ch;
- }
-
- /**
- * Creates a new consistent hash instance based on the type specified, and populates the consistent hash
- * with the collection of addresses passed in.
- *
- * @param clazz type of the consistent hash to create
- * @param addresses with which to populate the consistent hash
- * @param moreAddresses to add to the list of addresses
- * @return a new consistent hash instance
- */
- public static ConsistentHash createConsistentHash(Class<? extends ConsistentHash> clazz, List<Address> addresses, Address... moreAddresses) {
- List<Address> list = new LinkedList<Address>(addresses);
- list.addAll(Arrays.asList(moreAddresses));
- return createConsistentHash(clazz, list);
- }
-
- /**
- * Creates a new consistent hash instance based on the type specified, and populates the consistent hash
- * with the collection of addresses passed in.
- *
- * @param clazz type of the consistent hash to create
- * @param addresses with which to populate the consistent hash
- * @param moreAddresses to add to the list of addresses
- * @return a new consistent hash instance
- */
- public static ConsistentHash createConsistentHash(Class<? extends ConsistentHash> clazz, List<Address> addresses, Collection<Address> moreAddresses) {
- List<Address> list = new LinkedList<Address>(addresses);
- list.addAll(moreAddresses);
- return createConsistentHash(clazz, list);
- }
-}
Deleted: branches/4.2.x/core/src/main/java/org/infinispan/distribution/DefaultConsistentHash.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/DefaultConsistentHash.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/DefaultConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -1,213 +0,0 @@
-package org.infinispan.distribution;
-
-import org.infinispan.marshall.Ids;
-import org.infinispan.marshall.Marshallable;
-import org.infinispan.remoting.transport.Address;
-
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.SortedMap;
-import java.util.TreeMap;
-
-import static java.lang.Math.min;
-import static org.infinispan.util.hash.MurmurHash2.hash;
-
- at Marshallable(externalizer = DefaultConsistentHash.Externalizer.class, id = Ids.DEFAULT_CONSISTENT_HASH)
-public class DefaultConsistentHash extends AbstractConsistentHash {
-
- // make sure all threads see the current list
- ArrayList<Address> addresses;
- SortedMap<Integer, Address> positions;
- // TODO: Maybe address and addressToHashIds can be combined in a LinkedHashMap?
- Map<Address, Integer> addressToHashIds;
-
- final static int HASH_SPACE = 10240; // no more than 10k nodes?
-
- public void setCaches(List<Address> caches) {
-
- addresses = new ArrayList<Address>(caches);
-
- // this list won't grow.
- addresses.trimToSize();
-
- positions = new TreeMap<Integer, Address>();
- addressToHashIds = new HashMap<Address, Integer>();
-
- for (Address a : addresses) {
- int positionIndex = Math.abs(hash(a)) % HASH_SPACE;
- // this is deterministic since the address list is ordered and the order is consistent across the grid
- while (positions.containsKey(positionIndex)) positionIndex = positionIndex + 1 % HASH_SPACE;
- positions.put(positionIndex, a);
- // If address appears several times, take the lowest value to guarantee that
- // at least the initial value and subsequent +1 values would end up in the same node
- // TODO: Remove this check since https://jira.jboss.org/jira/browse/ISPN-428 contains a proper fix for this
- if (!addressToHashIds.containsKey(a))
- addressToHashIds.put(a, positionIndex);
- }
-
- addresses.clear();
- // reorder addresses as per the positions.
- for (Address a : positions.values()) addresses.add(a);
- }
-
- public List<Address> getCaches() {
- return addresses;
- }
-
- public List<Address> locate(Object key, int replCount) {
- int keyHashCode = hash(key);
- if (keyHashCode == Integer.MIN_VALUE) keyHashCode += 1;
- int hash = Math.abs(keyHashCode);
- int numCopiesToFind = min(replCount, addresses.size());
-
- List<Address> owners = new ArrayList<Address>(numCopiesToFind);
-
- SortedMap<Integer, Address> candidates = positions.tailMap(hash % HASH_SPACE);
-
- int numOwnersFound = 0;
-
- for (Address a : candidates.values()) {
- if (numOwnersFound < numCopiesToFind) {
- owners.add(a);
- numOwnersFound++;
- } else {
- break;
- }
- }
-
- if (numOwnersFound < numCopiesToFind) {
- for (Address a : positions.values()) {
- if (numOwnersFound < numCopiesToFind) {
- owners.add(a);
- numOwnersFound++;
- } else {
- break;
- }
- }
- }
-
- return owners;
- }
-
- @Override
- public boolean isKeyLocalToAddress(Address target, Object key, int replCount) {
- // more efficient impl
- int keyHashCode = hash(key);
- if (keyHashCode == Integer.MIN_VALUE) keyHashCode += 1;
- int hash = Math.abs(keyHashCode);
- int numCopiesToFind = min(replCount, addresses.size());
-
- SortedMap<Integer, Address> candidates = positions.tailMap(hash % HASH_SPACE);
- int nodesTested = 0;
- for (Address a : candidates.values()) {
- if (nodesTested < numCopiesToFind) {
- if (a.equals(target)) return true;
- nodesTested++;
- } else {
- break;
- }
- }
-
- // start from the beginning
- if (nodesTested < numCopiesToFind) {
- for (Address a : positions.values()) {
- if (nodesTested < numCopiesToFind) {
- if (a.equals(target)) return true;
- nodesTested++;
- } else {
- break;
- }
- }
- }
-
- return false;
- }
-
-
- public int getDistance(Address a1, Address a2) {
- if (a1 == null || a2 == null) throw new NullPointerException("Cannot deal with nulls as parameters!");
-
- int p1 = addresses.indexOf(a1);
- if (p1 < 0)
- return -1;
-
- int p2 = addresses.indexOf(a2);
- if (p2 < 0)
- return -1;
-
- if (p1 <= p2)
- return p2 - p1;
- else
- return addresses.size() - (p1 - p2);
- }
-
- public boolean isAdjacent(Address a1, Address a2) {
- int distance = getDistance(a1, a2);
- return distance == 1 || distance == addresses.size() - 1;
- }
-
- @Override
- public int getHashId(Address a) {
- Integer hashId = addressToHashIds.get(a);
- if (hashId == null)
- return -1;
- else
- return hashId.intValue();
- }
-
- @Override
- public int getHashSpace() {
- return HASH_SPACE;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
-
- DefaultConsistentHash that = (DefaultConsistentHash) o;
-
- if (addresses != null ? !addresses.equals(that.addresses) : that.addresses != null) return false;
- if (positions != null ? !positions.equals(that.positions) : that.positions != null) return false;
-
- return true;
- }
-
- @Override
- public int hashCode() {
- int result = addresses != null ? addresses.hashCode() : 0;
- result = 31 * result + (positions != null ? positions.hashCode() : 0);
- return result;
- }
-
- public static class Externalizer implements org.infinispan.marshall.Externalizer {
- public void writeObject(ObjectOutput output, Object subject) throws IOException {
- DefaultConsistentHash dch = (DefaultConsistentHash) subject;
- output.writeObject(dch.addresses);
- output.writeObject(dch.positions);
- output.writeObject(dch.addressToHashIds);
- }
-
- @SuppressWarnings("unchecked")
- public Object readObject(ObjectInput unmarshaller) throws IOException, ClassNotFoundException {
- DefaultConsistentHash dch = new DefaultConsistentHash();
- dch.addresses = (ArrayList<Address>) unmarshaller.readObject();
- dch.positions = (SortedMap<Integer, Address>) unmarshaller.readObject();
- dch.addressToHashIds = (Map<Address, Integer>) unmarshaller.readObject();
- return dch;
- }
- }
-
- @Override
- public String toString() {
- return "DefaultConsistentHash{" +
- "addresses =" + positions +
- ", hash space =" + HASH_SPACE +
- '}';
- }
-}
\ No newline at end of file
Modified: branches/4.2.x/core/src/main/java/org/infinispan/distribution/DistributionManager.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/DistributionManager.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/DistributionManager.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -3,6 +3,7 @@
import org.infinispan.container.entries.CacheEntry;
import org.infinispan.container.entries.InternalCacheEntry;
import org.infinispan.container.entries.InternalCacheValue;
+import org.infinispan.distribution.ch.ConsistentHash;
import org.infinispan.factories.scopes.Scope;
import org.infinispan.factories.scopes.Scopes;
import org.infinispan.loaders.CacheStore;
@@ -72,8 +73,8 @@
/**
* Retrieves the consistent hash instance currently in use, which may be an instance of the configured ConsistentHash
- * instance (which defaults to {@link org.infinispan.distribution.DefaultConsistentHash}, or an instance of
- * {@link org.infinispan.distribution.UnionConsistentHash} if a rehash is in progress.
+ * instance (which defaults to {@link org.infinispan.distribution.ch.DefaultConsistentHash}, or an instance of
+ * {@link org.infinispan.distribution.ch.UnionConsistentHash} if a rehash is in progress.
*
* @return a ConsistentHash instance
*/
@@ -110,7 +111,7 @@
List<Address> requestPermissionToJoin(Address joiner);
/**
- * This will cause all nodes to add the joiner to their consistent hash instance (usually by creating a {@link org.infinispan.distribution.UnionConsistentHash}
+ * This will cause all nodes to add the joiner to their consistent hash instance (usually by creating a {@link org.infinispan.distribution.ch.UnionConsistentHash}
*
* @param joiner address of joiner
* @param starting if true, the joiner is reporting that it is starting the join process. If false, the joiner is
Modified: branches/4.2.x/core/src/main/java/org/infinispan/distribution/DistributionManagerImpl.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/DistributionManagerImpl.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/DistributionManagerImpl.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -14,8 +14,11 @@
import org.infinispan.context.InvocationContext;
import org.infinispan.context.InvocationContextContainer;
-import static org.infinispan.distribution.ConsistentHashHelper.createConsistentHash;
+import static org.infinispan.distribution.ch.ConsistentHashHelper.createConsistentHash;
+import org.infinispan.distribution.ch.ConsistentHash;
+import org.infinispan.distribution.ch.ConsistentHashHelper;
+import org.infinispan.distribution.ch.UnionConsistentHash;
import org.infinispan.factories.annotations.Inject;
import org.infinispan.factories.annotations.Start;
import org.infinispan.factories.annotations.Stop;
@@ -27,7 +30,6 @@
import org.infinispan.loaders.CacheStore;
import org.infinispan.notifications.Listener;
import org.infinispan.notifications.cachemanagerlistener.CacheManagerNotifier;
-import org.infinispan.notifications.cachemanagerlistener.annotation.Merged;
import org.infinispan.notifications.cachemanagerlistener.annotation.ViewChanged;
import org.infinispan.notifications.cachemanagerlistener.event.ViewChangedEvent;
import org.infinispan.remoting.responses.ClusteredGetResponseValidityFilter;
@@ -48,15 +50,7 @@
import org.rhq.helpers.pluginAnnotations.agent.Operation;
import org.rhq.helpers.pluginAnnotations.agent.Parameter;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.HashMap;
+import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -70,7 +64,9 @@
/**
* The default distribution manager implementation
*
- * @author Manik Surtani, Vladimir Blagojevic
+ * @author Manik Surtani
+ * @author Vladimir Blagojevic
+ * @author Mircea.Markus at jboss.com
* @since 4.0
*/
@MBean(objectName = "DistributionManager", description = "Component that handles distribution of content across a cluster")
@@ -222,25 +218,31 @@
Address leaver = MembershipArithmetic.getMemberLeft(oldMembers, newMembers);
log.info("This is a LEAVE event! Node {0} has just left", leaver);
- boolean willReceiveLeaverState = willReceiveLeaverState(leaver);
- boolean willSendLeaverState = willSendLeaverState(leaver);
- List<Address> stateProviders = holdersOfLeaversState(newMembers, leaver);
-
try {
- if (!(consistentHash instanceof UnionConsistentHash)) oldConsistentHash = consistentHash;
- else oldConsistentHash = ((UnionConsistentHash) consistentHash).newCH;
+ if (!(consistentHash instanceof UnionConsistentHash)) {
+ oldConsistentHash = consistentHash;
+ } else {
+ oldConsistentHash = ((UnionConsistentHash) consistentHash).getNewCH();
+ }
consistentHash = ConsistentHashHelper.removeAddress(consistentHash, leaver, configuration);
} catch (Exception e) {
log.fatal("Unable to process leaver!!", e);
throw new CacheException(e);
}
+ boolean willReceiveLeaverState = willReceiveLeaverState(leaver);
+ List<Address> stateProviders = holdersOfLeaversState(leaver);
+ boolean willSendLeaverState = stateProviders.contains(self);
+
if (willReceiveLeaverState || willSendLeaverState) {
log.info("I {0} am participating in rehash", rpcManager.getTransport().getAddress());
transactionLogger.enable();
if (leaveTaskFuture != null
&& (!leaveTaskFuture.isCancelled() || !leaveTaskFuture.isDone())) {
+ if (log.isTraceEnabled()) {
+ log.trace("Canceling running leave task!");
+ }
leaveTaskFuture.cancel(true);
}
@@ -254,27 +256,14 @@
}
}
- boolean willSendLeaverState(Address leaver) {
- ConsistentHash ch = consistentHash instanceof UnionConsistentHash ? oldConsistentHash : consistentHash;
- return ch.isAdjacent(leaver, self);
+ List<Address> holdersOfLeaversState(Address leaver) {
+ List<Address> addresses = oldConsistentHash.getStateProvidersOnLeave(leaver, getReplCount());
+ if (log.isTraceEnabled()) log.trace("Holders of leaver's state are: " + addresses);
+ return addresses;
}
- List<Address> holdersOfLeaversState(List<Address> members, Address leaver) {
- ConsistentHash ch = consistentHash instanceof UnionConsistentHash ? oldConsistentHash : consistentHash;
- Set<Address> holders = new HashSet<Address>();
- for (Address address : members) {
-
- if (ch.isAdjacent(leaver, address)) {
- holders.add(address);
- }
- }
- return new ArrayList<Address>(holders);
- }
-
boolean willReceiveLeaverState(Address leaver) {
- ConsistentHash ch = consistentHash instanceof UnionConsistentHash ? oldConsistentHash : consistentHash;
- int dist = ch.getDistance(leaver, self);
- return dist >= 0 && dist <= getReplCount();
+ return oldConsistentHash.isStateReceiverOnLeave(leaver, self, getReplCount());
}
public boolean isLocal(Object key) {
@@ -333,6 +322,10 @@
this.consistentHash = consistentHash;
}
+ public void setOldConsistentHash(ConsistentHash oldConsistentHash) {
+ this.oldConsistentHash = oldConsistentHash;
+ }
+
@ManagedOperation(description = "Determines whether a given key is affected by an ongoing rehash, if any.")
@Operation(displayName = "Could key be affected by rehash?")
public boolean isAffectedByRehash(@Parameter(name = "key", description = "Key to check") Object key) {
@@ -501,4 +494,12 @@
public String toString() {
return "DistributionManagerImpl[rehashInProgress=" + rehashInProgress + ", consistentHash=" + consistentHash + "]";
}
+
+ public void setSelf(Address self) {
+ this.self = self;
+ }
+
+ public void setConfiguration(Configuration configuration) {
+ this.configuration = configuration;
+ }
}
Deleted: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ExperimentalDefaultConsistentHash.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ExperimentalDefaultConsistentHash.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ExperimentalDefaultConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -1,371 +0,0 @@
-package org.infinispan.distribution;
-
-import org.infinispan.marshall.Ids;
-import org.infinispan.marshall.Marshallable;
-import org.infinispan.remoting.transport.Address;
-import org.infinispan.util.Util;
-
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import static java.lang.Math.min;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
-/**
- * <a href = "http://weblogs.java.net/blog/tomwhite/archive/2007/11/consistent_hash.html">Consistent hashing
- * algorithm</a>. Each target is entered into the pool <i>weight[i]</i>*<i>weightFactor</i> times. Where
- * <i>weight[i]</i> and <i>weightFactor</i> are integers greater than zero.
- * <p/>
- * Based on akluge's impl on <a href-="http://www.vizitsolutions.com/ConsistentHashingCaching.html">http://www.vizitsolutions.com/ConsistentHashingCaching.html</a>
- *
- * @author akluge
- * @author Manik Surtani
- */
- at Marshallable(externalizer = ExperimentalDefaultConsistentHash.Externalizer.class, id = Ids.DEFAULT_CONSISTENT_HASH)
-public class ExperimentalDefaultConsistentHash extends AbstractConsistentHash {
- /**
- * A Weight and weight factor of 1 gives one node per address. In future we may decide to make these configurable,
- * to allow for virtual nodes and a better spread of state across nodes, but we need to ensure we deal with backups
- * not falling on virtual nodes on the same cache instances first.
- */
- private static final int DEFAULT_WEIGHT = 1;
- private static final int DEFAULT_WEIGHTFACTOR = 1;
-
- private List<Address> nodes;
- private List<Entry> pool;
- private int poolSize;
-
- public static class Externalizer implements org.infinispan.marshall.Externalizer {
-
- public void writeObject(ObjectOutput output, Object object) throws IOException {
- ExperimentalDefaultConsistentHash gch = (ExperimentalDefaultConsistentHash) object;
- output.writeObject(gch.nodes);
- }
-
- @SuppressWarnings("unchecked")
- public Object readObject(ObjectInput input) throws IOException, ClassNotFoundException {
- List<Address> addresses = (List<Address>) input.readObject();
- DefaultConsistentHash gch = new DefaultConsistentHash();
- gch.setCaches(addresses);
- return gch;
- }
- }
-
- public List<Address> getCaches() {
- return nodes;
- }
-
- public void setCaches(List<Address> caches) {
- nodes = caches;
- int numNodes = nodes.size();
-
- int poolSize = 0;
-
- for (int i = 0; i < numNodes; i++) {
- poolSize += DEFAULT_WEIGHT * DEFAULT_WEIGHTFACTOR;
- }
- this.poolSize = poolSize;
- pool = new ArrayList<Entry>(poolSize);
-
- int numEntries = 0;
- for (int i = 0; i < numNodes; i++) {
- numEntries = add(nodes.get(i), DEFAULT_WEIGHT * DEFAULT_WEIGHTFACTOR, numEntries);
- }
- Collections.sort(pool);
- nodes = getSortedCachesList();
- }
-
- private List<Address> getSortedCachesList() {
- ArrayList<Address> caches = new ArrayList<Address>();
- for (Entry e : pool) {
- if (!caches.contains(e.address)) caches.add(e.address);
- }
- caches.trimToSize();
- return caches;
- }
-
- /**
- * Adds an Address to the pool of available addresses.
- *
- * @param node Address to be added to the hash pool.
- * @param count An int giving the number of times the node is added to the pool. The count is greater than zero,
- * and is likely greater than 10 if weights are used. The count for the i-th node is usually the
- * <i>weights</i>[i]*<i>weightfactor</i>, if weights are used.
- * @param position The position in the pool to begin adding node entries.
- * @return position;
- */
- private int add(Address node, int count, int position) {
- int hash;
- String nodeName = node.toString();
- for (int i = 0; i < count; i++) {
- hash = hash((Integer.toString(i) + nodeName).getBytes());
- pool.add(position++, new Entry(node, nodeName, i, hash));
- }
- return position;
- }
-
- /**
- * The distance between the first entries in the address array for two caches, a1 and a2. Of questionable use when
- * virtual nodes are employed.
- *
- * @param a1 The address of the first cache.
- * @param a2 The address of the second cache.
- * @return Am int containing the difference between these two indices.
- */
- public int getDistance(Address a1, Address a2) {
- if (a1 == null || a2 == null) throw new NullPointerException("Cannot find the distance between null servers.");
-
- int p1 = nodes.indexOf(a1);
- if (p1 < 0)
- throw new IllegalArgumentException("Address " + a1 + " not in the addresses list of this consistent hash impl!");
-
- int p2 = nodes.indexOf(a2);
- if (p2 < 0)
- throw new IllegalArgumentException("Address " + a2 + " not in the addresses list of this consistent hash impl!");
-
- if (p1 <= p2)
- return p2 - p1;
- else
- return pool.size() - (p1 - p2);
- }
-
- /**
- * Two hashes are adjacent if they are next to each other in the consistent hash.
- *
- * @param a1 The address of the first cache.
- * @param a2 The address of the second cache.
- * @return A boolean, true if they are adjacent, false if not.
- */
- public boolean isAdjacent(Address a1, Address a2) {
- int distance = getDistance(a1, a2);
- return distance == 1 || distance == pool.size() - 1;
- }
-
- public List<Address> locate(Object key, int replCount) {
- if (key == null) throw new NullPointerException("Attempt to get with null key");
-
- int clusterSize = pool.size();
- int numCopiesToFind = min(replCount, clusterSize);
- int hashValue = hash(key);
- return locate(hashValue, numCopiesToFind, replCount);
- }
-
- /**
- * Returns a List of <i>numCopiesToFind</i> unique Addresses.
- *
- * @param hashValue An int, usually a hash, to be mapped to a bin via the CH.
- * @param numCopiesToFind number of copies to find
- * @param replCount replication count
- * @return Returns a List of <i>numCopiesToFind</i> unique Addresses.
- */
- private List<Address> locate(int hashValue, int numCopiesToFind, int replCount) {
- // Stop looking if we have checked the entire pool.
- int checked = 0;
- // Start looking at the first (primary) node for entries for this value.
- int inode = findNearestNodeInPool(hashValue);
- List<Address> nodes = new ArrayList<Address>(numCopiesToFind);
-
- while (nodes.size() < replCount && checked < poolSize) {
- Entry poolEntry;
- if ((poolEntry = pool.get(inode)) != null && nodes.indexOf(poolEntry.address) < 0) {
- nodes.add(poolEntry.address);
- }
- inode = (++inode) % poolSize;
- checked++;
- }
- return nodes;
- }
-
- /**
- * Find a target for a hash key within the pool of node Entries. We search within a slice of the array bounded by
- * lowerBound and upperBound. Further we assume that lowerBound and upperBound are small enough that their sum will
- * not overflow an int.
- * <p/>
- *
- * @param hash The desired hash to locate.
- * @return An int giving the index of the desired entry in the list of targets. If the target is not found, then
- * -(lowerBound +1) will be returned, where lowerBound is the lower bound of the search after possibly
- * several iterations.
- */
- private int binarySearch(int hash) {
- int lowerBound = 0;
- int upperBound = pool.size() - 1;
- while (lowerBound <= upperBound) {
- // Fast div by 2. We assume that the number of targets is small enough
- // that the sum will not overflow an int.
- int mid = (lowerBound + upperBound) >>> 1;
- int currentHash = pool.get(mid).hash;
-
- if (currentHash < hash) {
- lowerBound = mid + 1;
- } else if (currentHash > hash) {
- upperBound = mid - 1;
- } else {
- return mid;
- }
- }
- // The +1 ensures that the return value is negative, even when the hash
- // is off the left edge of the array.
- return -(lowerBound + 1);
- }
-
- /**
- * Finds the lowest index into the pool ArrayList such that the hash of the i-th entry >= hash.
- *
- * @param hash The hash being mapped to a bin via the consistent hash.
- * @return An int, the lowest index into the target array such that the hash of the i-th entry >= hash.
- */
- private int findNearestNodeInPool(int hash) {
- // Find the index of the node - or at least a near one.
- // We only search up to targets.length-1. If the element
- // is greater than the last entry in the list, then map
- // it to the first one.
- int nodeIndex = binarySearch(hash);
-
- // If the returned value is less than zero, then no exact match was found.
- if (nodeIndex < 0) {
- // The value returned is -(lowerBound +1), we want the lower bound back.
- nodeIndex = -(nodeIndex + 1);
-
- // If hash is greater than the last entry, wrap around to the first.
- if (nodeIndex >= pool.size()) {
- nodeIndex = 0;
- }
- }
-
- return nodeIndex;
- }
-
- /**
- * Use the objects built in hash to obtain an initial value, then use a second four byte hash to obtain a more
- * uniform distribution of hash values. This uses a <a href = "http://burtleburtle.net/bob/hash/integer.html">4-byte
- * (integer) hash</a>, which produces well distributed values even when the original hash produces thghtly clustered
- * values.
- * <p/>
- * It is important that the object implement its own hashcode, and not use the Object hashcode.
- *
- * @param object object to hash
- * @return an appropriately spread hash code
- */
- private int hash(Object object) {
- int hash = object.hashCode();
-
- hash = (hash + 0x7ED55D16) + (hash << 12);
- hash = (hash ^ 0xc761c23c) ^ (hash >> 19);
- hash = (hash + 0x165667b1) + (hash << 5);
- hash = (hash + 0xd3a2646c) ^ (hash << 9);
- hash = (hash + 0xfd7046c5) + (hash << 3);
- hash = (hash ^ 0xb55a4f09) ^ (hash >> 16);
-
- return hash;
- }
-
- @Override
- public int getHashId(Address a) {
- throw new RuntimeException("Not yet implemented");
- }
-
- @Override
- public int getHashSpace() {
- return Integer.MAX_VALUE; // Entire positive integer range
- }
-
- /**
- * @return A String representing the object pool.
- */
- @Override
- public String toString() {
- return " pool: " + pool;
- }
-
- @Override
- public boolean equals(Object other) {
- if (other == null
- || !(other instanceof ExperimentalDefaultConsistentHash)) {
- return false;
- }
-
- ExperimentalDefaultConsistentHash otherHash = (ExperimentalDefaultConsistentHash) other;
- return Util.safeEquals(pool, otherHash.pool);
- }
-
- @Override
- public int hashCode() {
- int hashCode = 1;
- for (Entry e : pool) hashCode = 31 * hashCode + e.hash;
- return hashCode;
- }
-
- /**
- * An entry into a consistent hash. It wraps the original object, the object's hash as used to generate the
- * consistent hash, the value extracted from the object used to generate the hash, and the modifier used to
- * differentiate the hash.
- */
- public static class Entry implements Comparable<Entry> {
- public final int differentiator;
- public final int hash;
- public final Address address;
- public final String string;
-
- public Entry(Address address, String string, int differentiator, int hash) {
- this.differentiator = differentiator;
- this.hash = hash;
- this.address = address;
- this.string = string;
- }
-
- /**
- * Compare this Entry with another Entry. First the hash values are compared, then the differentiator is compared.
- * if the hash values are equal.
- *
- * @param other An Entry object to be compared with this object. Returns <ul> <li>-1 if this Entry is less than
- * the other Entry.</li> <li>0 if they are equal.</li> <li>+1 if this Entry is greater than the
- * other Entry.</li> </ul>
- * @return
- */
- public int compareTo(Entry other) {
- if (this.hash < other.hash) {
- return -1;
- }
-
- if (this.hash > other.hash) {
- return 1;
- }
-
- if (this.differentiator < other.differentiator) {
- return -1;
- }
-
- if (this.differentiator > other.differentiator) {
- return +1;
- }
-
- return 0;
- }
-
- @Override
- public boolean equals(Object other) {
- if (other instanceof Entry) {
- Entry otherEntry = (Entry) other;
-
- return hash == otherEntry.hash
- && differentiator == otherEntry.differentiator
- && address.equals(otherEntry.address);
- }
- return false;
- }
-
- @Override
- public int hashCode() {
- return hash;
- }
-
-
- @Override
- public String toString() {
- return string + ":" + Integer.toHexString(hash);
- }
- }
-}
\ No newline at end of file
Modified: branches/4.2.x/core/src/main/java/org/infinispan/distribution/InvertedLeaveTask.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/InvertedLeaveTask.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/InvertedLeaveTask.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -23,6 +23,8 @@
import org.infinispan.config.Configuration;
import org.infinispan.container.DataContainer;
import org.infinispan.container.entries.InternalCacheValue;
+import org.infinispan.distribution.ch.ConsistentHash;
+import org.infinispan.distribution.ch.ConsistentHashHelper;
import org.infinispan.remoting.responses.Response;
import org.infinispan.remoting.responses.SuccessfulResponse;
import org.infinispan.remoting.rpc.RpcManager;
@@ -84,12 +86,10 @@
if (isReceiver) {
Address myAddress = rpcManager.getTransport().getAddress();
RehashControlCommand cmd = cf.buildRehashControlCommand(PULL_STATE_LEAVE, myAddress,
- null, oldCH, newCH,leaversHandled);
+ null, oldCH, newCH, leaversHandled);
- List<Address> addressesWhoMaySendStuff = getStateProviderTargets();
- log.debug("I {0} am pulling state from {1}", self, addressesWhoMaySendStuff);
- List<Response> resps = rpcManager.invokeRemotely(addressesWhoMaySendStuff, cmd,
- SYNCHRONOUS, configuration.getRehashRpcTimeout(), true);
+ log.debug("I {0} am pulling state from {1}", self, stateProviders);
+ List<Response> resps = rpcManager.invokeRemotely(stateProviders, cmd, SYNCHRONOUS, configuration.getRehashRpcTimeout(), true);
log.debug("I {0} received response {1} ", self, resps);
for (Response r : resps) {
@@ -118,17 +118,13 @@
}
}
- private List<Address> getStateProviderTargets() {
- return stateProviders;
- }
-
private void processAndDrainTxLog(ConsistentHash oldCH, ConsistentHash newCH, int replCount) {
- if (trace)
- log.trace("Processing transaction log iteratively");
List<WriteCommand> c;
int i = 0;
TransactionLogger transactionLogger = dmi.getTransactionLogger();
+ if (trace)
+ log.trace("Processing transaction log iteratively: " + transactionLogger);
while (transactionLogger.shouldDrainWithoutLock()) {
if (trace)
log.trace("Processing transaction log, iteration {0}", i++);
@@ -182,8 +178,7 @@
transactionLogger.unlockAndDisable();
}
- private void apply(ConsistentHash oldCH, ConsistentHash newCH, int replCount,
- List<WriteCommand> wc) {
+ private void apply(ConsistentHash oldCH, ConsistentHash newCH, int replCount, List<WriteCommand> wc) {
// need to create another "state map"
TransactionLogMap state = new TransactionLogMap(leavers, oldCH, newCH, replCount);
for (WriteCommand c : wc)
Modified: branches/4.2.x/core/src/main/java/org/infinispan/distribution/JoinTask.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/JoinTask.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/JoinTask.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -7,7 +7,9 @@
import org.infinispan.config.Configuration;
import org.infinispan.container.DataContainer;
import org.infinispan.container.entries.InternalCacheValue;
-import static org.infinispan.distribution.ConsistentHashHelper.createConsistentHash;
+import static org.infinispan.distribution.ch.ConsistentHashHelper.createConsistentHash;
+
+import org.infinispan.distribution.ch.ConsistentHash;
import org.infinispan.remoting.responses.Response;
import org.infinispan.remoting.responses.SuccessfulResponse;
import static org.infinispan.remoting.rpc.ResponseMode.SYNCHRONOUS;
@@ -18,7 +20,6 @@
import org.infinispan.util.logging.Log;
import org.infinispan.util.logging.LogFactory;
-import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
@@ -210,25 +211,7 @@
* @return
*/
List<Address> getAddressesWhoMaySendStuff(ConsistentHash ch, int replCount) {
- List<Address> l = new LinkedList<Address>();
- List<Address> caches = ch.getCaches();
- int selfIdx = caches.indexOf(self);
- if (selfIdx >= replCount - 1) {
- l.addAll(caches.subList(selfIdx - replCount + 1, selfIdx));
- } else {
- l.addAll(caches.subList(0, selfIdx));
- int alreadyCollected = l.size();
- l.addAll(caches.subList(caches.size() - replCount + 1 + alreadyCollected, caches.size()));
- }
-
- Address plusOne;
- if (selfIdx == caches.size() - 1)
- plusOne = caches.get(0);
- else
- plusOne = caches.get(selfIdx + 1);
-
- if (!l.contains(plusOne)) l.add(plusOne);
- return l;
+ return ch.getStateProvidersOnJoin(self, replCount);
}
public Address getMyAddress() {
Modified: branches/4.2.x/core/src/main/java/org/infinispan/distribution/RehashTask.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/RehashTask.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/RehashTask.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -4,6 +4,7 @@
import org.infinispan.commands.write.InvalidateCommand;
import org.infinispan.config.Configuration;
import org.infinispan.container.DataContainer;
+import org.infinispan.distribution.ch.ConsistentHash;
import org.infinispan.remoting.rpc.RpcManager;
import org.infinispan.remoting.transport.Address;
import org.infinispan.util.concurrent.NotifyingFutureImpl;
Modified: branches/4.2.x/core/src/main/java/org/infinispan/distribution/TransactionLoggerImpl.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/TransactionLoggerImpl.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/TransactionLoggerImpl.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -166,4 +166,12 @@
uncommittedPrepares.clear();
return commands;
}
+
+ @Override
+ public String toString() {
+ return "TransactionLoggerImpl{" +
+ "commandQueue=" + commandQueue +
+ ", uncommittedPrepares=" + uncommittedPrepares +
+ '}';
+ }
}
Deleted: branches/4.2.x/core/src/main/java/org/infinispan/distribution/UnionConsistentHash.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/UnionConsistentHash.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/UnionConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -1,92 +0,0 @@
-package org.infinispan.distribution;
-
-import org.infinispan.CacheException;
-import org.infinispan.marshall.Ids;
-import org.infinispan.marshall.Marshallable;
-import org.infinispan.remoting.transport.Address;
-import org.infinispan.util.Immutables;
-
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.util.Collections;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Set;
-
-/**
- * A delegating wrapper that locates keys by getting a union of locations reported by two other ConsistentHash
- * implementations it delegates to.
- *
- * @author Manik Surtani
- * @since 4.0
- */
- at Marshallable(externalizer = UnionConsistentHash.Externalizer.class, id = Ids.UNION_CONSISTENT_HASH)
-public class UnionConsistentHash extends AbstractConsistentHash {
-
- ConsistentHash oldCH, newCH;
-
- public UnionConsistentHash(ConsistentHash oldCH, ConsistentHash newCH) {
- if ((oldCH instanceof UnionConsistentHash) || (newCH instanceof UnionConsistentHash))
- throw new CacheException("Expecting both newCH and oldCH to not be Unions!! oldCH=[" + oldCH.getClass() + "] and newCH=[" + newCH.getClass() + "]");
- this.oldCH = oldCH;
- this.newCH = newCH;
- }
-
- public void setCaches(List<Address> caches) {
- // no op
- }
-
- public List<Address> getCaches() {
- return Collections.emptyList();
- }
-
- public List<Address> locate(Object key, int replCount) {
- Set<Address> addresses = new LinkedHashSet<Address>();
- addresses.addAll(oldCH.locate(key, replCount));
- addresses.addAll(newCH.locate(key, replCount));
- return Immutables.immutableListConvert(addresses);
- }
-
- public int getDistance(Address a1, Address a2) {
- throw new UnsupportedOperationException("Unsupported!");
- }
-
- public boolean isAdjacent(Address a1, Address a2) {
- throw new UnsupportedOperationException("Unsupported!");
- }
-
- @Override
- public int getHashId(Address a) {
- throw new UnsupportedOperationException("Unsupported!");
- }
-
- @Override
- public int getHashSpace() {
- int oldHashSpace = oldCH.getHashSpace();
- int newHashSpace = newCH.getHashSpace();
- // In a union, the hash space is the biggest of the hash spaces.
- return oldHashSpace > newHashSpace ? oldHashSpace : newHashSpace;
- }
-
- public ConsistentHash getNewConsistentHash() {
- return newCH;
- }
-
- public ConsistentHash getOldConsistentHash() {
- return oldCH;
- }
-
- public static class Externalizer implements org.infinispan.marshall.Externalizer {
-
- public void writeObject(ObjectOutput output, Object object) throws IOException {
- UnionConsistentHash uch = (UnionConsistentHash) object;
- output.writeObject(uch.oldCH);
- output.writeObject(uch.newCH);
- }
-
- public Object readObject(ObjectInput input) throws IOException, ClassNotFoundException {
- return new UnionConsistentHash((ConsistentHash) input.readObject(), (ConsistentHash) input.readObject());
- }
- }
-}
Copied: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/AbstractConsistentHash.java (from rev 2610, branches/4.2.x/core/src/main/java/org/infinispan/distribution/AbstractConsistentHash.java)
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/AbstractConsistentHash.java (rev 0)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/AbstractConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,45 @@
+package org.infinispan.distribution.ch;
+
+import org.infinispan.remoting.transport.Address;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An abstract consistent hash implementation that handles common implementations of certain methods. In particular,
+ * default implementations of {@link #locateAll(java.util.Collection, int)} and {@link #isKeyLocalToAddress(org.infinispan.remoting.transport.Address, Object, int)}.
+ * <p />
+ * The versions provided here are relatively inefficient in that they call {@link #locate(Object, int)} first (and
+ * sometimes in a loop). Depending on the algorithm used, there may be more efficient ways to achieve the same results
+ * and in such cases the methods provided here should be overridden.
+ * <p />
+ * @author Manik Surtani
+ * @author Mircea.Markus at jboss.com
+ * @since 4.0
+ */
+public abstract class AbstractConsistentHash implements ConsistentHash {
+
+ protected volatile List<Address> caches;
+ protected TopologyInfo topologyInfo;
+
+ public void setCaches(List<Address> caches) {
+ this.caches = caches;
+ }
+
+ public Map<Object, List<Address>> locateAll(Collection<Object> keys, int replCount) {
+ Map<Object, List<Address>> locations = new HashMap<Object, List<Address>>();
+ for (Object k : keys) locations.put(k, locate(k, replCount));
+ return locations;
+ }
+
+ public boolean isKeyLocalToAddress(Address a, Object key, int replCount) {
+ // simple, brute-force impl
+ return locate(key, replCount).contains(a);
+ }
+
+ public void setTopologyInfo(TopologyInfo topologyInfo) {
+ this.topologyInfo = topologyInfo;
+ }
+}
Property changes on: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/AbstractConsistentHash.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF
Added: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/AbstractWheelConsistentHash.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/AbstractWheelConsistentHash.java (rev 0)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/AbstractWheelConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,99 @@
+package org.infinispan.distribution.ch;
+
+import org.infinispan.remoting.transport.Address;
+import org.infinispan.util.logging.Log;
+import org.infinispan.util.logging.LogFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+import static org.infinispan.util.hash.MurmurHash2.hash;
+
+/**
+ * Abstract class for the wheel-based CH implementations.
+ *
+ * @author Mircea.Markus at jboss.com
+ * @since 4.2
+ */
+public abstract class AbstractWheelConsistentHash extends AbstractConsistentHash {
+
+ private static Log log = LogFactory.getLog(AbstractWheelConsistentHash.class);
+ protected ArrayList<Address> addresses;
+ protected SortedMap<Integer, Address> positions;
+ // TODO: Maybe address and addressToHashIds can be combined in a LinkedHashMap?
+ protected Map<Address, Integer> addressToHashIds;
+
+ final static int HASH_SPACE = 10240; // no more than 10k nodes?
+
+ public void setCaches(List<Address> caches) {
+ super.setCaches(caches);
+
+ addresses = new ArrayList<Address>(caches);
+
+ // this list won't grow.
+ addresses.trimToSize();
+
+ positions = new TreeMap<Integer, Address>();
+ addressToHashIds = new HashMap<Address, Integer>();
+
+ for (Address a : addresses) {
+ int positionIndex = Math.abs(hash(a)) % HASH_SPACE;
+ // this is deterministic since the address list is ordered and the order is consistent across the grid
+ while (positions.containsKey(positionIndex)) positionIndex = positionIndex + 1 % HASH_SPACE;
+ positions.put(positionIndex, a);
+ // If address appears several times, take the lowest value to guarantee that
+ // at least the initial value and subsequent +1 values would end up in the same node
+ // TODO: Remove this check since https://jira.jboss.org/jira/browse/ISPN-428 contains a proper fix for this
+ if (!addressToHashIds.containsKey(a))
+ addressToHashIds.put(a, positionIndex);
+ }
+
+ addresses.clear();
+ // reorder addresses as per the positions.
+ for (Address a : positions.values()) addresses.add(a);
+ if (log.isTraceEnabled()) {
+ log.trace("Position are: " + positions);
+ }
+ }
+
+ public boolean isStateReceiverOnLeave(Address leaver, Address node, int replCount) {
+ for (Address address : addresses) {
+ List<Address> backups = locate(address, replCount + 1);
+ if (backups.contains(leaver) && (backups.indexOf(node) == backups.size() - 1)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+
+ public List<Address> getCaches() {
+ return addresses;
+ }
+
+ @Override
+ public int getHashSpace() {
+ return HASH_SPACE;
+ }
+
+
+ @Override
+ public int getHashId(Address a) {
+ Integer hashId = addressToHashIds.get(a);
+ if (hashId == null)
+ return -1;
+ else
+ return hashId.intValue();
+ }
+
+ protected int getNormalizedHash(Object key) {
+ // more efficient impl
+ int keyHashCode = hash(key);
+ if (keyHashCode == Integer.MIN_VALUE) keyHashCode += 1;
+ return Math.abs(keyHashCode) % HASH_SPACE;
+ }
+}
Copied: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ConsistentHash.java (from rev 2610, branches/4.2.x/core/src/main/java/org/infinispan/distribution/ConsistentHash.java)
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ConsistentHash.java (rev 0)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,114 @@
+package org.infinispan.distribution.ch;
+
+import org.infinispan.remoting.transport.Address;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A consistent hash algorithm implementation. Implementations would typically be constructed via reflection so should
+ * implement a public, no-arg constructor.
+ *
+ * @author Manik Surtani
+ * @author Mircea.Markus at jboss.com
+ * @since 4.0
+ */
+public interface ConsistentHash {
+
+ /**
+ * Sets the collection of cache addresses in the cluster. The implementation should store these internally and use
+ * these to locate keys.
+ *
+ * @param caches caches in cluster.
+ */
+ void setCaches(List<Address> caches);
+
+ /**
+ * Sets cluster topology information that can be used by CH to improve fault tolerance by smart choosing of backups.
+ * More about it <a href="http://community.jboss.org/wiki/DesigningServerHinting">here<a/>
+ */
+ void setTopologyInfo(TopologyInfo topologyInfo);
+
+ /**
+ * Should return a collection of cache addresses in the cluster.
+ *
+ * @return collection of cache addresses
+ */
+ List<Address> getCaches();
+
+ /**
+ * Locates a key, given a replication count (number of copies).
+ *
+ * @param key key to locate
+ * @param replCount replication count (number of copies)
+ * @return a list of addresses where the key resides, where this list is a subset of the addresses set in {@link
+ * #setCaches(java.util.List)}. Should never be null, and should contain replCount elements or the max
+ * number of caches available, whichever is smaller.
+ */
+ List<Address> locate(Object key, int replCount);
+
+ /**
+ * The logical equivalent of calling {@link #locate(Object, int)} multiple times for each key in the collection of
+ * keys. Implementations may be optimised for such a bulk lookup, or may just repeatedly call {@link #locate(Object,
+ * int)}.
+ *
+ * @param keys keys to locate
+ * @param replCount replication count (number of copies) for each key
+ * @return Map of locations, keyed on key.
+ */
+ Map<Object, List<Address>> locateAll(Collection<Object> keys, int replCount);
+
+ /**
+ * Test to see whether a key is mapped to a given address.
+ * @param a address to test
+ * @param key key to test
+ * @param replCount repl count
+ * @return true if the key is mapped to the address; false otherwise
+ */
+ boolean isKeyLocalToAddress(Address a, Object key, int replCount);
+
+ /**
+ * Returns the value between 0 and the hash space limit, or hash id, for a particular address. If there's no such
+ * value for an address, this method will return -1.
+ *
+ * @return An int between 0 and hash space if the address is present in the hash wheel, otherwise it returns -1.
+ */
+ int getHashId(Address a);
+
+ /**
+ * Returns the hash space constant for this consistent hash algorithm class. This integer is often used as modulus
+ * for arithmetic operations within the algorithm, for example, limiting the range of possible hash values.
+ *
+ * @return A positive integer containing the hash space constant or 0 is not supported by implementation.
+ */
+ int getHashSpace();
+
+ /**
+ * Returns the nodes that need will replicate their state if the specified node crashes. The return collection
+ * should contain all the nodes that backup-ed on leaver and one of the nodes which acted as a backup for the leaver .
+ * <p>
+ * Pre: leaver must be present in the caches known to this CH, as returned by {@link #getCaches()}
+ * @param leaver the node that leaves the cluster
+ * @param replCount
+ */
+ List<Address> getStateProvidersOnLeave(Address leaver, int replCount);
+
+ /**
+ * Is the specified node going to receive state if when another node leaves the cluster?
+ * When a node leaves the cluster following nodes would need to receive state as result of the rehashing:
+ * - a new backup node for the lever to satisfy numOwners condition
+ * - the nodes that would replace the leaver as a backup for other nodes
+ * @param leaver node that leaves
+ * @param node is this state receiver?
+ * @param replCount numOwners
+ */
+ boolean isStateReceiverOnLeave(Address leaver, Address node, int replCount);
+
+ /**
+ * Returns the nodes that would act as state providers when a new node joins:
+ * - the nodes for which the joiner is a backup
+ * - the nodes that held joiner's state
+ */
+ List<Address> getStateProvidersOnJoin(Address joiner, int replCount);
+}
Property changes on: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ConsistentHash.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF
Copied: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ConsistentHashHelper.java (from rev 2610, branches/4.2.x/core/src/main/java/org/infinispan/distribution/ConsistentHashHelper.java)
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ConsistentHashHelper.java (rev 0)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ConsistentHashHelper.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,144 @@
+package org.infinispan.distribution.ch;
+
+import org.infinispan.config.Configuration;
+import org.infinispan.remoting.transport.Address;
+import org.infinispan.util.Util;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * A helper class that handles the construction of consistent hash instances based on configuration.
+ *
+ * @author Manik Surtani
+ * @since 4.0
+ */
+public class ConsistentHashHelper {
+
+ /**
+ * Returns a new consistent hash of the same type with the given address removed.
+ *
+ * @param ch consistent hash to start with
+ * @param toRemove address to remove
+ * @param c configuration
+ * @return a new consistent hash instance of the same type
+ */
+ public static ConsistentHash removeAddress(ConsistentHash ch, Address toRemove, Configuration c) {
+ if (ch instanceof UnionConsistentHash)
+ return removeAddressFromUnionConsistentHash((UnionConsistentHash) ch, toRemove, c);
+ else {
+ ConsistentHash newCH = (ConsistentHash) Util.getInstance(c.getConsistentHashClass());
+ List<Address> caches = new ArrayList<Address>(ch.getCaches());
+ caches.remove(toRemove);
+ newCH.setCaches(caches);
+ return newCH;
+ }
+ }
+
+ /**
+ * Creates a new UnionConsistentHash instance based on the old instance, removing the provided address from both
+ * target consistent hash instances in the union.
+ *
+ * @param uch union consistent hash instance
+ * @param toRemove address to remove
+ * @param c configuration
+ * @return a new UnionConsistentHash instance
+ */
+ public static UnionConsistentHash removeAddressFromUnionConsistentHash(UnionConsistentHash uch, Address toRemove, Configuration c) {
+ ConsistentHash newFirstCH = removeAddress(uch.getOldConsistentHash(), toRemove, c);
+ ConsistentHash newSecondCH = removeAddress(uch.getNewConsistentHash(), toRemove, c);
+ return new UnionConsistentHash(newFirstCH, newSecondCH);
+ }
+
+ /**
+ * Creates a new consistent hash instance based on the user's configuration, and populates the consistent hash
+ * with the collection of addresses passed in.
+ *
+ * @param c configuration
+ * @param addresses with which to populate the consistent hash
+ * @return a new consistent hash instance
+ */
+ public static ConsistentHash createConsistentHash(Configuration c, List<Address> addresses) {
+ ConsistentHash ch = (ConsistentHash) Util.getInstance(c.getConsistentHashClass());
+ ch.setCaches(addresses);
+ return ch;
+ }
+
+ /**
+ * Creates a new consistent hash instance based on the user's configuration, and populates the consistent hash
+ * with the collection of addresses passed in.
+ *
+ * @param c configuration
+ * @param addresses with which to populate the consistent hash
+ * @param moreAddresses to add to the list of addresses
+ * @return a new consistent hash instance
+ */
+ public static ConsistentHash createConsistentHash(Configuration c, List<Address> addresses, Address... moreAddresses) {
+ List<Address> list = new LinkedList<Address>(addresses);
+ list.addAll(Arrays.asList(moreAddresses));
+ return createConsistentHash(c, list);
+ }
+
+ /**
+ * Creates a new consistent hash instance based on the user's configuration, and populates the consistent hash
+ * with the collection of addresses passed in.
+ *
+ * @param c configuration
+ * @param addresses with which to populate the consistent hash
+ * @param moreAddresses to add to the list of addresses
+ * @return a new consistent hash instance
+ */
+ public static ConsistentHash createConsistentHash(Configuration c, List<Address> addresses, Collection<Address> moreAddresses) {
+ List<Address> list = new LinkedList<Address>(addresses);
+ list.addAll(moreAddresses);
+ return createConsistentHash(c, list);
+ }
+
+ /**
+ * Creates a new consistent hash instance based on the type specified, and populates the consistent hash
+ * with the collection of addresses passed in.
+ *
+ * @param clazz type of the consistent hash to create
+ * @param addresses with which to populate the consistent hash
+ * @return a new consistent hash instance
+ */
+ public static ConsistentHash createConsistentHash(Class<? extends ConsistentHash> clazz, List<Address> addresses) {
+ ConsistentHash ch;
+ ch = Util.getInstance(clazz);
+ if (addresses != null && !addresses.isEmpty()) ch.setCaches(addresses);
+ return ch;
+ }
+
+ /**
+ * Creates a new consistent hash instance based on the type specified, and populates the consistent hash
+ * with the collection of addresses passed in.
+ *
+ * @param clazz type of the consistent hash to create
+ * @param addresses with which to populate the consistent hash
+ * @param moreAddresses to add to the list of addresses
+ * @return a new consistent hash instance
+ */
+ public static ConsistentHash createConsistentHash(Class<? extends ConsistentHash> clazz, List<Address> addresses, Address... moreAddresses) {
+ List<Address> list = new LinkedList<Address>(addresses);
+ list.addAll(Arrays.asList(moreAddresses));
+ return createConsistentHash(clazz, list);
+ }
+
+ /**
+ * Creates a new consistent hash instance based on the type specified, and populates the consistent hash
+ * with the collection of addresses passed in.
+ *
+ * @param clazz type of the consistent hash to create
+ * @param addresses with which to populate the consistent hash
+ * @param moreAddresses to add to the list of addresses
+ * @return a new consistent hash instance
+ */
+ public static ConsistentHash createConsistentHash(Class<? extends ConsistentHash> clazz, List<Address> addresses, Collection<Address> moreAddresses) {
+ List<Address> list = new LinkedList<Address>(addresses);
+ list.addAll(moreAddresses);
+ return createConsistentHash(clazz, list);
+ }
+}
Property changes on: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ConsistentHashHelper.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF
Copied: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/DefaultConsistentHash.java (from rev 2610, branches/4.2.x/core/src/main/java/org/infinispan/distribution/DefaultConsistentHash.java)
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/DefaultConsistentHash.java (rev 0)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/DefaultConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,199 @@
+package org.infinispan.distribution.ch;
+
+import org.infinispan.marshall.Ids;
+import org.infinispan.marshall.Marshallable;
+import org.infinispan.remoting.transport.Address;
+import org.infinispan.util.logging.Log;
+import org.infinispan.util.logging.LogFactory;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
+
+import static java.lang.Math.min;
+
+ at Marshallable(externalizer = DefaultConsistentHash.Externalizer.class, id = Ids.DEFAULT_CONSISTENT_HASH)
+public class DefaultConsistentHash extends AbstractWheelConsistentHash {
+
+ private static Log log = LogFactory.getLog(DefaultConsistentHash.class);
+
+ public List<Address> locate(Object key, int replCount) {
+ int hash = getNormalizedHash(key);
+ int numCopiesToFind = min(replCount, addresses.size());
+
+ List<Address> owners = new ArrayList<Address>(numCopiesToFind);
+
+ SortedMap<Integer, Address> candidates = positions.tailMap(hash);
+
+ int numOwnersFound = 0;
+
+ for (Address a : candidates.values()) {
+ if (numOwnersFound < numCopiesToFind) {
+ owners.add(a);
+ numOwnersFound++;
+ } else {
+ break;
+ }
+ }
+
+ if (numOwnersFound < numCopiesToFind) {
+ for (Address a : positions.values()) {
+ if (numOwnersFound < numCopiesToFind) {
+ owners.add(a);
+ numOwnersFound++;
+ } else {
+ break;
+ }
+ }
+ }
+
+ return owners;
+ }
+
+ @Override
+ public boolean isKeyLocalToAddress(Address target, Object key, int replCount) {
+ int hash = getNormalizedHash(key);
+ int numCopiesToFind = min(replCount, addresses.size());
+
+ SortedMap<Integer, Address> candidates = positions.tailMap(hash);
+ int nodesTested = 0;
+ for (Address a : candidates.values()) {
+ if (nodesTested < numCopiesToFind) {
+ if (a.equals(target)) return true;
+ nodesTested++;
+ } else {
+ break;
+ }
+ }
+
+ // start from the beginning
+ if (nodesTested < numCopiesToFind) {
+ for (Address a : positions.values()) {
+ if (nodesTested < numCopiesToFind) {
+ if (a.equals(target)) return true;
+ nodesTested++;
+ } else {
+ break;
+ }
+ }
+ }
+
+ return false;
+ }
+
+
+ public int getDistance(Address a1, Address a2) {
+ if (a1 == null || a2 == null) throw new NullPointerException("Cannot deal with nulls as parameters!");
+
+ int p1 = addresses.indexOf(a1);
+ if (p1 < 0)
+ return -1;
+
+ int p2 = addresses.indexOf(a2);
+ if (p2 < 0)
+ return -1;
+
+ if (p1 <= p2)
+ return p2 - p1;
+ else
+ return addresses.size() - (p1 - p2);
+ }
+
+ public boolean isAdjacent(Address a1, Address a2) {
+ int distance = getDistance(a1, a2);
+ return distance == 1 || distance == addresses.size() - 1;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ DefaultConsistentHash that = (DefaultConsistentHash) o;
+
+ if (addresses != null ? !addresses.equals(that.addresses) : that.addresses != null) return false;
+ if (positions != null ? !positions.equals(that.positions) : that.positions != null) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = addresses != null ? addresses.hashCode() : 0;
+ result = 31 * result + (positions != null ? positions.hashCode() : 0);
+ return result;
+ }
+
+ public static class Externalizer implements org.infinispan.marshall.Externalizer {
+ public void writeObject(ObjectOutput output, Object subject) throws IOException {
+ DefaultConsistentHash dch = (DefaultConsistentHash) subject;
+ output.writeObject(dch.addresses);
+ output.writeObject(dch.positions);
+ output.writeObject(dch.addressToHashIds);
+ }
+
+ @SuppressWarnings("unchecked")
+ public Object readObject(ObjectInput unmarshaller) throws IOException, ClassNotFoundException {
+ DefaultConsistentHash dch = new DefaultConsistentHash();
+ dch.addresses = (ArrayList<Address>) unmarshaller.readObject();
+ dch.positions = (SortedMap<Integer, Address>) unmarshaller.readObject();
+ dch.addressToHashIds = (Map<Address, Integer>) unmarshaller.readObject();
+ return dch;
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "DefaultConsistentHash{" +
+ "addresses =" + positions +
+ ", hash space =" + HASH_SPACE +
+ '}';
+ }
+
+ public List<Address> getAddressOnTheWheel() {
+ return addresses;
+ }
+
+ public List<Address> getStateProvidersOnJoin(Address self, int replCount) {
+ List<Address> l = new LinkedList<Address>();
+ List<Address> caches = getCaches();
+ int selfIdx = caches.indexOf(self);
+ if (selfIdx >= replCount - 1) {
+ l.addAll(caches.subList(selfIdx - replCount + 1, selfIdx));
+ } else {
+ l.addAll(caches.subList(0, selfIdx));
+ int alreadyCollected = l.size();
+ l.addAll(caches.subList(caches.size() - replCount + 1 + alreadyCollected, caches.size()));
+ }
+
+ Address plusOne;
+ if (selfIdx == caches.size() - 1)
+ plusOne = caches.get(0);
+ else
+ plusOne = caches.get(selfIdx + 1);
+
+ if (!l.contains(plusOne)) l.add(plusOne);
+ return l;
+ }
+
+ public List<Address> getStateProvidersOnLeave(Address leaver, int replCount) {
+ if (log.isTraceEnabled()) log.trace("List of addresses is: " + addresses + ". leaver is: " + leaver);
+ Set<Address> holders = new HashSet<Address>();
+ for (Address address : addresses) {
+ if (isAdjacent(leaver, address)) {
+ holders.add(address);
+ if (log.isTraceEnabled()) log.trace(address + " is state holder");
+ } else {
+ if (log.isTraceEnabled()) log.trace(address + " NOT state holder");
+ }
+ }
+ return new ArrayList<Address>(holders);
+ }
+}
\ No newline at end of file
Property changes on: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/DefaultConsistentHash.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF
Copied: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ExperimentalDefaultConsistentHash.java (from rev 2610, branches/4.2.x/core/src/main/java/org/infinispan/distribution/ExperimentalDefaultConsistentHash.java)
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ExperimentalDefaultConsistentHash.java (rev 0)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ExperimentalDefaultConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,392 @@
+package org.infinispan.distribution.ch;
+
+import org.infinispan.marshall.Ids;
+import org.infinispan.marshall.Marshallable;
+import org.infinispan.remoting.transport.Address;
+import org.infinispan.util.Util;
+
+import javax.naming.OperationNotSupportedException;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import static java.lang.Math.min;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * <a href = "http://weblogs.java.net/blog/tomwhite/archive/2007/11/consistent_hash.html">Consistent hashing
+ * algorithm</a>. Each target is entered into the pool <i>weight[i]</i>*<i>weightFactor</i> times. Where
+ * <i>weight[i]</i> and <i>weightFactor</i> are integers greater than zero.
+ * <p/>
+ * Based on akluge's impl on <a href-="http://www.vizitsolutions.com/ConsistentHashingCaching.html">http://www.vizitsolutions.com/ConsistentHashingCaching.html</a>
+ *
+ * @author akluge
+ * @author Manik Surtani
+ */
+ at Marshallable(externalizer = ExperimentalDefaultConsistentHash.Externalizer.class, id = Ids.DEFAULT_CONSISTENT_HASH)
+public class ExperimentalDefaultConsistentHash extends AbstractConsistentHash {
+ /**
+ * A Weight and weight factor of 1 gives one node per address. In future we may decide to make these configurable,
+ * to allow for virtual nodes and a better spread of state across nodes, but we need to ensure we deal with backups
+ * not falling on virtual nodes on the same cache instances first.
+ */
+ private static final int DEFAULT_WEIGHT = 1;
+ private static final int DEFAULT_WEIGHTFACTOR = 1;
+
+ private List<Address> nodes;
+ private List<Entry> pool;
+ private int poolSize;
+
+ public static class Externalizer implements org.infinispan.marshall.Externalizer {
+
+ public void writeObject(ObjectOutput output, Object object) throws IOException {
+ ExperimentalDefaultConsistentHash gch = (ExperimentalDefaultConsistentHash) object;
+ output.writeObject(gch.nodes);
+ }
+
+ @SuppressWarnings("unchecked")
+ public Object readObject(ObjectInput input) throws IOException, ClassNotFoundException {
+ List<Address> addresses = (List<Address>) input.readObject();
+ DefaultConsistentHash gch = new DefaultConsistentHash();
+ gch.setCaches(addresses);
+ return gch;
+ }
+ }
+
+ public List<Address> getCaches() {
+ return nodes;
+ }
+
+ public void setCaches(List<Address> caches) {
+ nodes = caches;
+ int numNodes = nodes.size();
+
+ int poolSize = 0;
+
+ for (int i = 0; i < numNodes; i++) {
+ poolSize += DEFAULT_WEIGHT * DEFAULT_WEIGHTFACTOR;
+ }
+ this.poolSize = poolSize;
+ pool = new ArrayList<Entry>(poolSize);
+
+ int numEntries = 0;
+ for (int i = 0; i < numNodes; i++) {
+ numEntries = add(nodes.get(i), DEFAULT_WEIGHT * DEFAULT_WEIGHTFACTOR, numEntries);
+ }
+ Collections.sort(pool);
+ nodes = getSortedCachesList();
+ }
+
+ private List<Address> getSortedCachesList() {
+ ArrayList<Address> caches = new ArrayList<Address>();
+ for (Entry e : pool) {
+ if (!caches.contains(e.address)) caches.add(e.address);
+ }
+ caches.trimToSize();
+ return caches;
+ }
+
+ /**
+ * Adds an Address to the pool of available addresses.
+ *
+ * @param node Address to be added to the hash pool.
+ * @param count An int giving the number of times the node is added to the pool. The count is greater than zero,
+ * and is likely greater than 10 if weights are used. The count for the i-th node is usually the
+ * <i>weights</i>[i]*<i>weightfactor</i>, if weights are used.
+ * @param position The position in the pool to begin adding node entries.
+ * @return position;
+ */
+ private int add(Address node, int count, int position) {
+ int hash;
+ String nodeName = node.toString();
+ for (int i = 0; i < count; i++) {
+ hash = hash((Integer.toString(i) + nodeName).getBytes());
+ pool.add(position++, new Entry(node, nodeName, i, hash));
+ }
+ return position;
+ }
+
+ /**
+ * The distance between the first entries in the address array for two caches, a1 and a2. Of questionable use when
+ * virtual nodes are employed.
+ *
+ * @param a1 The address of the first cache.
+ * @param a2 The address of the second cache.
+ * @return Am int containing the difference between these two indices.
+ */
+ public int getDistance(Address a1, Address a2) {
+ if (a1 == null || a2 == null) throw new NullPointerException("Cannot find the distance between null servers.");
+
+ int p1 = nodes.indexOf(a1);
+ if (p1 < 0)
+ throw new IllegalArgumentException("Address " + a1 + " not in the addresses list of this consistent hash impl!");
+
+ int p2 = nodes.indexOf(a2);
+ if (p2 < 0)
+ throw new IllegalArgumentException("Address " + a2 + " not in the addresses list of this consistent hash impl!");
+
+ if (p1 <= p2)
+ return p2 - p1;
+ else
+ return pool.size() - (p1 - p2);
+ }
+
+ /**
+ * Two hashes are adjacent if they are next to each other in the consistent hash.
+ *
+ * @param a1 The address of the first cache.
+ * @param a2 The address of the second cache.
+ * @return A boolean, true if they are adjacent, false if not.
+ */
+ public boolean isAdjacent(Address a1, Address a2) {
+ int distance = getDistance(a1, a2);
+ return distance == 1 || distance == pool.size() - 1;
+ }
+
+ public List<Address> locate(Object key, int replCount) {
+ if (key == null) throw new NullPointerException("Attempt to get with null key");
+
+ int clusterSize = pool.size();
+ int numCopiesToFind = min(replCount, clusterSize);
+ int hashValue = hash(key);
+ return locate(hashValue, numCopiesToFind, replCount);
+ }
+
+ /**
+ * Returns a List of <i>numCopiesToFind</i> unique Addresses.
+ *
+ * @param hashValue An int, usually a hash, to be mapped to a bin via the CH.
+ * @param numCopiesToFind number of copies to find
+ * @param replCount replication count
+ * @return Returns a List of <i>numCopiesToFind</i> unique Addresses.
+ */
+ private List<Address> locate(int hashValue, int numCopiesToFind, int replCount) {
+ // Stop looking if we have checked the entire pool.
+ int checked = 0;
+ // Start looking at the first (primary) node for entries for this value.
+ int inode = findNearestNodeInPool(hashValue);
+ List<Address> nodes = new ArrayList<Address>(numCopiesToFind);
+
+ while (nodes.size() < replCount && checked < poolSize) {
+ Entry poolEntry;
+ if ((poolEntry = pool.get(inode)) != null && nodes.indexOf(poolEntry.address) < 0) {
+ nodes.add(poolEntry.address);
+ }
+ inode = (++inode) % poolSize;
+ checked++;
+ }
+ return nodes;
+ }
+
+ /**
+ * Find a target for a hash key within the pool of node Entries. We search within a slice of the array bounded by
+ * lowerBound and upperBound. Further we assume that lowerBound and upperBound are small enough that their sum will
+ * not overflow an int.
+ * <p/>
+ *
+ * @param hash The desired hash to locate.
+ * @return An int giving the index of the desired entry in the list of targets. If the target is not found, then
+ * -(lowerBound +1) will be returned, where lowerBound is the lower bound of the search after possibly
+ * several iterations.
+ */
+ private int binarySearch(int hash) {
+ int lowerBound = 0;
+ int upperBound = pool.size() - 1;
+ while (lowerBound <= upperBound) {
+ // Fast div by 2. We assume that the number of targets is small enough
+ // that the sum will not overflow an int.
+ int mid = (lowerBound + upperBound) >>> 1;
+ int currentHash = pool.get(mid).hash;
+
+ if (currentHash < hash) {
+ lowerBound = mid + 1;
+ } else if (currentHash > hash) {
+ upperBound = mid - 1;
+ } else {
+ return mid;
+ }
+ }
+ // The +1 ensures that the return value is negative, even when the hash
+ // is off the left edge of the array.
+ return -(lowerBound + 1);
+ }
+
+ /**
+ * Finds the lowest index into the pool ArrayList such that the hash of the i-th entry >= hash.
+ *
+ * @param hash The hash being mapped to a bin via the consistent hash.
+ * @return An int, the lowest index into the target array such that the hash of the i-th entry >= hash.
+ */
+ private int findNearestNodeInPool(int hash) {
+ // Find the index of the node - or at least a near one.
+ // We only search up to targets.length-1. If the element
+ // is greater than the last entry in the list, then map
+ // it to the first one.
+ int nodeIndex = binarySearch(hash);
+
+ // If the returned value is less than zero, then no exact match was found.
+ if (nodeIndex < 0) {
+ // The value returned is -(lowerBound +1), we want the lower bound back.
+ nodeIndex = -(nodeIndex + 1);
+
+ // If hash is greater than the last entry, wrap around to the first.
+ if (nodeIndex >= pool.size()) {
+ nodeIndex = 0;
+ }
+ }
+
+ return nodeIndex;
+ }
+
+ /**
+ * Use the objects built in hash to obtain an initial value, then use a second four byte hash to obtain a more
+ * uniform distribution of hash values. This uses a <a href = "http://burtleburtle.net/bob/hash/integer.html">4-byte
+ * (integer) hash</a>, which produces well distributed values even when the original hash produces thghtly clustered
+ * values.
+ * <p/>
+ * It is important that the object implement its own hashcode, and not use the Object hashcode.
+ *
+ * @param object object to hash
+ * @return an appropriately spread hash code
+ */
+ private int hash(Object object) {
+ int hash = object.hashCode();
+
+ hash = (hash + 0x7ED55D16) + (hash << 12);
+ hash = (hash ^ 0xc761c23c) ^ (hash >> 19);
+ hash = (hash + 0x165667b1) + (hash << 5);
+ hash = (hash + 0xd3a2646c) ^ (hash << 9);
+ hash = (hash + 0xfd7046c5) + (hash << 3);
+ hash = (hash ^ 0xb55a4f09) ^ (hash >> 16);
+
+ return hash;
+ }
+
+ @Override
+ public int getHashId(Address a) {
+ throw new RuntimeException("Not yet implemented");
+ }
+
+ @Override
+ public int getHashSpace() {
+ return Integer.MAX_VALUE; // Entire positive integer range
+ }
+
+ /**
+ * @return A String representing the object pool.
+ */
+ @Override
+ public String toString() {
+ return " pool: " + pool;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null
+ || !(other instanceof ExperimentalDefaultConsistentHash)) {
+ return false;
+ }
+
+ ExperimentalDefaultConsistentHash otherHash = (ExperimentalDefaultConsistentHash) other;
+ return Util.safeEquals(pool, otherHash.pool);
+ }
+
+ @Override
+ public int hashCode() {
+ int hashCode = 1;
+ for (Entry e : pool) hashCode = 31 * hashCode + e.hash;
+ return hashCode;
+ }
+
+ /**
+ * An entry into a consistent hash. It wraps the original object, the object's hash as used to generate the
+ * consistent hash, the value extracted from the object used to generate the hash, and the modifier used to
+ * differentiate the hash.
+ */
+ public static class Entry implements Comparable<Entry> {
+ public final int differentiator;
+ public final int hash;
+ public final Address address;
+ public final String string;
+
+ public Entry(Address address, String string, int differentiator, int hash) {
+ this.differentiator = differentiator;
+ this.hash = hash;
+ this.address = address;
+ this.string = string;
+ }
+
+ /**
+ * Compare this Entry with another Entry. First the hash values are compared, then the differentiator is compared.
+ * if the hash values are equal.
+ *
+ * @param other An Entry object to be compared with this object. Returns <ul> <li>-1 if this Entry is less than
+ * the other Entry.</li> <li>0 if they are equal.</li> <li>+1 if this Entry is greater than the
+ * other Entry.</li> </ul>
+ * @return
+ */
+ public int compareTo(Entry other) {
+ if (this.hash < other.hash) {
+ return -1;
+ }
+
+ if (this.hash > other.hash) {
+ return 1;
+ }
+
+ if (this.differentiator < other.differentiator) {
+ return -1;
+ }
+
+ if (this.differentiator > other.differentiator) {
+ return +1;
+ }
+
+ return 0;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other instanceof Entry) {
+ Entry otherEntry = (Entry) other;
+
+ return hash == otherEntry.hash
+ && differentiator == otherEntry.differentiator
+ && address.equals(otherEntry.address);
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return hash;
+ }
+
+
+ @Override
+ public String toString() {
+ return string + ":" + Integer.toHexString(hash);
+ }
+ }
+
+ public List<Address> getStateProvidersOnLeave(Address leaver, int replCount) {
+ Set<Address> holders = new HashSet<Address>();
+ for (Address address : nodes) {
+ if (isAdjacent(leaver, address)) {
+ holders.add(address);
+ }
+ }
+ return new ArrayList<Address>(holders);
+ }
+
+ public boolean isStateReceiverOnLeave(Address leaver, Address node, int replCount) {
+ throw new RuntimeException("Not implemented!");
+ }
+
+ public List<Address> getStateProvidersOnJoin(Address joiner, int replCount) {
+ throw new RuntimeException("Not implemented!");
+ }
+}
\ No newline at end of file
Property changes on: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/ExperimentalDefaultConsistentHash.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF
Added: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/NodeTopologyInfo.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/NodeTopologyInfo.java (rev 0)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/NodeTopologyInfo.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,48 @@
+package org.infinispan.distribution.ch;
+
+/**
+ * Holds topology information about a a node.
+ *
+ * @author Mircea.Markus at jboss.com
+ * @since 4.2
+ */
+public class NodeTopologyInfo {
+
+ private final String machineId;
+ private final String rackId;
+ private final String siteId;
+
+ public NodeTopologyInfo(String machineId, String rackId, String siteId) {
+ this.machineId = machineId;
+ this.rackId = rackId;
+ this.siteId = siteId;
+ }
+
+ public String getMachineId() {
+ return machineId;
+ }
+
+ public String getRackId() {
+ return rackId;
+ }
+
+ public String getSiteId() {
+ return siteId;
+ }
+
+ public boolean sameSite(NodeTopologyInfo info2) {
+ return equalObjects(siteId, info2.siteId);
+ }
+
+ public boolean sameRack(NodeTopologyInfo info2) {
+ return sameSite(info2) && equalObjects(rackId, info2.rackId);
+ }
+
+ public boolean sameMachine(NodeTopologyInfo info2) {
+ return sameRack(info2) && equalObjects(machineId, info2.machineId);
+ }
+
+ private boolean equalObjects(Object first, Object second) {
+ return first == null ? second == null : first.equals(second);
+ }
+}
Added: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/TopologyAwareConsistentHash.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/TopologyAwareConsistentHash.java (rev 0)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/TopologyAwareConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,122 @@
+package org.infinispan.distribution.ch;
+
+import org.infinispan.remoting.transport.Address;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * Consistent hash that is aware of cluster topology.
+ * Design described here: http://community.jboss.org/wiki/DesigningServerHinting.
+ * <p>
+ * Algorithm:
+ * - place nodes on the hash wheel based address's hash code
+ * - For selecting owner nodes:
+ * - pick the first one based on key's hash code
+ * - for subsequent nodes, walk clockwise and pick nodes that have a different site id
+ * - if not enough nodes found repeat walk again and pick nodes that have different site id and rack id
+ * - if not enough nodes found repeat walk again and pick nodes that have different site id, rack id and machine id
+ * - Ultimately cycle back to the first node selected, don't discard any nodes, regardless of machine id/rack
+ * id/site id match.
+
+ *
+ * @author Mircea.Markus at jboss.com
+ * @since 4.2
+ */
+public class TopologyAwareConsistentHash extends AbstractWheelConsistentHash {
+
+ public List<Address> locate(Object key, int replCount) {
+ Address owner = getOwner(key);
+ int ownerCount = min(replCount, addresses.size());
+ return getOwners(owner, ownerCount);
+ }
+
+ public List<Address> getStateProvidersOnLeave(Address leaver, int replCount) {
+ Set<Address> result = new HashSet<Address>();
+
+ //1. first get all the node that replicated on leaver
+ for (Address address : addresses) {
+ if (address.equals(leaver)) continue;
+ if (getOwners(address, replCount).contains(leaver)) {
+ result.add(address);
+ }
+ }
+
+ //2. then get first leaver's backup
+ List<Address> addressList = getOwners(leaver, replCount);
+ if (addressList.size() > 1) {
+ result.add(addressList.get(1));
+ }
+ return new ArrayList<Address>(result);
+ }
+
+
+ /**
+ * In this situation are the same nodes providing state on join as the nodes that provide state on leave.
+ */
+ public List<Address> getStateProvidersOnJoin(Address joiner, int replCount) {
+ return getStateProvidersOnLeave(joiner, replCount);
+ }
+
+ private List<Address> getOwners(Address address, int numOwners) {
+ int ownerHash = getNormalizedHash(address);
+ Collection<Address> beforeOnWheel = positions.headMap(ownerHash).values();
+ Collection<Address> afterOnWheel = positions.tailMap(ownerHash).values();
+ ArrayList<Address> processSequence = new ArrayList<Address>(afterOnWheel);
+ processSequence.addAll(beforeOnWheel);
+ List<Address> result = new ArrayList<Address>();
+ result.add(processSequence.remove(0));
+ int level = 0;
+ while (result.size() < numOwners) {
+ Iterator<Address> addrIt = processSequence.iterator();
+ while (addrIt.hasNext()) {
+ Address a = addrIt.next();
+ switch (level) {
+ case 0 : { //site level
+ if (!topologyInfo.isSameSite(address, a)) {
+ result.add(a);
+ addrIt.remove();
+ }
+ break;
+ }
+ case 1 : { //rack level
+ if (!topologyInfo.isSameRack(address, a)) {
+ result.add(a);
+ addrIt.remove();
+ }
+ break;
+ }
+ case 2 : { //machine level
+ if (!topologyInfo.isSameMachine(address, a)) {
+ result.add(a);
+ addrIt.remove();
+ }
+ break;
+ }
+ case 3 : { //just add them in sequence
+ result.add(a);
+ addrIt.remove();
+ break;
+ }
+ }
+ if (result.size() == numOwners) break;
+ }
+ level++;
+ }
+ //assertion
+ if (result.size() != numOwners) throw new AssertionError("This should not happen!");
+ return result;
+ }
+
+ private Address getOwner(Object key) {
+ int hash = getNormalizedHash(key);
+ Integer ownerHash = positions.tailMap(hash).firstKey();
+ return positions.get(ownerHash);
+ }
+}
Added: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/TopologyInfo.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/TopologyInfo.java (rev 0)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/TopologyInfo.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,39 @@
+package org.infinispan.distribution.ch;
+
+import org.infinispan.remoting.transport.Address;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Aggregates topology information from all nodes within the cluster.
+ *
+ * @author Mircea.Markus at jboss.com
+ * @since 4.2
+ */
+public class TopologyInfo {
+
+ private Map<Address, NodeTopologyInfo> address2TopologyInfo = new HashMap<Address, NodeTopologyInfo>();
+
+ public void addNodeTopologyInfo(Address addr, NodeTopologyInfo ti) {
+ address2TopologyInfo.put(addr, ti);
+ }
+
+ public boolean isSameSite(Address a1, Address a2) {
+ NodeTopologyInfo info1 = address2TopologyInfo.get(a1);
+ NodeTopologyInfo info2 = address2TopologyInfo.get(a2);
+ return info1.sameSite(info2);
+ }
+
+ public boolean isSameRack(Address a1, Address a2) {
+ NodeTopologyInfo info1 = address2TopologyInfo.get(a1);
+ NodeTopologyInfo info2 = address2TopologyInfo.get(a2);
+ return info1.sameRack(info2);
+ }
+
+ public boolean isSameMachine(Address a1, Address a2) {
+ NodeTopologyInfo info1 = address2TopologyInfo.get(a1);
+ NodeTopologyInfo info2 = address2TopologyInfo.get(a2);
+ return info1.sameMachine(info2);
+ }
+}
Copied: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/UnionConsistentHash.java (from rev 2610, branches/4.2.x/core/src/main/java/org/infinispan/distribution/UnionConsistentHash.java)
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/UnionConsistentHash.java (rev 0)
+++ branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/UnionConsistentHash.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,104 @@
+package org.infinispan.distribution.ch;
+
+import org.infinispan.CacheException;
+import org.infinispan.marshall.Ids;
+import org.infinispan.marshall.Marshallable;
+import org.infinispan.remoting.transport.Address;
+import org.infinispan.util.Immutables;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * A delegating wrapper that locates keys by getting a union of locations reported by two other ConsistentHash
+ * implementations it delegates to.
+ *
+ * @author Manik Surtani
+ * @since 4.0
+ */
+ at Marshallable(externalizer = UnionConsistentHash.Externalizer.class, id = Ids.UNION_CONSISTENT_HASH)
+public class UnionConsistentHash extends AbstractConsistentHash {
+
+ ConsistentHash oldCH, newCH;
+
+ public UnionConsistentHash(ConsistentHash oldCH, ConsistentHash newCH) {
+ if ((oldCH instanceof UnionConsistentHash) || (newCH instanceof UnionConsistentHash))
+ throw new CacheException("Expecting both newCH and oldCH to not be Unions!! oldCH=[" + oldCH.getClass() + "] and newCH=[" + newCH.getClass() + "]");
+ this.oldCH = oldCH;
+ this.newCH = newCH;
+ }
+
+ public void setCaches(List<Address> caches) {
+ // no op
+ }
+
+ public List<Address> getCaches() {
+ return Collections.emptyList();
+ }
+
+ public List<Address> locate(Object key, int replCount) {
+ Set<Address> addresses = new LinkedHashSet<Address>();
+ addresses.addAll(oldCH.locate(key, replCount));
+ addresses.addAll(newCH.locate(key, replCount));
+ return Immutables.immutableListConvert(addresses);
+ }
+
+ @Override
+ public int getHashId(Address a) {
+ throw new UnsupportedOperationException("Unsupported!");
+ }
+
+ public List<Address> getStateProvidersOnLeave(Address leaver, int replCount) {
+ throw new UnsupportedOperationException("Unsupported!");
+ }
+
+ public boolean isStateReceiverOnLeave(Address leaver, Address node, int replCount) {
+ throw new UnsupportedOperationException("Unsupported!");
+ }
+
+ public List<Address> getStateProvidersOnJoin(Address joiner, int replCount) {
+ throw new UnsupportedOperationException("Unsupported!");
+ }
+
+ @Override
+ public int getHashSpace() {
+ int oldHashSpace = oldCH.getHashSpace();
+ int newHashSpace = newCH.getHashSpace();
+ // In a union, the hash space is the biggest of the hash spaces.
+ return oldHashSpace > newHashSpace ? oldHashSpace : newHashSpace;
+ }
+
+ public ConsistentHash getNewConsistentHash() {
+ return newCH;
+ }
+
+ public ConsistentHash getOldConsistentHash() {
+ return oldCH;
+ }
+
+ public static class Externalizer implements org.infinispan.marshall.Externalizer {
+
+ public void writeObject(ObjectOutput output, Object object) throws IOException {
+ UnionConsistentHash uch = (UnionConsistentHash) object;
+ output.writeObject(uch.oldCH);
+ output.writeObject(uch.newCH);
+ }
+
+ public Object readObject(ObjectInput input) throws IOException, ClassNotFoundException {
+ return new UnionConsistentHash((ConsistentHash) input.readObject(), (ConsistentHash) input.readObject());
+ }
+ }
+
+ public ConsistentHash getOldCH() {
+ return oldCH;
+ }
+
+ public ConsistentHash getNewCH() {
+ return newCH;
+ }
+}
Property changes on: branches/4.2.x/core/src/main/java/org/infinispan/distribution/ch/UnionConsistentHash.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF
Modified: branches/4.2.x/core/src/main/java/org/infinispan/marshall/jboss/ConstantObjectTable.java
===================================================================
--- branches/4.2.x/core/src/main/java/org/infinispan/marshall/jboss/ConstantObjectTable.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/main/java/org/infinispan/marshall/jboss/ConstantObjectTable.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -53,8 +53,8 @@
import org.infinispan.container.entries.TransientCacheValue;
import org.infinispan.container.entries.TransientMortalCacheEntry;
import org.infinispan.container.entries.TransientMortalCacheValue;
-import org.infinispan.distribution.DefaultConsistentHash;
-import org.infinispan.distribution.UnionConsistentHash;
+import org.infinispan.distribution.ch.DefaultConsistentHash;
+import org.infinispan.distribution.ch.UnionConsistentHash;
import org.infinispan.loaders.bucket.Bucket;
import org.infinispan.marshall.Externalizer;
import org.infinispan.marshall.Marshallable;
@@ -84,7 +84,6 @@
import org.jboss.marshalling.Marshaller;
import org.jboss.marshalling.ObjectTable;
import org.jboss.marshalling.Unmarshaller;
-import org.jboss.marshalling.river.RiverUnmarshaller;
import java.io.IOException;
import java.util.ArrayList;
Modified: branches/4.2.x/core/src/test/java/org/infinispan/affinity/BaseKeyAffinityServiceTest.java
===================================================================
--- branches/4.2.x/core/src/test/java/org/infinispan/affinity/BaseKeyAffinityServiceTest.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/test/java/org/infinispan/affinity/BaseKeyAffinityServiceTest.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -1,7 +1,7 @@
package org.infinispan.affinity;
import org.infinispan.distribution.BaseDistFunctionalTest;
-import org.infinispan.distribution.ConsistentHash;
+import org.infinispan.distribution.ch.ConsistentHash;
import org.infinispan.manager.CacheContainer;
import org.infinispan.remoting.transport.Address;
import org.infinispan.test.TestingUtil;
Modified: branches/4.2.x/core/src/test/java/org/infinispan/affinity/KeyAffinityServiceTest.java
===================================================================
--- branches/4.2.x/core/src/test/java/org/infinispan/affinity/KeyAffinityServiceTest.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/test/java/org/infinispan/affinity/KeyAffinityServiceTest.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -1,7 +1,7 @@
package org.infinispan.affinity;
import org.infinispan.Cache;
-import org.infinispan.distribution.ConsistentHash;
+import org.infinispan.distribution.ch.ConsistentHash;
import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.remoting.transport.Address;
import org.testng.annotations.Test;
Modified: branches/4.2.x/core/src/test/java/org/infinispan/config/parsing/XmlFileParsingTest.java
===================================================================
--- branches/4.2.x/core/src/test/java/org/infinispan/config/parsing/XmlFileParsingTest.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/test/java/org/infinispan/config/parsing/XmlFileParsingTest.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -1,6 +1,5 @@
package org.infinispan.config.parsing;
-import org.infinispan.Cache;
import org.infinispan.Version;
import org.infinispan.config.CacheLoaderManagerConfig;
import org.infinispan.config.Configuration;
@@ -8,20 +7,17 @@
import org.infinispan.config.GlobalConfiguration;
import org.infinispan.config.GlobalConfiguration.ShutdownHookBehavior;
import org.infinispan.config.InfinispanConfiguration;
-import org.infinispan.distribution.DefaultConsistentHash;
+import org.infinispan.distribution.ch.DefaultConsistentHash;
import org.infinispan.eviction.EvictionStrategy;
import org.infinispan.eviction.EvictionThreadPolicy;
import org.infinispan.loaders.file.FileCacheStoreConfig;
import org.infinispan.remoting.transport.jgroups.JGroupsTransport;
import org.infinispan.test.AbstractInfinispanTest;
import org.infinispan.test.TestingUtil;
-import org.infinispan.test.fwk.TestCacheManagerFactory;
import org.infinispan.util.concurrent.IsolationLevel;
import org.testng.annotations.Test;
import java.io.ByteArrayInputStream;
-import java.io.File;
-import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
Modified: branches/4.2.x/core/src/test/java/org/infinispan/distribution/BaseDistFunctionalTest.java
===================================================================
--- branches/4.2.x/core/src/test/java/org/infinispan/distribution/BaseDistFunctionalTest.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/test/java/org/infinispan/distribution/BaseDistFunctionalTest.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -8,6 +8,10 @@
import org.infinispan.container.entries.ImmortalCacheEntry;
import org.infinispan.container.entries.InternalCacheEntry;
import org.infinispan.container.entries.MortalCacheEntry;
+import org.infinispan.distribution.ch.ConsistentHash;
+import org.infinispan.distribution.ch.ConsistentHashHelper;
+import org.infinispan.distribution.ch.DefaultConsistentHash;
+import org.infinispan.distribution.ch.UnionConsistentHash;
import org.infinispan.manager.CacheContainer;
import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.remoting.transport.Address;
Modified: branches/4.2.x/core/src/test/java/org/infinispan/distribution/ConsistentHashPerfTest.java
===================================================================
--- branches/4.2.x/core/src/test/java/org/infinispan/distribution/ConsistentHashPerfTest.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/test/java/org/infinispan/distribution/ConsistentHashPerfTest.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -1,5 +1,6 @@
package org.infinispan.distribution;
+import org.infinispan.distribution.ch.ConsistentHash;
import org.infinispan.remoting.transport.Address;
import org.infinispan.remoting.transport.jgroups.JGroupsAddress;
import org.infinispan.test.TestingUtil;
Modified: branches/4.2.x/core/src/test/java/org/infinispan/distribution/DefaultConsistentHashTest.java
===================================================================
--- branches/4.2.x/core/src/test/java/org/infinispan/distribution/DefaultConsistentHashTest.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/test/java/org/infinispan/distribution/DefaultConsistentHashTest.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -1,5 +1,6 @@
package org.infinispan.distribution;
+import org.infinispan.distribution.ch.DefaultConsistentHash;
import org.infinispan.remoting.transport.Address;
import org.infinispan.test.AbstractInfinispanTest;
import org.testng.annotations.AfterTest;
@@ -17,7 +18,7 @@
public class DefaultConsistentHashTest extends AbstractInfinispanTest {
List<Address> servers;
- ConsistentHash ch;
+ DefaultConsistentHash ch;
@BeforeTest
public void setUp() {
@@ -27,7 +28,7 @@
servers.add(new TestAddress(i));
}
- ch = BaseDistFunctionalTest.createNewConsistentHash(servers);
+ ch = (DefaultConsistentHash) BaseDistFunctionalTest.createNewConsistentHash(servers);
}
@AfterTest
@@ -82,7 +83,7 @@
Address a3 = new TestAddress(3000);
Address a4 = new TestAddress(4000);
- ch = BaseDistFunctionalTest.createNewConsistentHash(Arrays.asList(a1, a2, a3, a4));
+ ch = (DefaultConsistentHash) BaseDistFunctionalTest.createNewConsistentHash(Arrays.asList(a1, a2, a3, a4));
// the CH may reorder the addresses. Get the new order.
List<Address> adds = ch.getCaches();
@@ -109,7 +110,7 @@
Address a3 = new TestAddress(3000);
Address a4 = new TestAddress(4000);
- ch = BaseDistFunctionalTest.createNewConsistentHash(Arrays.asList(a1, a2, a3, a4));
+ ch = (DefaultConsistentHash) BaseDistFunctionalTest.createNewConsistentHash(Arrays.asList(a1, a2, a3, a4));
String[] keys = new String[10000];
Random r = new Random();
@@ -126,6 +127,12 @@
class TestAddress implements Address {
int addressNum;
+ String name;
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
TestAddress(int addressNum) {
this.addressNum = addressNum;
}
@@ -157,7 +164,7 @@
@Override
public String toString() {
- return "TestAddress#"+addressNum;
+ return "TestAddress#"+addressNum + (name != null? (" " + name) : "");
}
public int compareTo(Object o) {
Added: branches/4.2.x/core/src/test/java/org/infinispan/distribution/DistributionManagerImplTest.java
===================================================================
--- branches/4.2.x/core/src/test/java/org/infinispan/distribution/DistributionManagerImplTest.java (rev 0)
+++ branches/4.2.x/core/src/test/java/org/infinispan/distribution/DistributionManagerImplTest.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,85 @@
+package org.infinispan.distribution;
+
+import org.infinispan.config.Configuration;
+import org.infinispan.distribution.ch.DefaultConsistentHash;
+import org.infinispan.remoting.transport.Address;
+import org.infinispan.test.AbstractInfinispanTest;
+import org.testng.annotations.BeforeTest;
+import org.testng.annotations.Test;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * @author Mircea.Markus at jboss.com
+ * @since 4.2
+ */
+ at Test (groups = "functional", testName = "distribution.DistributionManagerImplTest")
+public class DistributionManagerImplTest extends AbstractInfinispanTest {
+ List<Address> servers;
+ DefaultConsistentHash ch;
+ Address a0;
+ Address a1;
+ Address a2;
+ Address a3;
+ Address a4;
+
+ @BeforeTest
+ public void setUp() {
+ servers = new LinkedList<Address>();
+ int numServers = 5;
+ for (int i = 0; i < numServers; i++) {
+ servers.add(new TestAddress(i));
+ }
+ ch = (DefaultConsistentHash) BaseDistFunctionalTest.createNewConsistentHash(servers);
+ a0 = ch.getAddressOnTheWheel().get(0);
+ a1 = ch.getAddressOnTheWheel().get(1);
+ a2 = ch.getAddressOnTheWheel().get(2);
+ a3 = ch.getAddressOnTheWheel().get(3);
+ a4 = ch.getAddressOnTheWheel().get(4);
+ }
+
+
+ /**
+ * numOwners = 3. Let's a a2 leaves.
+ */
+ public void testLeaver() {
+ DistributionManagerImpl dm = newDM(3);
+ dm.setSelf(a0);
+ assert dm.willReceiveLeaverState(a2) : " a0 will be the 2nd backup for a3's state";
+ dm.setSelf(a1);
+ assert !dm.willReceiveLeaverState(a2) : " a1 is not affected";
+ dm.setSelf(a3);
+ assert dm.willReceiveLeaverState(a2) : "this needs to receive state from a0";
+ dm.setSelf(a4);
+ assert dm.willReceiveLeaverState(a2) : "this needs to receive state from a1";
+ }
+
+ private DistributionManagerImpl newDM(int numOwners) {
+ DistributionManagerImpl dm = new DistributionManagerImpl();
+ Configuration configuration = new Configuration();
+ configuration.setNumOwners(numOwners);
+ dm.setConfiguration(configuration);
+ dm.setOldConsistentHash(ch);
+ return dm;
+ }
+
+
+ public void testHoldersOfLeaversState() {
+ DistributionManagerImpl dm = newDM(3);
+ dm.setSelf(a0);
+ List<Address> addressList = dm.holdersOfLeaversState(a2);
+ assert addressList.size() == 2;
+ assert addressList.contains(a1);
+ assert addressList.contains(a3);
+ }
+
+ public void testHoldersOfLeaversState2() {
+ DistributionManagerImpl dm = newDM(2);
+ dm.setSelf(a0);
+ List<Address> addressList = dm.holdersOfLeaversState(a1);
+ assert addressList.size() == 2;
+ assert addressList.contains(a0);
+ assert addressList.contains(a2);
+ }
+}
Added: branches/4.2.x/core/src/test/java/org/infinispan/distribution/TopologyAwareConsistentHashTest.java
===================================================================
--- branches/4.2.x/core/src/test/java/org/infinispan/distribution/TopologyAwareConsistentHashTest.java (rev 0)
+++ branches/4.2.x/core/src/test/java/org/infinispan/distribution/TopologyAwareConsistentHashTest.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -0,0 +1,528 @@
+package org.infinispan.distribution;
+
+import org.infinispan.distribution.ch.NodeTopologyInfo;
+import org.infinispan.distribution.ch.TopologyAwareConsistentHash;
+import org.infinispan.distribution.ch.TopologyInfo;
+
+import org.infinispan.remoting.transport.Address;
+import org.infinispan.test.TestingUtil;
+import org.infinispan.util.logging.Log;
+import org.infinispan.util.logging.LogFactory;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.testng.Assert.assertEquals;
+
+/**
+ * @author Mircea.Markus at jboss.com
+ * @since 4.2
+ */
+ at Test(groups = "functional", testName = "distribution.TopologyAwareConsistentHashTest")
+public class TopologyAwareConsistentHashTest {
+
+ private static Log log = LogFactory.getLog(TopologyAwareConsistentHashTest.class);
+
+ TopologyInfo ti;
+ TopologyAwareConsistentHash ch;
+ ArrayList<Address> addresses;
+ TestAddress a0;
+ TestAddress a1;
+ TestAddress a2;
+ TestAddress a3;
+ TestAddress a4;
+ TestAddress a5;
+ TestAddress a6;
+ TestAddress a7;
+ TestAddress a8;
+ TestAddress a9;
+ private TestAddress[] staticAddresses;
+
+
+ @BeforeMethod
+ public void setUp() {
+ ti = new TopologyInfo();
+ ch = new TopologyAwareConsistentHash();
+ addresses = new ArrayList<Address>();
+ for (int i = 0; i < 10; i++) {
+ addresses.add(new TestAddress(i * 100));
+ }
+ ch.setCaches(addresses);
+ addresses = new ArrayList(ch.getCaches());
+ for (int i = 0; i < addresses.size(); i++) {
+ TestingUtil.replaceField(addresses.get(i), "a"+i, this, TopologyAwareConsistentHashTest.class);
+ }
+
+ addresses.clear();
+ ch = new TopologyAwareConsistentHash();
+ ch.setTopologyInfo(ti);
+ }
+
+ public void testIsStateReceiverOnLeave() {
+ addNode(a0, "m0", null, null);
+ addNode(a1, "m1", null, null);
+ addNode(a2, "m0", null, null);
+ addNode(a3, "m1", null, null);
+ setAddresses();
+
+ assert !ch.isStateReceiverOnLeave(a0, a1, 2);
+ assert ch.isStateReceiverOnLeave(a0, a2, 2);
+ assert !ch.isStateReceiverOnLeave(a0, a1, 2);
+ }
+
+ public void testDifferentMachines() {
+ addNode(a0, "m0", null, null);
+ addNode(a1, "m1", null, null);
+ addNode(a2, "m0", null, null);
+ addNode(a3, "m1", null, null);
+ setAddresses();
+
+ assertLocation(ch.locate(a0, 1), true, a0);
+ assertLocation(ch.locate(a1, 1), true, a1);
+ assertLocation(ch.locate(a2, 1), true, a2);
+ assertLocation(ch.locate(a3, 1), true, a3);
+
+ assertLocation(ch.getStateProvidersOnLeave(a0, 1), false);
+ assertLocation(ch.getStateProvidersOnLeave(a1, 1), false);
+ assertLocation(ch.getStateProvidersOnLeave(a2, 1), false);
+ assertLocation(ch.getStateProvidersOnLeave(a3, 1), false);
+
+ assertLocation(ch.locate(a0, 2), true, a0, a1);
+ assertLocation(ch.locate(a1, 2), true, a1, a2);
+ assertLocation(ch.locate(a2, 2), true, a2, a3);
+ assertLocation(ch.locate(a3, 2), true, a3, a0);
+
+ assertLocation(ch.getStateProvidersOnLeave(a0, 2), false, a1, a3);
+ assertLocation(ch.getStateProvidersOnLeave(a1, 2), false, a0, a2);
+ assertLocation(ch.getStateProvidersOnLeave(a2, 2), false, a3, a1);
+ assertLocation(ch.getStateProvidersOnLeave(a3, 2), false, a0, a2);
+
+
+
+ assertLocation(ch.locate(a0, 3), true, a0, a1, a3);
+ assertLocation(ch.locate(a1, 3), true, a1, a2, a0);
+ assertLocation(ch.locate(a2, 3), true, a2, a3, a1);
+ assertLocation(ch.locate(a3, 3), true, a3, a0, a2);
+
+ assertLocation(ch.getStateProvidersOnLeave(a0, 3), false, a1, a3);
+ assertLocation(ch.getStateProvidersOnLeave(a1, 3), false, a0, a2);
+ assertLocation(ch.getStateProvidersOnLeave(a2, 3), false, a3, a1);
+ assertLocation(ch.getStateProvidersOnLeave(a3, 3), false, a0, a2);
+ }
+
+ public void testDifferentMachines2() {
+ addNode(a0, "m0", null, null);
+ addNode(a1, "m0", null, null);
+ addNode(a2, "m1", null, null);
+ addNode(a3, "m1", null, null);
+ addNode(a4, "m2", null, null);
+ addNode(a5, "m2", null, null);
+ setAddresses();
+ assertLocation(ch.locate(a0, 1), true, a0);
+ assertLocation(ch.locate(a1, 1), true, a1);
+ assertLocation(ch.locate(a2, 1), true, a2);
+ assertLocation(ch.locate(a3, 1), true, a3);
+ assertLocation(ch.locate(a4, 1), true, a4);
+ assertLocation(ch.locate(a5, 1), true, a5);
+
+ assertLocation(ch.locate(a0, 2), true, a0, a2);
+ assertLocation(ch.locate(a1, 2), true, a1, a2);
+ assertLocation(ch.locate(a2, 2), true, a2, a4);
+ assertLocation(ch.locate(a3, 2), true, a3, a4);
+ assertLocation(ch.locate(a4, 2), true, a4, a0);
+ assertLocation(ch.locate(a5, 2), true, a5, a0);
+
+ assertLocation(ch.locate(a0, 3), true, a0, a2, a3);
+ assertLocation(ch.locate(a1, 3), true, a1, a2, a3);
+ assertLocation(ch.locate(a2, 3), true, a2, a4, a5);
+ assertLocation(ch.locate(a3, 3), true, a3, a4, a5);
+ assertLocation(ch.locate(a4, 3), true, a4, a0, a1);
+ assertLocation(ch.locate(a5, 3), true, a5, a0, a1);
+ }
+
+ public void testDifferentRacksAndMachines() {
+ addNode(a0, "m0", "r0", null);
+ addNode(a1, "m0", "r0", null);
+ addNode(a2, "m1", "r1", null);
+ addNode(a3, "m2", "r2", null);
+ addNode(a4, "m1", "r1", null);
+ addNode(a5, "m2", "r3", null);
+ setAddresses();
+ assertLocation(ch.locate(a0, 1), true, a0);
+ assertLocation(ch.locate(a1, 1), true, a1);
+ assertLocation(ch.locate(a2, 1), true, a2);
+ assertLocation(ch.locate(a3, 1), true, a3);
+ assertLocation(ch.locate(a4, 1), true, a4);
+ assertLocation(ch.locate(a5, 1), true, a5);
+
+ assertLocation(ch.locate(a0, 2), true, a0, a2);
+ assertLocation(ch.locate(a1, 2), true, a1, a2);
+ assertLocation(ch.locate(a2, 2), true, a2, a3);
+ assertLocation(ch.locate(a3, 2), true, a3, a4);
+ assertLocation(ch.locate(a4, 2), true, a4, a5);
+ assertLocation(ch.locate(a5, 2), true, a5, a0);
+
+ assertLocation(ch.locate(a0, 3), true, a0, a2, a3);
+ assertLocation(ch.locate(a1, 3), true, a1, a2, a3);
+ assertLocation(ch.locate(a2, 3), true, a2, a3, a5);
+ assertLocation(ch.locate(a3, 3), true, a3, a4, a5);
+ assertLocation(ch.locate(a4, 3), true, a4, a5, a0);
+ assertLocation(ch.locate(a5, 3), true, a5, a0, a1);
+ }
+
+ public void testAllSameMachine() {
+ addNode(a0, "m0", null, null);
+ addNode(a1, "m0", null, null);
+ addNode(a2, "m0", null, null);
+ addNode(a3, "m0", null, null);
+ addNode(a4, "m0", null, null);
+ addNode(a5, "m0", null, null);
+ setAddresses();
+
+ assertLocation(ch.locate(a0, 1), true, a0);
+ assertLocation(ch.locate(a1, 1), true, a1);
+ assertLocation(ch.locate(a2, 1), true, a2);
+ assertLocation(ch.locate(a3, 1), true, a3);
+ assertLocation(ch.locate(a4, 1), true, a4);
+ assertLocation(ch.locate(a5, 1), true, a5);
+
+ assertLocation(ch.locate(a0, 2), true, a0, a1);
+ assertLocation(ch.locate(a1, 2), true, a1, a2);
+ assertLocation(ch.locate(a2, 2), true, a2, a3);
+ assertLocation(ch.locate(a3, 2), true, a3, a4);
+ assertLocation(ch.locate(a4, 2), true, a4, a5);
+ assertLocation(ch.locate(a5, 2), true, a5, a0);
+
+ assertLocation(ch.locate(a0, 3), true, a0, a1, a2);
+ assertLocation(ch.locate(a1, 3), true, a1, a2, a3);
+ assertLocation(ch.locate(a2, 3), true, a2, a3, a4);
+ assertLocation(ch.locate(a3, 3), true, a3, a4, a5);
+ assertLocation(ch.locate(a4, 3), true, a4, a5, a0);
+ assertLocation(ch.locate(a5, 3), true, a5, a0, a1);
+ }
+
+ public void testDifferentSites() {
+ addNode(a0, "m0", null, "s0");
+ addNode(a1, "m1", null, "s0");
+ addNode(a2, "m2", null, "s1");
+ addNode(a3, "m3", null, "s1");
+ setAddresses();
+ assertLocation(ch.locate(a0, 1), true, a0);
+ assertLocation(ch.locate(a1, 1), true, a1);
+ assertLocation(ch.locate(a2, 1), true, a2);
+ assertLocation(ch.locate(a3, 1), true, a3);
+
+ assertLocation(ch.locate(a0, 2), true, a0, a2);
+ assertLocation(ch.locate(a1, 2), true, a1, a2);
+ assertLocation(ch.locate(a2, 2), true, a2, a0);
+ assertLocation(ch.locate(a3, 2), true, a3, a0);
+
+ assertLocation(ch.locate(a0, 3), true, a0, a2, a3);
+ assertLocation(ch.locate(a1, 3), true, a1, a2, a3);
+ assertLocation(ch.locate(a2, 3), true, a2, a0, a1);
+ assertLocation(ch.locate(a3, 3), true, a3, a0, a1);
+ }
+
+ public void testSitesMachines2() {
+ addNode(a0, "m0", null, "s0");
+ addNode(a1, "m1", null, "s1");
+ addNode(a2, "m2", null, "s0");
+ addNode(a3, "m3", null, "s2");
+ addNode(a4, "m4", null, "s1");
+ addNode(a5, "m5", null, "s1");
+
+ setAddresses();
+ assertLocation(ch.locate(a0, 1), true, a0);
+ assertLocation(ch.locate(a1, 1), true, a1);
+ assertLocation(ch.locate(a2, 1), true, a2);
+ assertLocation(ch.locate(a3, 1), true, a3);
+ assertLocation(ch.locate(a4, 1), true, a4);
+ assertLocation(ch.locate(a5, 1), true, a5);
+
+ assertLocation(ch.locate(a0, 2), true, a0, a1);
+ assertLocation(ch.locate(a1, 2), true, a1, a2);
+ assertLocation(ch.locate(a2, 2), true, a2, a3);
+ assertLocation(ch.locate(a3, 2), true, a3, a4);
+ assertLocation(ch.locate(a4, 2), true, a4, a0);
+ assertLocation(ch.locate(a5, 2), true, a5, a0);
+
+ assertLocation(ch.locate(a0, 3), true, a0, a1, a3);
+ assertLocation(ch.locate(a1, 3), true, a1, a2, a3);
+ assertLocation(ch.locate(a2, 3), true, a2, a3, a4);
+ assertLocation(ch.locate(a3, 3), true, a3, a4, a5);
+ assertLocation(ch.locate(a4, 3), true, a4, a0, a2);
+ assertLocation(ch.locate(a5, 3), true, a5, a0, a2);
+ }
+
+ public void testSitesMachinesSameMachineName() {
+ addNode(a0, "m0", null, "r0");
+ addNode(a1, "m0", null, "r1");
+ addNode(a2, "m0", null, "r0");
+ addNode(a3, "m0", null, "r2");
+ addNode(a4, "m0", null, "r1");
+ addNode(a5, "m0", null, "r1");
+
+ setAddresses();
+ assertLocation(ch.locate(a0, 1), true, a0);
+ assertLocation(ch.locate(a1, 1), true, a1);
+ assertLocation(ch.locate(a2, 1), true, a2);
+ assertLocation(ch.locate(a3, 1), true, a3);
+ assertLocation(ch.locate(a4, 1), true, a4);
+ assertLocation(ch.locate(a5, 1), true, a5);
+
+ assertLocation(ch.locate(a0, 2), true, a0, a1);
+ assertLocation(ch.locate(a1, 2), true, a1, a2);
+ assertLocation(ch.locate(a2, 2), true, a2, a3);
+ assertLocation(ch.locate(a3, 2), true, a3, a4);
+ assertLocation(ch.locate(a4, 2), true, a4, a0);
+ assertLocation(ch.locate(a5, 2), true, a5, a0);
+
+ assertLocation(ch.locate(a0, 3), true, a0, a1, a3);
+ assertLocation(ch.locate(a1, 3), true, a1, a2, a3);
+ assertLocation(ch.locate(a2, 3), true, a2, a3, a4);
+ assertLocation(ch.locate(a3, 3), true, a3, a4, a5);
+ assertLocation(ch.locate(a4, 3), true, a4, a0, a2);
+ assertLocation(ch.locate(a5, 3), true, a5, a0, a2);
+ }
+
+ public void testDifferentRacks() {
+ addNode(a0, "m0", "r0", null);
+ addNode(a1, "m1", "r0", null);
+ addNode(a2, "m2", "r1", null);
+ addNode(a3, "m3", "r1", null);
+ setAddresses();
+ assertLocation(ch.locate(a0, 1), true, a0);
+ assertLocation(ch.locate(a1, 1), true, a1);
+ assertLocation(ch.locate(a2, 1), true, a2);
+ assertLocation(ch.locate(a3, 1), true, a3);
+
+ assertLocation(ch.locate(a0, 2), true, a0, a2);
+ assertLocation(ch.locate(a1, 2), true, a1, a2);
+ assertLocation(ch.locate(a2, 2), true, a2, a0);
+ assertLocation(ch.locate(a3, 2), true, a3, a0);
+
+ assertLocation(ch.locate(a0, 3), true, a0, a2, a3);
+ assertLocation(ch.locate(a1, 3), true, a1, a2, a3);
+ assertLocation(ch.locate(a2, 3), true, a2, a0, a1);
+ assertLocation(ch.locate(a3, 3), true, a3, a0, a1);
+ }
+
+ public void testRacksMachines2() {
+ addNode(a0, "m0", "r0", null);
+ addNode(a1, "m1", "r1", null);
+ addNode(a2, "m2", "r0", null);
+ addNode(a3, "m3", "r2", null);
+ addNode(a4, "m4", "r1", null);
+ addNode(a5, "m5", "r1", null);
+
+ setAddresses();
+ assertLocation(ch.locate(a0, 1), true, a0);
+ assertLocation(ch.locate(a1, 1), true, a1);
+ assertLocation(ch.locate(a2, 1), true, a2);
+ assertLocation(ch.locate(a3, 1), true, a3);
+ assertLocation(ch.locate(a4, 1), true, a4);
+ assertLocation(ch.locate(a5, 1), true, a5);
+
+ assertLocation(ch.locate(a0, 2), true, a0, a1);
+ assertLocation(ch.locate(a1, 2), true, a1, a2);
+ assertLocation(ch.locate(a2, 2), true, a2, a3);
+ assertLocation(ch.locate(a3, 2), true, a3, a4);
+ assertLocation(ch.locate(a4, 2), true, a4, a0);
+ assertLocation(ch.locate(a5, 2), true, a5, a0);
+
+ assertLocation(ch.locate(a0, 3), true, a0, a1, a3);
+ assertLocation(ch.locate(a1, 3), true, a1, a2, a3);
+ assertLocation(ch.locate(a2, 3), true, a2, a3, a4);
+ assertLocation(ch.locate(a3, 3), true, a3, a4, a5);
+ assertLocation(ch.locate(a4, 3), true, a4, a0, a2);
+ assertLocation(ch.locate(a5, 3), true, a5, a0, a2);
+ }
+
+ public void testRacksMachinesSameMachineName() {
+ addNode(a0, "m0", "r0", null);
+ addNode(a1, "m0", "r1", null);
+ addNode(a2, "m0", "r0", null);
+ addNode(a3, "m0", "r2", null);
+ addNode(a4, "m0", "r1", null);
+ addNode(a5, "m0", "r1", null);
+
+ setAddresses();
+ assertLocation(ch.locate(a0, 1), true, a0);
+ assertLocation(ch.locate(a1, 1), true, a1);
+ assertLocation(ch.locate(a2, 1), true, a2);
+ assertLocation(ch.locate(a3, 1), true, a3);
+ assertLocation(ch.locate(a4, 1), true, a4);
+ assertLocation(ch.locate(a5, 1), true, a5);
+
+ assertLocation(ch.locate(a0, 2), true, a0, a1);
+ assertLocation(ch.locate(a1, 2), true, a1, a2);
+ assertLocation(ch.locate(a2, 2), true, a2, a3);
+ assertLocation(ch.locate(a3, 2), true, a3, a4);
+ assertLocation(ch.locate(a4, 2), true, a4, a0);
+ assertLocation(ch.locate(a5, 2), true, a5, a0);
+
+ assertLocation(ch.locate(a0, 3), true, a0, a1, a3);
+ assertLocation(ch.locate(a1, 3), true, a1, a2, a3);
+ assertLocation(ch.locate(a2, 3), true, a2, a3, a4);
+ assertLocation(ch.locate(a3, 3), true, a3, a4, a5);
+ assertLocation(ch.locate(a4, 3), true, a4, a0, a2);
+ assertLocation(ch.locate(a5, 3), true, a5, a0, a2);
+ }
+
+ public void testComplexScenario() {
+ addNode(a0, "m2", "r0", "s1");
+ addNode(a1, "m1", "r0", "s0");
+ addNode(a2, "m1", "r0", "s1");
+ addNode(a3, "m1", "r1", "s0");
+ addNode(a4, "m0", "r0", "s1");
+ addNode(a5, "m0", "r1", "s1");
+ addNode(a6, "m0", "r1", "s0");
+ addNode(a7, "m0", "r0", "s3");
+ addNode(a8, "m0", "r0", "s2");
+ addNode(a9, "m0", "r0", "s0");
+ setAddresses();
+
+ assertLocation(ch.locate(a0, 1), true, a0);
+ assertLocation(ch.locate(a1, 1), true, a1);
+ assertLocation(ch.locate(a2, 1), true, a2);
+ assertLocation(ch.locate(a3, 1), true, a3);
+ assertLocation(ch.locate(a4, 1), true, a4);
+ assertLocation(ch.locate(a5, 1), true, a5);
+ assertLocation(ch.locate(a6, 1), true, a6);
+ assertLocation(ch.locate(a7, 1), true, a7);
+ assertLocation(ch.locate(a8, 1), true, a8);
+ assertLocation(ch.locate(a9, 1), true, a9);
+
+ assertLocation(ch.locate(a0, 2), true, a0, a1);
+ assertLocation(ch.locate(a1, 2), true, a1, a2);
+ assertLocation(ch.locate(a2, 2), true, a2, a3);
+ assertLocation(ch.locate(a3, 2), true, a3, a4);
+ assertLocation(ch.locate(a4, 2), true, a4, a6);
+ assertLocation(ch.locate(a5, 2), true, a5, a6);
+ assertLocation(ch.locate(a6, 2), true, a6, a7);
+ assertLocation(ch.locate(a7, 2), true, a7, a8);
+ assertLocation(ch.locate(a8, 2), true, a8, a9);
+ assertLocation(ch.locate(a9, 2), true, a9, a0);
+
+
+ assertLocation(ch.getStateProvidersOnLeave(a0, 2), false, a1, a9);
+ assertLocation(ch.getStateProvidersOnLeave(a1, 2), false, a0, a2);
+ assertLocation(ch.getStateProvidersOnLeave(a2, 2), false, a3, a1);
+ assertLocation(ch.getStateProvidersOnLeave(a3, 2), false, a4, a2);
+ assertLocation(ch.getStateProvidersOnLeave(a4, 2), false, a6, a3);
+ assertLocation(ch.getStateProvidersOnLeave(a5, 2), false, a6);
+ assertLocation(ch.getStateProvidersOnLeave(a6, 2), false, a4, a7, a5);
+ assertLocation(ch.getStateProvidersOnLeave(a7, 2), false, a8, a6);
+ assertLocation(ch.getStateProvidersOnLeave(a8, 2), false, a9, a7);
+ assertLocation(ch.getStateProvidersOnLeave(a9, 2), false, a0, a8);
+
+ assertLocation(ch.getStateProvidersOnJoin(a0, 2), false, a1, a9);
+ assertLocation(ch.getStateProvidersOnJoin(a1, 2), false, a0, a2);
+ assertLocation(ch.getStateProvidersOnJoin(a2, 2), false, a3, a1);
+ assertLocation(ch.getStateProvidersOnJoin(a3, 2), false, a4, a2);
+ assertLocation(ch.getStateProvidersOnJoin(a4, 2), false, a6, a3);
+ assertLocation(ch.getStateProvidersOnJoin(a5, 2), false, a6);
+ assertLocation(ch.getStateProvidersOnJoin(a6, 2), false, a4, a7, a5);
+ assertLocation(ch.getStateProvidersOnJoin(a7, 2), false, a8, a6);
+ assertLocation(ch.getStateProvidersOnJoin(a8, 2), false, a9, a7);
+ assertLocation(ch.getStateProvidersOnJoin(a9, 2), false, a0, a8);
+
+
+ assertLocation(ch.locate(a0, 3), true, a0, a1, a3);
+ assertLocation(ch.locate(a1, 3), true, a1, a2, a4);
+ assertLocation(ch.locate(a2, 3), true, a2, a3, a6);
+ assertLocation(ch.locate(a3, 3), true, a3, a4, a5);
+ assertLocation(ch.locate(a4, 3), true, a4, a6, a7);
+ assertLocation(ch.locate(a5, 3), true, a5, a6, a7);
+ assertLocation(ch.locate(a6, 3), true, a6, a7, a8);
+ assertLocation(ch.locate(a7, 3), true, a7, a8, a9);
+ assertLocation(ch.locate(a8, 3), true, a8, a9, a0);
+ assertLocation(ch.locate(a9, 3), true, a9, a0, a2);
+ }
+
+ public void testConsistencyWhenNodeLeaves() {
+ addNode(a0, "m2", "r0", "s1");
+ addNode(a1, "m1", "r0", "s0");
+ addNode(a2, "m1", "r0", "s1");
+ addNode(a3, "m1", "r1", "s0");
+ addNode(a4, "m0", "r0", "s1");
+ addNode(a5, "m0", "r1", "s1");
+ addNode(a6, "m0", "r1", "s0");
+ addNode(a7, "m0", "r0", "s3");
+ addNode(a8, "m0", "r0", "s2");
+ addNode(a9, "m0", "r0", "s0");
+ setAddresses();
+
+ List<Address> a0List = ch.locate(a0, 3);
+ List<Address> a1List = ch.locate(a1, 3);
+ List<Address> a2List = ch.locate(a2, 3);
+ List<Address> a3List = ch.locate(a3, 3);
+ List<Address> a4List = ch.locate(a4, 3);
+ List<Address> a5List = ch.locate(a5, 3);
+ List<Address> a6List = ch.locate(a6, 3);
+ List<Address> a7List = ch.locate(a7, 3);
+ List<Address> a8List = ch.locate(a8, 3);
+ List<Address> a9List = ch.locate(a9, 3);
+
+ for (Address addr: addresses) {
+ System.out.println("addr = " + addr);
+ List<Address> addressCopy = (List<Address>) addresses.clone();
+ addressCopy.remove(addr);
+ ch.setCaches(addressCopy);
+ checkConsistency(a0List, a0, addr, 3);
+ checkConsistency(a1List, a1, addr, 3);
+ checkConsistency(a2List, a2, addr, 3);
+ checkConsistency(a3List, a3, addr, 3);
+ checkConsistency(a4List, a4, addr, 3);
+ checkConsistency(a5List, a5, addr, 3);
+ checkConsistency(a6List, a6, addr, 3);
+ checkConsistency(a7List, a7, addr, 3);
+ checkConsistency(a8List, a8, addr, 3);
+ checkConsistency(a9List, a9, addr, 3);
+ }
+ }
+
+ private void checkConsistency(List<Address> a0List, TestAddress a0, Address addr, int replCount) {
+ a0List = new ArrayList(a0List);
+ a0List.remove(addr);
+ if (a0.equals(addr)) return;
+ List<Address> currentBackupList = ch.locate(a0, replCount);
+ assertEquals(replCount, currentBackupList.size(), currentBackupList.toString());
+ assert currentBackupList.containsAll(a0List) : "Current backups are: " + currentBackupList + "Previous: " + a0List;
+ }
+
+
+ private void assertLocation(List<Address> received, boolean enforceSequence, TestAddress... expected) {
+ if (expected == null) {
+ assert received.isEmpty();
+ }
+ assertEquals(expected.length, received.size());
+ if (enforceSequence) {
+ assert received.equals(Arrays.asList(expected)) : "Received: " + received + " Expected: " + Arrays.toString(expected);
+ } else {
+ assert received.containsAll(Arrays.asList(expected)) : "Received: " + received + " Expected: " + Arrays.toString(expected);
+ }
+ }
+
+ private void addNode(TestAddress address, String machineId, String rackId, String siteId) {
+ addresses.add(address);
+ NodeTopologyInfo nti = new NodeTopologyInfo(machineId, rackId, siteId);
+ ti.addNodeTopologyInfo(address, nti);
+ }
+
+ private void setAddresses() {
+ ch.setCaches(addresses);
+ staticAddresses = new TestAddress[]{a0, a1, a2, a3, a4, a5, a6, a7, a8, a9};
+ for (int i = 0; i < staticAddresses.length; i++) {
+ if (staticAddresses[i] != null) staticAddresses[i].setName("a" + i);
+ }
+ log.info("Static addresses: " + Arrays.toString(staticAddresses));
+ }
+
+ public TestAddress address(int hashCode) {
+ return new TestAddress(hashCode);
+ }
+}
Modified: branches/4.2.x/core/src/test/java/org/infinispan/distribution/rehash/RehashTestBase.java
===================================================================
--- branches/4.2.x/core/src/test/java/org/infinispan/distribution/rehash/RehashTestBase.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/test/java/org/infinispan/distribution/rehash/RehashTestBase.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -44,7 +44,7 @@
TestingUtil.sleepThread(1000);
}
- private List<MagicKey> init() {
+ protected List<MagicKey> init() {
List<MagicKey> keys = new ArrayList<MagicKey>(Arrays.asList(
new MagicKey(c1, "k1"), new MagicKey(c2, "k2"),
new MagicKey(c3, "k3"), new MagicKey(c4, "k4")
Modified: branches/4.2.x/core/src/test/java/org/infinispan/distribution/rehash/WorkDuringJoinTest.java
===================================================================
--- branches/4.2.x/core/src/test/java/org/infinispan/distribution/rehash/WorkDuringJoinTest.java 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/core/src/test/java/org/infinispan/distribution/rehash/WorkDuringJoinTest.java 2010-10-27 20:44:10 UTC (rev 2621)
@@ -2,8 +2,8 @@
import org.infinispan.Cache;
import org.infinispan.distribution.BaseDistFunctionalTest;
-import org.infinispan.distribution.ConsistentHash;
-import org.infinispan.distribution.ConsistentHashHelper;
+import org.infinispan.distribution.ch.ConsistentHash;
+import org.infinispan.distribution.ch.ConsistentHashHelper;
import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.remoting.transport.Address;
import org.testng.annotations.Test;
Modified: branches/4.2.x/server/hotrod/src/test/scala/org/infinispan/server/hotrod/HotRodDistributionTest.scala
===================================================================
--- branches/4.2.x/server/hotrod/src/test/scala/org/infinispan/server/hotrod/HotRodDistributionTest.scala 2010-10-27 17:02:35 UTC (rev 2620)
+++ branches/4.2.x/server/hotrod/src/test/scala/org/infinispan/server/hotrod/HotRodDistributionTest.scala 2010-10-27 20:44:10 UTC (rev 2621)
@@ -9,7 +9,7 @@
import test.HotRodTestingUtil._
import org.testng.Assert._
import org.infinispan.test.TestingUtil
-import org.infinispan.distribution.UnionConsistentHash
+import org.infinispan.distribution.ch.UnionConsistentHash
import collection.mutable.ListBuffer
import org.infinispan.test.AbstractCacheTest._ // Do not remove, otherwise getDefaultClusteredConfig is not found
More information about the infinispan-commits
mailing list