natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

1. Switch bootstrapNodes in TokenMetadata to Map<Token,EndPoint> so RackUnawareStrategy can use it. 

2. Fix AbstractStrategy and RackUnawareStrategy to incorporate the bootstrapping nodes for
getHintedStorageEndPoints through StorageService.getNStorageEndPointMap (now used by insert and
insertBlocking after 383)
3. Add unit test fot RackUnawareStrategy to test if bootstrapping nodes are being returned correctly.

patch by Sandeep Tata; reviewed by jbellis for CASSANDRA-375

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@808161 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-08-26 19:04:07 +00:00
parent 489143566d
commit 847f3b0293
6 changed files with 124 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/java/org/apache/cassandra/locator/AbstractStrategy.java
View File

@ -101,10 +101,15 @@ public abstract class AbstractStrategy implements IReplicaPlacementStrategy
* Get the map of top N to the live nodes currently.
*/
public Map<EndPoint, EndPoint> getHintedStorageEndPoints(Token token)
{
EndPoint[] topN = getStorageEndPointsForWrite( token );
return getHintedMapForEndpoints(topN);
}
private Map<EndPoint, EndPoint> getHintedMapForEndpoints(EndPoint[] topN)
{
List<EndPoint> liveList = new ArrayList<EndPoint>();
Map<EndPoint, EndPoint> map = new HashMap<EndPoint, EndPoint>();
EndPoint[] topN = getStorageEndPoints( token );
for( int i = 0 ; i < topN.length ; i++)
{

1
src/java/org/apache/cassandra/locator/IReplicaPlacementStrategy.java
View File

@ -33,6 +33,7 @@ import org.apache.cassandra.net.EndPoint;
public interface IReplicaPlacementStrategy
{
public EndPoint[] getStorageEndPoints(Token token);
public EndPoint[] getStorageEndPointsForWrite(Token token);
public Map<String, EndPoint[]> getStorageEndPoints(String[] keys);
public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap);
public Map<EndPoint, EndPoint> getHintedStorageEndPoints(Token token);

11
src/java/org/apache/cassandra/locator/RackAwareStrategy.java
View File

@ -139,4 +139,15 @@ public class RackAwareStrategy extends AbstractStrategy
{
throw new UnsupportedOperationException("This operation is not currently supported");
}
public EndPoint[] getStorageEndPointsForWrite(Token token)
{
throw new UnsupportedOperationException("Rack-aware bootstrapping not supported");
}
public Map<EndPoint, EndPoint> getHintedStorageEndPointsForWrite(Token token)
{
throw new UnsupportedOperationException("Rack-aware bootstrapping not supported");
}
}

56
src/java/org/apache/cassandra/locator/RackUnawareStrategy.java
View File

@ -46,12 +46,47 @@ public class RackUnawareStrategy extends AbstractStrategy
return getStorageEndPoints(token, tokenMetadata_.cloneTokenEndPointMap());
}
public EndPoint[] getStorageEndPointsForWrite(Token token)
{
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Map<Token, EndPoint> bootstrapTokensToEndpointMap = tokenMetadata_.cloneBootstrapNodes();
List<Token> tokenList = getStorageTokens(token, tokenToEndPointMap, bootstrapTokensToEndpointMap);
List<EndPoint> list = new ArrayList<EndPoint>();
for (Token t: tokenList)
{
EndPoint e = tokenToEndPointMap.get(t);
if (e == null)
e = bootstrapTokensToEndpointMap.get(t);
assert e != null;
list.add(e);
}
retrofitPorts(list);
return list.toArray(new EndPoint[list.size()]);
}
public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap)
{
int startIndex;
List<Token> tokenList = getStorageTokens(token, tokenToEndPointMap, null);
List<EndPoint> list = new ArrayList<EndPoint>();
for (Token t: tokenList)
list.add(tokenToEndPointMap.get(t));
retrofitPorts(list);
return list.toArray(new EndPoint[list.size()]);
}
private List<Token> getStorageTokens(Token token, Map<Token, EndPoint> tokenToEndPointMap, Map<Token, EndPoint> bootStrapTokenToEndPointMap)
{
int startIndex;
List<Token> tokenList = new ArrayList<Token>();
int foundCount = 0;
List tokens = new ArrayList<Token>(tokenToEndPointMap.keySet());
List<Token> bsTokens = null;
if (bootStrapTokenToEndPointMap != null)
{
bsTokens = new ArrayList<Token>(bootStrapTokenToEndPointMap.keySet());
tokens.addAll(bsTokens);
}
Collections.sort(tokens);
int index = Collections.binarySearch(tokens, token);
if(index < 0)
@ -61,22 +96,24 @@ public class RackUnawareStrategy extends AbstractStrategy
index = 0;
}
int totalNodes = tokens.size();
// Add the node at the index by default
list.add(tokenToEndPointMap.get(tokens.get(index)));
foundCount++;
// Add the token at the index by default
tokenList.add((Token)tokens.get(index));
if (bsTokens == null || !bsTokens.contains(tokens.get(index)))
foundCount++;
startIndex = (index + 1)%totalNodes;
// If we found N number of nodes we are good. This loop will just exit. Otherwise just
// loop through the list and add until we have N nodes.
for (int i = startIndex, count = 1; count < totalNodes && foundCount < replicas_; ++count, i = (i+1)%totalNodes)
{
if( ! list.contains(tokenToEndPointMap.get(tokens.get(i))))
if(!tokenList.contains(tokens.get(i)))
{
list.add(tokenToEndPointMap.get(tokens.get(i)));
foundCount++;
tokenList.add((Token)tokens.get(i));
//Don't count bootstrapping tokens towards the count
if (bsTokens==null || !bsTokens.contains(tokens.get(i)))
foundCount++;
}
}
retrofitPorts(list);
return list.toArray(new EndPoint[list.size()]);
return tokenList;
}
public Map<String, EndPoint[]> getStorageEndPoints(String[] keys)
@ -87,7 +124,6 @@ public class RackUnawareStrategy extends AbstractStrategy
{
results.put(key, getStorageEndPoints(partitioner_.getToken(key)));
}
return results;
}
}

18
src/java/org/apache/cassandra/locator/TokenMetadata.java
View File

@ -32,7 +32,7 @@ public class TokenMetadata
/* Maintains a reverse index of endpoint to token in the cluster. */
private Map<EndPoint, Token> endPointToTokenMap_ = new HashMap<EndPoint, Token>();
/* Bootstrapping nodes and their tokens */
private Map<EndPoint, Token> bootstrapNodes = Collections.synchronizedMap(new HashMap<EndPoint, Token>());
private Map<Token, EndPoint> bootstrapNodes = Collections.synchronizedMap(new HashMap<Token, EndPoint>());
/* Use this lock for manipulating the token map */
private final ReadWriteLock lock_ = new ReentrantReadWriteLock(true);
@ -40,19 +40,19 @@ public class TokenMetadata
public TokenMetadata()
{
}
public TokenMetadata(Map<Token, EndPoint> tokenToEndPointMap, Map<EndPoint, Token> endPointToTokenMap, Map<EndPoint, Token> bootstrapNodes)
public TokenMetadata(Map<Token, EndPoint> tokenToEndPointMap, Map<EndPoint, Token> endPointToTokenMap, Map<Token, EndPoint> bootstrapNodes)
{
tokenToEndPointMap_ = tokenToEndPointMap;
endPointToTokenMap_ = endPointToTokenMap;
this.bootstrapNodes = bootstrapNodes;
this.bootstrapNodes = bootstrapNodes;
}
public TokenMetadata cloneMe()
{
return new TokenMetadata(cloneTokenEndPointMap(), cloneEndPointTokenMap(), cloneBootstrapNodes());
}
public void update(Token token, EndPoint endpoint)
{
this.update(token, endpoint, false);
@ -67,12 +67,12 @@ public class TokenMetadata
{
if (bootstrapState)
{
bootstrapNodes.put(endpoint, token);
bootstrapNodes.put(token, endpoint);
this.remove(endpoint);
}
else
{
bootstrapNodes.remove(endpoint); // If this happened to be there
bootstrapNodes.remove(token); // If this happened to be there
Token oldToken = endPointToTokenMap_.get(endpoint);
if ( oldToken != null )
tokenToEndPointMap_.remove(oldToken);
@ -168,12 +168,12 @@ public class TokenMetadata
}
}
public Map<EndPoint, Token> cloneBootstrapNodes()
public Map<Token, EndPoint> cloneBootstrapNodes()
{
lock_.readLock().lock();
try
{
return new HashMap<EndPoint, Token>( bootstrapNodes );
return new HashMap<Token, EndPoint>( bootstrapNodes );
}
finally
{

53
test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java
View File

@ -18,6 +18,9 @@
*/
package org.apache.cassandra.locator;
import static org.junit.Assert.*;
import java.util.Arrays;
import java.util.List;
import java.util.ArrayList;
@ -79,11 +82,57 @@ public class RackUnawareStrategyTest
for (int i = 0; i < keyTokens.length; i++)
{
EndPoint[] endPoints = strategy.getStorageEndPoints(keyTokens[i]);
assert endPoints.length == 3;
assertEquals(3, endPoints.length);
for (int j = 0; j < endPoints.length; j++)
{
assert endPoints[j] == hosts.get((i + j + 1) % hosts.size());
assertEquals(endPoints[j], hosts.get((i + j + 1) % hosts.size()));
}
}
}
@Test
public void testGetStorageEndPointsDuringBootstrap()
{
TokenMetadata tmd = new TokenMetadata();
IPartitioner partitioner = new RandomPartitioner();
IReplicaPlacementStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000);
Token[] endPointTokens = new Token[5];
Token[] keyTokens = new Token[5];
for (int i = 0; i < 5; i++)
{
endPointTokens[i] = new BigIntegerToken(String.valueOf(10 * i));
keyTokens[i] = new BigIntegerToken(String.valueOf(10 * i + 5));
}
List<EndPoint> hosts = new ArrayList<EndPoint>();
for (int i = 0; i < endPointTokens.length; i++)
{
EndPoint ep = new EndPoint("127.0.0." + String.valueOf(i + 1), 7001);
tmd.update(endPointTokens[i], ep);
hosts.add(ep);
}
//Add bootstrap node id=6
Token bsToken = new BigIntegerToken(String.valueOf(25));
EndPoint bootstrapEndPoint = new EndPoint("127.0.0.6", 7001);
tmd.update(bsToken, bootstrapEndPoint, true);
for (int i = 0; i < keyTokens.length; i++)
{
EndPoint[] endPoints = strategy.getStorageEndPointsForWrite(keyTokens[i]);
assertTrue(endPoints.length >=3);
List<EndPoint> endPointsList = Arrays.asList(endPoints);
for (int j = 0; j < 3; j++)
{
//Check that the old nodes are definitely included
assertTrue(endPointsList.contains(hosts.get((i + j + 1) % hosts.size())));
}
// for 5, 15, 25 this should include bootstrap node
if (i < 3)
assertTrue(endPointsList.contains(bootstrapEndPoint));
}
}
}