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

replace BigInteger tokens with BigIntegerToken and StringToken in RandomPartitioner and OrderPreservingPartitioner, respectively. (OrderPreservingHashPartitioner is no more.) 

Doing order preserving partitioning based on the raw string has a number of compelling advantages:
 - there is no key length that all tokens must be padded to (which can be expensive if there is a range of key lengths) and which cannot be increased after deployment
 - it allows user-defined collations [sorting] rather than being limited to sorting by code point value (which is useless in the unicode world, and not always what you want even for ascii keys)
 - it will work with all UTF-16 characters, not just the UCS-2 subset (this is a limitation of using as a base 2**16, i.e., assuming for your order preserving hash that all characters are two bytes).

patch by jbellis; reviewed by Jun Rao for CASSANDRA-65

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@769016 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-04-27 15:01:29 +00:00
parent 2e9466c305
commit 907d92e21b
31 changed files with 578 additions and 512 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -43,6 +43,7 @@ import org.apache.log4j.Logger;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.io.DataOutputBuffer;
import org.apache.cassandra.io.IndexHelper;
@ -50,7 +51,6 @@ import org.apache.cassandra.io.SSTable;
import org.apache.cassandra.io.SequenceFile;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.IPartitioner;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.FileUtils;
import org.apache.cassandra.utils.LogUtil;
@ -1113,7 +1113,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
continue;
}
}
if ( Range.isKeyInRanges(ranges, p.undecorateKey(lastkey)) )
if (Range.isKeyInRanges(p.undecorateKey(lastkey), ranges))
{
if(ssTableRange == null )
{
@ -1143,7 +1143,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
continue;
}
/* keep on looping until we find a key in the range */
while ( !Range.isKeyInRanges(ranges, p.undecorateKey(filestruct.getKey())) )
while (!Range.isKeyInRanges(p.undecorateKey(filestruct.getKey()), ranges))
{
filestruct.advance();
if (filestruct.isExhausted())
@ -1151,7 +1151,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
break;
}
/* check if we need to continue , if we are done with ranges empty the queue and close all file handles and exit */
//if( !isLoop && StorageService.hash(filestruct.key).compareTo(maxRange.right()) > 0 && !filestruct.key.equals(""))
//if( !isLoop && StorageService.token(filestruct.key).compareTo(maxRange.right()) > 0 && !filestruct.key.equals(""))
//{
//filestruct.reader.close();
//filestruct = null;

41
src/org/apache/cassandra/db/DBManager.java
View File

@ -19,18 +19,17 @@
package org.apache.cassandra.db;
import java.io.IOException;
import java.math.BigInteger;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.BasicUtilities;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.GuidGenerator;
/**
@ -62,23 +61,23 @@ public class DBManager
public static class StorageMetadata
{
private BigInteger storageId_;
private Token myToken;
private int generation_;
StorageMetadata(BigInteger storageId, int generation)
StorageMetadata(Token storageId, int generation)
{
storageId_ = storageId;
myToken = storageId;
generation_ = generation;
}
public BigInteger getStorageId()
public Token getStorageId()
{
return storageId_;
return myToken;
}
public void setStorageId(BigInteger storageId)
public void setStorageId(Token storageId)
{
storageId_ = storageId;
myToken = storageId;
}
public int getGeneration()
@ -115,22 +114,17 @@ public class DBManager
SystemTable sysTable = SystemTable.openSystemTable(SystemTable.name_);
Row row = sysTable.get(FBUtilities.getHostAddress());
Random random = new Random();
IPartitioner p = StorageService.getPartitioner();
if ( row == null )
{
/* Generate a token for this Storage node */
String guid = GuidGenerator.guid();
BigInteger token = StorageService.hash(guid);
if ( token.signum() == -1 )
token = token.multiply(BigInteger.valueOf(-1L));
Token token = p.getDefaultToken();
int generation = 1;
String key = FBUtilities.getHostAddress();
row = new Row(key);
ColumnFamily cf = new ColumnFamily(SystemTable.cfName_, "Standard");
cf.addColumn(new Column(SystemTable.token_, token.toByteArray()));
cf.addColumn(new Column(SystemTable.generation_, BasicUtilities.intToByteArray(generation)));
cf.addColumn(new Column(SystemTable.token_, p.getTokenFactory().toByteArray(token)));
cf.addColumn(new Column(SystemTable.generation_, BasicUtilities.intToByteArray(generation)) );
row.addColumnFamily(cf);
sysTable.apply(row);
storageMetadata = new StorageMetadata( token, generation);
@ -145,14 +139,15 @@ public class DBManager
{
ColumnFamily columnFamily = columnFamilies.get(cfName);
IColumn token = columnFamily.getColumn(SystemTable.token_);
BigInteger bi = new BigInteger( token.value() );
IColumn tokenColumn = columnFamily.getColumn(SystemTable.token_);
Token token = p.getTokenFactory().fromByteArray(tokenColumn.value());
IColumn generation = columnFamily.getColumn(SystemTable.generation_);
int gen = BasicUtilities.byteArrayToInt(generation.value()) + 1;
columnFamily.addColumn(new Column("Generation", BasicUtilities.intToByteArray(gen), generation.timestamp() + 1));
storageMetadata = new StorageMetadata( bi, gen );
Column generation2 = new Column("Generation", BasicUtilities.intToByteArray(gen), generation.timestamp() + 1);
columnFamily.addColumn(generation2);
storageMetadata = new StorageMetadata(token, gen);
break;
}
sysTable.reset(row);

2
src/org/apache/cassandra/db/FileStruct.java
View File

@ -24,7 +24,7 @@ import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.io.DataOutputBuffer;
import org.apache.cassandra.io.IFileReader;
import org.apache.cassandra.io.SSTable;
import org.apache.cassandra.service.IPartitioner;
import org.apache.cassandra.dht.IPartitioner;
public class FileStruct implements Comparable<FileStruct>

25
src/org/apache/cassandra/db/SystemTable.java
View File

@ -19,10 +19,11 @@
package org.apache.cassandra.db;
import java.io.IOException;
import java.math.BigInteger;
import java.util.HashMap;
import java.util.Map;
import org.apache.log4j.Logger;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.io.DataOutputBuffer;
@ -31,7 +32,8 @@ import org.apache.cassandra.io.IFileWriter;
import org.apache.cassandra.io.SequenceFile;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.LogUtil;
import org.apache.log4j.Logger;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.dht.IPartitioner;
/**
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
@ -148,8 +150,9 @@ public class SystemTable
* This method is used to update the SystemTable with the
* new token.
*/
public void updateToken(BigInteger token) throws IOException
public void updateToken(Token token) throws IOException
{
IPartitioner p = StorageService.getPartitioner();
if ( systemRow_ != null )
{
Map<String, ColumnFamily> columnFamilies = systemRow_.getColumnFamilyMap();
@ -157,9 +160,9 @@ public class SystemTable
ColumnFamily columnFamily = columnFamilies.get(SystemTable.cfName_);
long oldTokenColumnTimestamp = columnFamily.getColumn(SystemTable.token_).timestamp();
/* create the "Token" whose value is the new token. */
IColumn tokenColumn = new Column(SystemTable.token_, token.toByteArray(), oldTokenColumnTimestamp + 1);
IColumn tokenColumn = new Column(SystemTable.token_, p.getTokenFactory().toByteArray(token), oldTokenColumnTimestamp + 1);
/* replace the old "Token" column with this new one. */
logger_.debug("Replacing old token " + new BigInteger( columnFamily.getColumn(SystemTable.token_).value() ).toString() + " with token " + token.toString());
logger_.debug("Replacing old token " + p.getTokenFactory().fromByteArray(columnFamily.getColumn(SystemTable.token_).value()) + " with " + token);
columnFamily.addColumn(tokenColumn);
reset(systemRow_);
}
@ -180,17 +183,7 @@ public class SystemTable
{
LogUtil.init();
StorageService.instance().start();
SystemTable.openSystemTable(SystemTable.cfName_).updateToken( StorageService.hash("503545744:0") );
SystemTable.openSystemTable(SystemTable.cfName_).updateToken(StorageService.token("503545744:0"));
System.out.println("Done");
/*
BigInteger hash = StorageService.hash("304700067:0");
List<Range> ranges = new ArrayList<Range>();
ranges.add( new Range(new BigInteger("1218069462158869448693347920504606362273788442553"), new BigInteger("1092770595533781724218060956188429069")) );
if ( Range.isKeyInRanges(ranges, "304700067:0") )
{
System.out.println("Done");
}
*/
}
}

16
src/org/apache/cassandra/dht/BigIntegerToken.java Normal file
View File

@ -0,0 +1,16 @@
package org.apache.cassandra.dht;
import java.math.BigInteger;
public class BigIntegerToken extends Token<BigInteger>
{
public BigIntegerToken(BigInteger token)
{
super(token);
}
// convenience method for testing
public BigIntegerToken(String token) {
this(new BigInteger(token));
}
}

42
src/org/apache/cassandra/dht/BootStrapper.java
View File

@ -18,24 +18,20 @@
package org.apache.cassandra.dht;
import java.io.IOException;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.LogUtil;
import org.apache.log4j.Logger;
import org.apache.log4j.Logger;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.LogUtil;
/**
@ -48,18 +44,18 @@ public class BootStrapper implements Runnable
/* endpoints that need to be bootstrapped */
protected EndPoint[] targets_ = new EndPoint[0];
/* tokens of the nodes being bootstapped. */
protected BigInteger[] tokens_ = new BigInteger[0];
protected final Token[] tokens_;
protected TokenMetadata tokenMetadata_ = null;
private List<EndPoint> filters_ = new ArrayList<EndPoint>();
public BootStrapper(EndPoint[] target, BigInteger[] token)
public BootStrapper(EndPoint[] target, Token... token)
{
targets_ = target;
tokens_ = token;
tokenMetadata_ = StorageService.instance().getTokenMetadata();
}
public BootStrapper(EndPoint[] target, BigInteger[] token, EndPoint[] filters)
public BootStrapper(EndPoint[] target, Token[] token, EndPoint[] filters)
{
this(target, token);
Collections.addAll(filters_, filters);
@ -71,14 +67,14 @@ public class BootStrapper implements Runnable
{
logger_.debug("Beginning bootstrap process for " + targets_ + " ...");
/* copy the token to endpoint map */
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
/* remove the tokens associated with the endpoints being bootstrapped */
for ( BigInteger token : tokens_ )
for (Token token : tokens_)
{
tokenToEndPointMap.remove(token);
}
Set<BigInteger> oldTokens = new HashSet<BigInteger>( tokenToEndPointMap.keySet() );
Set<Token> oldTokens = new HashSet<Token>( tokenToEndPointMap.keySet() );
Range[] oldRanges = StorageService.instance().getAllRanges(oldTokens);
logger_.debug("Total number of old ranges " + oldRanges.length);
/*

10
src/org/apache/cassandra/dht/IPartitioner.java
View File

@ -18,14 +18,10 @@
package org.apache.cassandra.dht;
import java.math.BigInteger;
import java.util.Comparator;
import java.util.Iterator;
public interface IPartitioner
{
public BigInteger hash(String key);
/**
* transform key to on-disk format s.t. keys are stored in node comparison order.
* this lets bootstrap rip out parts of the sstable sequentially instead of doing random seeks.
@ -40,4 +36,10 @@ public interface IPartitioner
public Comparator<String> getDecoratedKeyComparator();
public Comparator<String> getReverseDecoratedKeyComparator();
public Token getTokenForKey(String key);
public Token getDefaultToken();
public Token.TokenFactory getTokenFactory();
}

34
src/org/apache/cassandra/dht/LeaveJoinProtocolHelper.java
View File

@ -18,19 +18,19 @@
package org.apache.cassandra.dht;
import java.io.IOException;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.log4j.Logger;
import org.apache.log4j.Logger;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
class LeaveJoinProtocolHelper
@ -42,20 +42,20 @@ class LeaveJoinProtocolHelper
* a-----x-----y-----b then we want a mapping from
* (a, b] --> (a, x], (x, y], (y, b]
*/
protected static Map<Range, List<Range>> getRangeSplitRangeMapping(Range[] oldRanges, BigInteger[] allTokens)
protected static Map<Range, List<Range>> getRangeSplitRangeMapping(Range[] oldRanges, Token[] allTokens)
{
Map<Range, List<Range>> splitRanges = new HashMap<Range, List<Range>>();
BigInteger[] tokens = new BigInteger[allTokens.length];
Token[] tokens = new Token[allTokens.length];
System.arraycopy(allTokens, 0, tokens, 0, tokens.length);
Arrays.sort(tokens);
Range prevRange = null;
BigInteger prevToken = null;
Token prevToken = null;
boolean bVal = false;
for ( Range oldRange : oldRanges )
{
if ( bVal && prevRange != null )
if (bVal)
{
bVal = false;
List<Range> subRanges = splitRanges.get(prevRange);
@ -65,7 +65,7 @@ class LeaveJoinProtocolHelper
prevRange = oldRange;
prevToken = oldRange.left();
for ( BigInteger token : tokens )
for (Token token : tokens)
{
List<Range> subRanges = splitRanges.get(oldRange);
if ( oldRange.contains(token) )

65
src/org/apache/cassandra/dht/LeaveJoinProtocolImpl.java
View File

@ -18,21 +18,20 @@
package org.apache.cassandra.dht;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.LogUtil;
import org.apache.log4j.Logger;
import org.apache.log4j.Logger;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.LogUtil;
/**
@ -48,11 +47,11 @@ public class LeaveJoinProtocolImpl implements Runnable
/* endpoints that are to be moved. */
protected EndPoint[] targets_ = new EndPoint[0];
/* position where they need to be moved */
protected BigInteger[] tokens_ = new BigInteger[0];
protected final Token[] tokens_;
/* token metadata information */
protected TokenMetadata tokenMetadata_ = null;
public LeaveJoinProtocolImpl(EndPoint[] targets, BigInteger[] tokens)
public LeaveJoinProtocolImpl(EndPoint[] targets, Token[] tokens)
{
targets_ = targets;
tokens_ = tokens;
@ -65,24 +64,24 @@ public class LeaveJoinProtocolImpl implements Runnable
{
logger_.debug("Beginning leave/join process for ...");
/* copy the token to endpoint map */
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
/* copy the endpoint to token map */
Map<EndPoint, BigInteger> endpointToTokenMap = tokenMetadata_.cloneEndPointTokenMap();
Map<EndPoint, Token> endpointToTokenMap = tokenMetadata_.cloneEndPointTokenMap();
Set<BigInteger> oldTokens = new HashSet<BigInteger>( tokenToEndPointMap.keySet() );
Set<Token> oldTokens = new HashSet<Token>( tokenToEndPointMap.keySet() );
Range[] oldRanges = StorageService.instance().getAllRanges(oldTokens);
logger_.debug("Total number of old ranges " + oldRanges.length);
/* Calculate the list of nodes that handle the old ranges */
Map<Range, List<EndPoint>> oldRangeToEndPointMap = StorageService.instance().constructRangeToEndPointMap(oldRanges);
/* Remove the tokens of the nodes leaving the ring */
Set<BigInteger> tokens = getTokensForLeavingNodes();
Set<Token> tokens = getTokensForLeavingNodes();
oldTokens.removeAll(tokens);
Range[] rangesAfterNodesLeave = StorageService.instance().getAllRanges(oldTokens);
/* Get expanded range to initial range mapping */
Map<Range, List<Range>> expandedRangeToOldRangeMap = getExpandedRangeToOldRangeMapping(oldRanges, rangesAfterNodesLeave);
/* add the new token positions to the old tokens set */
for ( BigInteger token : tokens_ )
for (Token token : tokens_)
oldTokens.add(token);
Range[] rangesAfterNodesJoin = StorageService.instance().getAllRanges(oldTokens);
/* replace the ranges that were split with the split ranges in the old configuration */
@ -196,12 +195,12 @@ public class LeaveJoinProtocolImpl implements Runnable
}
}
private Set<BigInteger> getTokensForLeavingNodes()
private Set<Token> getTokensForLeavingNodes()
{
Set<BigInteger> tokens = new HashSet<BigInteger>();
Set<Token> tokens = new HashSet<Token>();
for ( EndPoint target : targets_ )
{
tokens.add( tokenMetadata_.getToken(target) );
tokens.add(tokenMetadata_.getToken(target));
}
return tokens;
}
@ -276,16 +275,16 @@ public class LeaveJoinProtocolImpl implements Runnable
public static void main(String[] args) throws Throwable
{
StorageService ss = StorageService.instance();
ss.updateTokenMetadata(BigInteger.valueOf(3), new EndPoint("A", 7000));
ss.updateTokenMetadata(BigInteger.valueOf(6), new EndPoint("B", 7000));
ss.updateTokenMetadata(BigInteger.valueOf(9), new EndPoint("C", 7000));
ss.updateTokenMetadata(BigInteger.valueOf(12), new EndPoint("D", 7000));
ss.updateTokenMetadata(BigInteger.valueOf(15), new EndPoint("E", 7000));
ss.updateTokenMetadata(BigInteger.valueOf(18), new EndPoint("F", 7000));
ss.updateTokenMetadata(BigInteger.valueOf(21), new EndPoint("G", 7000));
ss.updateTokenMetadata(BigInteger.valueOf(24), new EndPoint("H", 7000));
ss.updateTokenMetadata(new BigIntegerToken("3"), new EndPoint("A", 7000));
ss.updateTokenMetadata(new BigIntegerToken("6"), new EndPoint("B", 7000));
ss.updateTokenMetadata(new BigIntegerToken("9"), new EndPoint("C", 7000));
ss.updateTokenMetadata(new BigIntegerToken("12"), new EndPoint("D", 7000));
ss.updateTokenMetadata(new BigIntegerToken("15"), new EndPoint("E", 7000));
ss.updateTokenMetadata(new BigIntegerToken("18"), new EndPoint("F", 7000));
ss.updateTokenMetadata(new BigIntegerToken("21"), new EndPoint("G", 7000));
ss.updateTokenMetadata(new BigIntegerToken("24"), new EndPoint("H", 7000));
Runnable runnable = new LeaveJoinProtocolImpl( new EndPoint[]{new EndPoint("C", 7000), new EndPoint("D", 7000)}, new BigInteger[]{BigInteger.valueOf(22), BigInteger.valueOf(23)} );
Runnable runnable = new LeaveJoinProtocolImpl( new EndPoint[]{new EndPoint("C", 7000), new EndPoint("D", 7000)}, new Token[]{new BigIntegerToken("22"), new BigIntegerToken("23")} );
runnable.run();
}
}

81
src/org/apache/cassandra/dht/OrderPreservingHashPartitioner.java
View File

@ -1,81 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.dht;
import java.math.BigInteger;
import java.util.Comparator;
import org.apache.cassandra.dht.IPartitioner;
public class OrderPreservingHashPartitioner implements IPartitioner
{
private final static int maxKeyHashLength_ = 36;
private final static BigInteger ONE = BigInteger.ONE;
private static final BigInteger prime_ = BigInteger.valueOf(Character.MAX_VALUE);
private static final Comparator<String> comparator = new Comparator<String>()
{
public int compare(String o1, String o2)
{
return o1.compareTo(o2);
}
};
private static final Comparator<String> rcomparator = new Comparator<String>()
{
public int compare(String o1, String o2)
{
return -comparator.compare(o1, o2);
}
};
public BigInteger hash(String key)
{
BigInteger h = BigInteger.ZERO;
char val[] = key.toCharArray();
for (int i = 0; i < maxKeyHashLength_; i++)
{
if( i < val.length )
h = prime_.multiply(h).add( BigInteger.valueOf(val[i]) );
else
h = prime_.multiply(h).add( ONE );
}
return h;
}
public String decorateKey(String key)
{
return key;
}
public String undecorateKey(String decoratedKey)
{
return decoratedKey;
}
public Comparator<String> getReverseDecoratedKeyComparator()
{
return rcomparator;
}
public Comparator<String> getDecoratedKeyComparator()
{
return comparator;
}
}

117
src/org/apache/cassandra/dht/OrderPreservingPartitioner.java Normal file
View File

@ -0,0 +1,117 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.dht;
import java.io.UnsupportedEncodingException;
import java.text.Collator;
import java.util.Comparator;
import java.util.Locale;
import java.util.Random;
public class OrderPreservingPartitioner implements IPartitioner
{
// TODO make locale configurable. But don't just leave it up to the OS or you could really screw
// people over if they deploy on nodes with different OS locales.
static final Collator collator = Collator.getInstance(new Locale("en", "US"));
private static final Comparator<String> comparator = new Comparator<String>() {
public int compare(String o1, String o2)
{
return collator.compare(o1, o2);
}
};
private static final Comparator<String> reverseComparator = new Comparator<String>() {
public int compare(String o1, String o2)
{
return -comparator.compare(o1, o2);
}
};
public String decorateKey(String key)
{
return key;
}
public String undecorateKey(String decoratedKey)
{
return decoratedKey;
}
public Comparator<String> getDecoratedKeyComparator()
{
return comparator;
}
public Comparator<String> getReverseDecoratedKeyComparator()
{
return reverseComparator;
}
public StringToken getDefaultToken()
{
String chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
Random r = new Random();
StringBuffer buffer = new StringBuffer();
for (int j = 0; j < 16; j++) {
buffer.append(chars.charAt(r.nextInt(chars.length())));
}
return new StringToken(buffer.toString());
}
private final Token.TokenFactory<String> tokenFactory = new Token.TokenFactory<String>() {
public byte[] toByteArray(Token<String> stringToken)
{
try
{
return stringToken.token.getBytes("UTF-8");
}
catch (UnsupportedEncodingException e)
{
throw new RuntimeException(e);
}
}
public Token<String> fromByteArray(byte[] bytes)
{
try
{
return new StringToken(new String(bytes, "UTF-8"));
}
catch (UnsupportedEncodingException e)
{
throw new RuntimeException(e);
}
}
public Token<String> fromString(String string)
{
return new StringToken(string);
}
};
public Token.TokenFactory<String> getTokenFactory()
{
return tokenFactory;
}
public Token getTokenForKey(String key)
{
return new StringToken(key);
}
}

51
src/org/apache/cassandra/dht/RandomPartitioner.java
View File

@ -22,14 +22,10 @@ import java.math.BigInteger;
import java.util.Comparator;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.utils.GuidGenerator;
/**
* This class generates a MD5 hash of the key. It uses the standard technique
* used in all DHT's.
*
* @author alakshman
*
* This class generates a BigIntegerToken using MD5 hash.
*/
public class RandomPartitioner implements IPartitioner
{
@ -56,14 +52,9 @@ public class RandomPartitioner implements IPartitioner
}
};
public BigInteger hash(String key)
{
return FBUtilities.hash(key);
}
public String decorateKey(String key)
{
return hash(key).toString() + ":" + key;
return FBUtilities.hash(key).toString() + ":" + key;
}
public String undecorateKey(String decoratedKey)
@ -80,4 +71,40 @@ public class RandomPartitioner implements IPartitioner
{
return rcomparator;
}
public BigIntegerToken getDefaultToken()
{
String guid = GuidGenerator.guid();
BigInteger token = FBUtilities.hash(guid);
if ( token.signum() == -1 )
token = token.multiply(BigInteger.valueOf(-1L));
return new BigIntegerToken(token);
}
private final Token.TokenFactory<BigInteger> tokenFactory = new Token.TokenFactory<BigInteger>() {
public byte[] toByteArray(Token<BigInteger> bigIntegerToken)
{
return bigIntegerToken.token.toByteArray();
}
public Token<BigInteger> fromByteArray(byte[] bytes)
{
return new BigIntegerToken(new BigInteger(bytes));
}
public Token<BigInteger> fromString(String string)
{
return new BigIntegerToken(new BigInteger(string));
}
};
public Token.TokenFactory<BigInteger> getTokenFactory()
{
return tokenFactory;
}
public Token getTokenForKey(String key)
{
return new BigIntegerToken(FBUtilities.hash(key));
}
}

81
src/org/apache/cassandra/dht/Range.java
View File

@ -21,17 +21,10 @@ package org.apache.cassandra.dht;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.math.BigInteger;
import org.apache.cassandra.gms.GossipDigest;
import org.apache.cassandra.io.ICompactSerializer;
import org.apache.cassandra.io.IFileReader;
import org.apache.cassandra.io.IFileWriter;
import org.apache.cassandra.net.CompactEndPointSerializationHelper;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.service.StorageService;
@ -53,36 +46,20 @@ public class Range implements Comparable<Range>
return serializer_;
}
public static boolean isKeyInRanges(List<Range> ranges, String key)
{
if(ranges == null )
return false;
for ( Range range : ranges)
{
if(range.contains(StorageService.hash(key)))
{
return true ;
}
}
return false;
}
private final BigInteger left_;
private final BigInteger right_;
public Range(BigInteger left, BigInteger right)
private final Token left_;
private final Token right_;
public Range(Token left, Token right)
{
left_ = left;
right_ = right;
}
/**
* Returns the left endpoint of a range.
* @return left endpoint
*/
public BigInteger left()
public Token left()
{
return left_;
}
@ -91,7 +68,7 @@ public class Range implements Comparable<Range>
* Returns the right endpoint of a range.
* @return right endpoint
*/
public BigInteger right()
public Token right()
{
return right_;
}
@ -102,9 +79,9 @@ public class Range implements Comparable<Range>
* @param bi point in question
* @return true if the point contains within the range else false.
*/
public boolean contains(BigInteger bi)
public boolean contains(Token bi)
{
if ( left_.subtract(right_).signum() > 0 )
if ( left_.compareTo(right_) > 0 )
{
/*
* left is greater than right we are wrapping around.
@ -114,16 +91,16 @@ public class Range implements Comparable<Range>
* (2) k < b -- return true
* (3) b < k < a -- return false
*/
if ( bi.subtract(left_).signum() >= 0 )
if ( bi.compareTo(left_) >= 0 )
return true;
else return right_.subtract(bi).signum() > 0;
else return right_.compareTo(bi) > 0;
}
else if ( left_.subtract(right_).signum() < 0 )
else if ( left_.compareTo(right_) < 0 )
{
/*
* This is the range [a, b) where a < b.
*/
return ( bi.subtract(left_).signum() >= 0 && right_.subtract(bi).signum() >=0 );
return ( bi.compareTo(left_) >= 0 && right_.compareTo(bi) >=0 );
}
else
{
@ -136,9 +113,9 @@ public class Range implements Comparable<Range>
* @param range
* @return
*/
private boolean isWrapAround(Range range)
private static boolean isWrapAround(Range range)
{
return range.left_.subtract(range.right_).signum() > 0;
return range.left_.compareTo(range.right_) > 0;
}
public int compareTo(Range rhs)
@ -156,6 +133,22 @@ public class Range implements Comparable<Range>
return right_.compareTo(rhs.right_);
}
public static boolean isKeyInRanges(String key, List<Range> ranges)
{
assert ranges != null;
Token token = StorageService.token(key);
for (Range range : ranges)
{
if(range.contains(token))
{
return true;
}
}
return false;
}
public boolean equals(Object o)
{
if ( !(o instanceof Range) )
@ -178,15 +171,13 @@ public class Range implements Comparable<Range>
class RangeSerializer implements ICompactSerializer<Range>
{
public void serialize(Range range, DataOutputStream dos) throws IOException
{
dos.writeUTF(range.left().toString());
dos.writeUTF(range.right().toString());
{
Token.serializer().serialize(range.left(), dos);
Token.serializer().serialize(range.right(), dos);
}
public Range deserialize(DataInputStream dis) throws IOException
{
BigInteger left = new BigInteger(dis.readUTF());
BigInteger right = new BigInteger(dis.readUTF());
return new Range(left, right);
return new Range(Token.serializer().deserialize(dis), Token.serializer().deserialize(dis));
}
}

14
src/org/apache/cassandra/dht/StringToken.java Normal file
View File

@ -0,0 +1,14 @@
package org.apache.cassandra.dht;
public class StringToken extends Token<String>
{
protected StringToken(String token)
{
super(token);
}
public int compareTo(Token<String> o)
{
return OrderPreservingPartitioner.collator.compare(this.token, o.token);
}
}

77
src/org/apache/cassandra/dht/Token.java Normal file
View File

@ -0,0 +1,77 @@
package org.apache.cassandra.dht;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import org.apache.cassandra.io.ICompactSerializer;
import org.apache.cassandra.service.StorageService;
public abstract class Token<T extends Comparable> implements Comparable<Token<T>>
{
private static final TokenSerializer serializer = new TokenSerializer();
public static TokenSerializer serializer()
{
return serializer;
}
T token;
protected Token(T token)
{
this.token = token;
}
/**
* This determines the comparison for node destination purposes.
*/
public int compareTo(Token<T> o)
{
return token.compareTo(o.token);
}
public String toString()
{
return "Token(" + token + ")";
}
public boolean equals(Object obj)
{
if (!(obj instanceof Token)) {
return false;
}
return token.equals(((Token)obj).token);
}
public int hashCode()
{
return token.hashCode();
}
public static abstract class TokenFactory<T extends Comparable>
{
public abstract byte[] toByteArray(Token<T> token);
public abstract Token<T> fromByteArray(byte[] bytes);
public abstract Token<T> fromString(String string);
}
public static class TokenSerializer implements ICompactSerializer<Token>
{
public void serialize(Token token, DataOutputStream dos) throws IOException
{
IPartitioner p = StorageService.getPartitioner();
byte[] b = p.getTokenFactory().toByteArray(token);
dos.writeInt(b.length);
dos.write(b);
}
public Token deserialize(DataInputStream dis) throws IOException
{
IPartitioner p = StorageService.getPartitioner();
int size = dis.readInt();
byte[] bytes = new byte[size];
dis.readFully(bytes);
return p.getTokenFactory().fromByteArray(bytes);
}
}
}

22
src/org/apache/cassandra/io/SSTable.java
View File

@ -18,19 +18,31 @@
package org.apache.cassandra.io;
import java.io.*;
import java.util.*;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Hashtable;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import org.apache.log4j.Logger;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.RowMutation;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.utils.BasicUtilities;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.FileUtils;
import org.apache.cassandra.utils.LogUtil;
import org.apache.cassandra.db.RowMutation;
import org.apache.log4j.Logger;
import org.apache.cassandra.dht.IPartitioner;
/**
* This class is built on top of the SequenceFile. It stores

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

@ -1,6 +1,5 @@
package org.apache.cassandra.locator;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@ -8,11 +7,12 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.log4j.Logger;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.service.StorageService;
import org.apache.log4j.Logger;
/**
* This class contains a helper method that will be used by
@ -45,10 +45,10 @@ public abstract class AbstractStrategy implements IReplicaPlacementStrategy
protected EndPoint getNextAvailableEndPoint(EndPoint startPoint, List<EndPoint> topN, List<EndPoint> liveNodes)
{
EndPoint endPoint = null;
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List<BigInteger> tokens = new ArrayList<BigInteger>(tokenToEndPointMap.keySet());
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List tokens = new ArrayList(tokenToEndPointMap.keySet());
Collections.sort(tokens);
BigInteger token = tokenMetadata_.getToken(startPoint);
Token token = tokenMetadata_.getToken(startPoint);
int index = Collections.binarySearch(tokens, token);
if(index < 0)
{
@ -76,7 +76,7 @@ public abstract class AbstractStrategy implements IReplicaPlacementStrategy
* endpoint which is in the top N.
* Get the map of top N to the live nodes currently.
*/
public Map<EndPoint, EndPoint> getHintedStorageEndPoints(BigInteger token)
public Map<EndPoint, EndPoint> getHintedStorageEndPoints(Token token)
{
List<EndPoint> liveList = new ArrayList<EndPoint>();
Map<EndPoint, EndPoint> map = new HashMap<EndPoint, EndPoint>();
@ -107,6 +107,6 @@ public abstract class AbstractStrategy implements IReplicaPlacementStrategy
return map;
}
public abstract EndPoint[] getStorageEndPoints(BigInteger token);
public abstract EndPoint[] getStorageEndPoints(Token token);
}

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

@ -18,9 +18,10 @@
package org.apache.cassandra.locator;
import java.math.BigInteger;
import java.util.Map;
import java.math.BigInteger;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.net.EndPoint;
@ -32,8 +33,8 @@ import org.apache.cassandra.net.EndPoint;
*/
public interface IReplicaPlacementStrategy
{
public EndPoint[] getStorageEndPoints(BigInteger token);
public Map<String, EndPoint[]> getStorageEndPoints(String[] keys);
public EndPoint[] getStorageEndPoints(BigInteger token, Map<BigInteger, EndPoint> tokenToEndPointMap);
public Map<EndPoint, EndPoint> getHintedStorageEndPoints(BigInteger token);
public EndPoint[] getStorageEndPoints(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);
}

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

@ -1,14 +1,15 @@
package org.apache.cassandra.locator;
import java.math.BigInteger;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.math.BigInteger;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.LogUtil;
@ -28,7 +29,7 @@ public class RackAwareStrategy extends AbstractStrategy
super(tokenMetadata);
}
public EndPoint[] getStorageEndPoints(BigInteger token)
public EndPoint[] getStorageEndPoints(Token token)
{
int startIndex;
List<EndPoint> list = new ArrayList<EndPoint>();
@ -36,8 +37,8 @@ public class RackAwareStrategy extends AbstractStrategy
boolean bOtherRack = false;
int foundCount = 0;
int N = DatabaseDescriptor.getReplicationFactor();
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List<BigInteger> tokens = new ArrayList<BigInteger>(tokenToEndPointMap.keySet());
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List tokens = new ArrayList(tokenToEndPointMap.keySet());
Collections.sort(tokens);
int index = Collections.binarySearch(tokens, token);
if(index < 0)
@ -112,14 +113,13 @@ public class RackAwareStrategy extends AbstractStrategy
for ( String key : keys )
{
BigInteger token = StorageService.hash(key);
results.put(key, getStorageEndPoints(token));
results.put(key, getStorageEndPoints(StorageService.token(key)));
}
return results;
}
public EndPoint[] getStorageEndPoints(BigInteger token, Map<BigInteger, EndPoint> tokenToEndPointMap)
public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap)
{
throw new UnsupportedOperationException("This operation is not currently supported");
}

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

@ -1,16 +1,13 @@
package org.apache.cassandra.locator;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.service.StorageService;
@ -23,27 +20,23 @@ import org.apache.cassandra.service.StorageService;
*/
public class RackUnawareStrategy extends AbstractStrategy
{
/* Use this flag to check if initialization is in order. */
private AtomicBoolean initialized_ = new AtomicBoolean(false);
private Map<Range, List<EndPoint>> rangeToEndPointMap_;
public RackUnawareStrategy(TokenMetadata tokenMetadata)
{
super(tokenMetadata);
}
public EndPoint[] getStorageEndPoints(BigInteger token)
public EndPoint[] getStorageEndPoints(Token token)
{
return getStorageEndPoints(token, tokenMetadata_.cloneTokenEndPointMap());
}
public EndPoint[] getStorageEndPoints(BigInteger token, Map<BigInteger, EndPoint> tokenToEndPointMap)
public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap)
{
int startIndex;
List<EndPoint> list = new ArrayList<EndPoint>();
int foundCount = 0;
int N = DatabaseDescriptor.getReplicationFactor();
List<BigInteger> tokens = new ArrayList<BigInteger>(tokenToEndPointMap.keySet());
List tokens = new ArrayList<Token>(tokenToEndPointMap.keySet());
Collections.sort(tokens);
int index = Collections.binarySearch(tokens, token);
if(index < 0)
@ -77,8 +70,7 @@ public class RackUnawareStrategy extends AbstractStrategy
for ( String key : keys )
{
BigInteger token = StorageService.hash(key);
results.put(key, getStorageEndPoints(token));
results.put(key, getStorageEndPoints(StorageService.token(key)));
}
return results;

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

@ -18,18 +18,13 @@
package org.apache.cassandra.locator;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.math.BigInteger;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.cassandra.io.ICompactSerializer;
import org.apache.cassandra.net.CompactEndPointSerializationHelper;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.net.EndPoint;
@ -40,9 +35,9 @@ import org.apache.cassandra.net.EndPoint;
public class TokenMetadata
{
/* Maintains token to endpoint map of every node in the cluster. */
private Map<BigInteger, EndPoint> tokenToEndPointMap_ = new HashMap<BigInteger, EndPoint>();
private Map<Token, EndPoint> tokenToEndPointMap_ = new HashMap<Token, EndPoint>();
/* Maintains a reverse index of endpoint to token in the cluster. */
private Map<EndPoint, BigInteger> endPointToTokenMap_ = new HashMap<EndPoint, BigInteger>();
private Map<EndPoint, Token> endPointToTokenMap_ = new HashMap<EndPoint, Token>();
/* Use this lock for manipulating the token map */
private final ReadWriteLock lock_ = new ReentrantReadWriteLock(true);
@ -51,7 +46,7 @@ public class TokenMetadata
{
}
private TokenMetadata(Map<BigInteger, EndPoint> tokenToEndPointMap, Map<EndPoint, BigInteger> endPointToTokenMap)
private TokenMetadata(Map<Token, EndPoint> tokenToEndPointMap, Map<EndPoint, Token> endPointToTokenMap)
{
tokenToEndPointMap_ = tokenToEndPointMap;
endPointToTokenMap_ = endPointToTokenMap;
@ -59,20 +54,18 @@ public class TokenMetadata
public TokenMetadata cloneMe()
{
Map<BigInteger, EndPoint> tokenToEndPointMap = cloneTokenEndPointMap();
Map<EndPoint, BigInteger> endPointToTokenMap = cloneEndPointTokenMap();
return new TokenMetadata( tokenToEndPointMap, endPointToTokenMap );
return new TokenMetadata(cloneTokenEndPointMap(), cloneEndPointTokenMap());
}
/**
* Update the two maps in an safe mode.
*/
public void update(BigInteger token, EndPoint endpoint)
public void update(Token token, EndPoint endpoint)
{
lock_.writeLock().lock();
try
{
BigInteger oldToken = endPointToTokenMap_.get(endpoint);
Token oldToken = endPointToTokenMap_.get(endpoint);
if ( oldToken != null )
tokenToEndPointMap_.remove(oldToken);
tokenToEndPointMap_.put(token, endpoint);
@ -93,7 +86,7 @@ public class TokenMetadata
lock_.writeLock().lock();
try
{
BigInteger oldToken = endPointToTokenMap_.get(endpoint);
Token oldToken = endPointToTokenMap_.get(endpoint);
if ( oldToken != null )
tokenToEndPointMap_.remove(oldToken);
endPointToTokenMap_.remove(endpoint);
@ -104,7 +97,7 @@ public class TokenMetadata
}
}
public BigInteger getToken(EndPoint endpoint)
public Token getToken(EndPoint endpoint)
{
lock_.readLock().lock();
try
@ -133,12 +126,12 @@ public class TokenMetadata
/*
* Returns a safe clone of tokenToEndPointMap_.
*/
public Map<BigInteger, EndPoint> cloneTokenEndPointMap()
public Map<Token, EndPoint> cloneTokenEndPointMap()
{
lock_.readLock().lock();
try
{
return new HashMap<BigInteger, EndPoint>( tokenToEndPointMap_ );
return new HashMap<Token, EndPoint>( tokenToEndPointMap_ );
}
finally
{
@ -149,12 +142,12 @@ public class TokenMetadata
/*
* Returns a safe clone of endPointTokenMap_.
*/
public Map<EndPoint, BigInteger> cloneEndPointTokenMap()
public Map<EndPoint, Token> cloneEndPointTokenMap()
{
lock_.readLock().lock();
try
{
return new HashMap<EndPoint, BigInteger>( endPointToTokenMap_ );
return new HashMap<EndPoint, Token>( endPointToTokenMap_ );
}
finally
{

20
src/org/apache/cassandra/service/StorageLoadBalancer.java
View File

@ -19,33 +19,30 @@
package org.apache.cassandra.service;
import java.io.Serializable;
import java.math.BigInteger;
import java.util.*;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.log4j.Logger;
import org.apache.cassandra.concurrent.DebuggableScheduledThreadPoolExecutor;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.concurrent.SingleThreadedStage;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.concurrent.ThreadFactoryImpl;
import org.apache.cassandra.dht.LeaveJoinProtocolImpl;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndPointState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.IEndPointStateChangeSubscriber;
import org.apache.cassandra.io.SSTable;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.log4j.Logger;
import org.apache.cassandra.net.*;
import org.apache.cassandra.utils.*;
/*
* The load balancing algorithm here is an implementation of
@ -164,7 +161,6 @@ final class StorageLoadBalancer implements IEndPointStateChangeSubscriber, IComp
if ( isMoveable_.get() )
{
MoveMessage moveMessage = (MoveMessage)message.getMessageBody()[0];
BigInteger targetToken = moveMessage.getTargetToken();
/* Start the leave operation and join the ring at the position specified */
isMoveable_.set(false);
}
@ -396,18 +392,18 @@ final class StorageLoadBalancer implements IEndPointStateChangeSubscriber, IComp
class MoveMessage implements Serializable
{
private BigInteger targetToken_;
private Token targetToken_;
private MoveMessage()
{
}
MoveMessage(BigInteger targetToken)
MoveMessage(Token targetToken)
{
targetToken_ = targetToken;
}
BigInteger getTargetToken()
Token getTargetToken()
{
return targetToken_;
}

144
src/org/apache/cassandra/service/StorageService.java
View File

@ -18,19 +18,32 @@
package org.apache.cassandra.service;
import java.util.*;
import java.io.File;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.io.*;
import java.lang.management.ManagementFactory;
import java.math.BigInteger;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.log4j.Logger;
import org.apache.cassandra.analytics.AnalyticsContext;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.concurrent.MultiThreadedStage;
@ -56,31 +69,31 @@ import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.dht.BootStrapper;
import org.apache.cassandra.dht.BootstrapInitiateMessage;
import org.apache.cassandra.dht.BootstrapMetadataVerbHandler;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.OrderPreservingPartitioner;
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.dht.OrderPreservingHashPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.IEndPointStateChangeSubscriber;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndPointState;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.IEndPointStateChangeSubscriber;
import org.apache.cassandra.locator.EndPointSnitch;
import org.apache.cassandra.locator.IEndPointSnitch;
import org.apache.cassandra.locator.IReplicaPlacementStrategy;
import org.apache.cassandra.locator.RackAwareStrategy;
import org.apache.cassandra.locator.RackUnawareStrategy;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.http.HttpConnection;
import org.apache.cassandra.net.io.StreamContextManager;
import org.apache.cassandra.tools.MembershipCleanerVerbHandler;
import org.apache.cassandra.tools.TokenUpdateVerbHandler;
import org.apache.cassandra.utils.FileUtils;
import org.apache.cassandra.net.http.HttpConnection;
import org.apache.cassandra.locator.IEndPointSnitch;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.locator.IReplicaPlacementStrategy;
import org.apache.cassandra.locator.EndPointSnitch;
import org.apache.cassandra.locator.RackUnawareStrategy;
import org.apache.cassandra.locator.RackAwareStrategy;
import org.apache.cassandra.utils.LogUtil;
import org.apache.cassandra.utils.FileUtils;
import org.apache.cassandra.tools.MembershipCleanerVerbHandler;
import org.apache.log4j.Logger;
@ -88,8 +101,8 @@ import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
/*
@ -127,7 +140,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
public final static String bsMetadataVerbHandler_ = "BS-METADATA-VERB-HANDLER";
public final static String calloutDeployVerbHandler_ = "CALLOUT-DEPLOY-VERB-HANDLER";
public final static String touchVerbHandler_ = "TOUCH-VERB-HANDLER";
public static enum ConsistencyLevel
{
WEAK,
@ -161,9 +174,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
* function used by the system for
* partitioning.
*/
public static BigInteger hash(String key)
public static Token token(String key)
{
return partitioner_.hash(key);
return partitioner_.getTokenForKey(key);
}
public static IPartitioner getPartitioner() {
@ -433,7 +446,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
String hashingStrategy = DatabaseDescriptor.getHashingStrategy();
if (DatabaseDescriptor.ophf_.equalsIgnoreCase(hashingStrategy))
{
partitioner_ = new OrderPreservingHashPartitioner();
partitioner_ = new OrderPreservingPartitioner();
}
else
{
@ -534,7 +547,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
}
/* TODO: remove later */
public void updateTokenMetadata(BigInteger token, EndPoint endpoint)
public void updateTokenMetadata(Token token, EndPoint endpoint)
{
tokenMetadata_.update(token, endpoint);
}
@ -568,10 +581,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
public Map<Range, List<EndPoint>> getRangeToEndPointMap()
{
/* Get the token to endpoint map. */
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Set<BigInteger> tokens = tokenToEndPointMap.keySet();
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
/* All the ranges for the tokens */
Range[] ranges = getAllRanges(tokens);
Range[] ranges = getAllRanges(tokenToEndPointMap.keySet());
return constructRangeToEndPointMap(ranges);
}
@ -601,7 +613,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
* @param tokenToEndPointMap mapping of token to endpoints.
* @return mapping of ranges to the replicas responsible for them.
*/
public Map<Range, List<EndPoint>> constructRangeToEndPointMap(Range[] ranges, Map<BigInteger, EndPoint> tokenToEndPointMap)
public Map<Range, List<EndPoint>> constructRangeToEndPointMap(Range[] ranges, Map<Token, EndPoint> tokenToEndPointMap)
{
logger_.debug("Constructing range to endpoint map ...");
Map<Range, List<EndPoint>> rangeToEndPointMap = new HashMap<Range, List<EndPoint>>();
@ -623,7 +635,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
public Map<EndPoint, List<Range>> constructEndPointToRangesMap()
{
Map<EndPoint, List<Range>> endPointToRangesMap = new HashMap<EndPoint, List<Range>>();
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Collection<EndPoint> mbrs = tokenToEndPointMap.values();
for ( EndPoint mbr : mbrs )
{
@ -644,9 +656,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
ApplicationState nodeIdState = epState.getApplicationState(StorageService.nodeId_);
if (nodeIdState != null)
{
BigInteger newToken = new BigInteger(nodeIdState.getState());
Token newToken = getPartitioner().getTokenFactory().fromString(nodeIdState.getState());
logger_.debug("CHANGE IN STATE FOR " + endpoint + " - has token " + nodeIdState.getState());
BigInteger oldToken = tokenMetadata_.getToken(ep);
Token oldToken = tokenMetadata_.getToken(ep);
if ( oldToken != null )
{
@ -709,7 +721,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
*/
public String getLoadInfo()
{
long diskSpace = FileUtils.getUsedDiskSpace();
long diskSpace = FileUtils.getUsedDiskSpace();
return FileUtils.stringifyFileSize(diskSpace);
}
@ -728,7 +740,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
* This method updates the token on disk and modifies the cached
* StorageMetadata instance. This is only for the local endpoint.
*/
public void updateToken(BigInteger token) throws IOException
public void updateToken(Token token) throws IOException
{
/* update the token on disk */
SystemTable.openSystemTable(SystemTable.name_).updateToken(token);
@ -770,12 +782,12 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
if ( keys.length > 0 )
{
isLoadState_ = true;
BigInteger token = tokenMetadata_.getToken(StorageService.tcpAddr_);
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
BigInteger[] tokens = tokenToEndPointMap.keySet().toArray( new BigInteger[0] );
Token token = tokenMetadata_.getToken(StorageService.tcpAddr_);
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Token[] tokens = tokenToEndPointMap.keySet().toArray(new Token[tokenToEndPointMap.keySet().size()]);
Arrays.sort(tokens);
int index = Arrays.binarySearch(tokens, token) * (keys.length/tokens.length);
BigInteger newToken = hash( keys[index] );
Token newToken = token( keys[index] );
/* update the token */
updateToken(newToken);
}
@ -814,7 +826,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
}
String[] allNodes = nodesToLoad.split(":");
EndPoint[] endpoints = new EndPoint[allNodes.length];
BigInteger[] tokens = new BigInteger[allNodes.length];
Token[] tokens = new Token[allNodes.length];
for ( int i = 0; i < allNodes.length; ++i )
{
@ -850,8 +862,8 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
switch ( mode )
{
case FULL:
BigInteger token = tokenMetadata_.getToken(endpoint);
bootStrapper_.submit( new BootStrapper(new EndPoint[]{endpoint}, new BigInteger[]{token}) );
Token token = tokenMetadata_.getToken(endpoint);
bootStrapper_.submit(new BootStrapper(new EndPoint[]{endpoint}, token));
break;
case HINT:
@ -869,26 +881,15 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
public String getToken(EndPoint ep)
{
EndPoint ep2 = new EndPoint(ep.getHost(), DatabaseDescriptor.getStoragePort());
BigInteger token = tokenMetadata_.getToken(ep2);
return ( token == null ) ? BigInteger.ZERO.toString() : token.toString();
Token token = tokenMetadata_.getToken(ep2);
// if there is no token for an endpoint, return an empty string to denote that
return ( token == null ) ? "" : token.toString();
}
public String getToken()
{
return tokenMetadata_.getToken(StorageService.tcpAddr_).toString();
}
public void updateToken(String token)
{
try
{
updateToken(new BigInteger(token));
}
catch ( IOException ex )
{
logger_.debug(LogUtil.throwableToString(ex));
}
}
public String getLiveNodes()
{
@ -971,9 +972,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
*/
EndPoint getPredecessor(EndPoint ep)
{
BigInteger token = tokenMetadata_.getToken(ep);
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List<BigInteger> tokens = new ArrayList<BigInteger>(tokenToEndPointMap.keySet());
Token token = tokenMetadata_.getToken(ep);
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List tokens = new ArrayList<Token>(tokenToEndPointMap.keySet());
Collections.sort(tokens);
int index = Collections.binarySearch(tokens, token);
return (index == 0) ? tokenToEndPointMap.get(tokens
@ -987,9 +988,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
*/
public EndPoint getSuccessor(EndPoint ep)
{
BigInteger token = tokenMetadata_.getToken(ep);
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List<BigInteger> tokens = new ArrayList<BigInteger>(tokenToEndPointMap.keySet());
Token token = tokenMetadata_.getToken(ep);
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List tokens = new ArrayList<Token>(tokenToEndPointMap.keySet());
Collections.sort(tokens);
int index = Collections.binarySearch(tokens, token);
return (index == (tokens.size() - 1)) ? tokenToEndPointMap
@ -1004,9 +1005,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
*/
public Range getPrimaryRangeForEndPoint(EndPoint ep)
{
BigInteger right = tokenMetadata_.getToken(ep);
Token right = tokenMetadata_.getToken(ep);
EndPoint predecessor = getPredecessor(ep);
BigInteger left = tokenMetadata_.getToken(predecessor);
Token left = tokenMetadata_.getToken(predecessor);
return new Range(left, right);
}
@ -1038,8 +1039,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
*/
public Range[] getAllRanges()
{
Set<BigInteger> allTokens = tokenMetadata_.cloneTokenEndPointMap().keySet();
return getAllRanges( allTokens );
return getAllRanges(tokenMetadata_.cloneTokenEndPointMap().keySet());
}
/**
@ -1048,10 +1048,10 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
* ranges.
* @return ranges in sorted order
*/
public Range[] getAllRanges(Set<BigInteger> tokens)
public Range[] getAllRanges(Set<Token> tokens)
{
List<Range> ranges = new ArrayList<Range>();
List<BigInteger> allTokens = new ArrayList<BigInteger>(tokens);
List<Token> allTokens = new ArrayList<Token>(tokens);
Collections.sort(allTokens);
int size = allTokens.size();
for ( int i = 1; i < size; ++i )
@ -1074,9 +1074,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
public EndPoint getPrimary(String key)
{
EndPoint endpoint = StorageService.tcpAddr_;
BigInteger token = hash(key);
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List<BigInteger> tokens = new ArrayList<BigInteger>(tokenToEndPointMap.keySet());
Token token = token(key);
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List tokens = new ArrayList<Token>(tokenToEndPointMap.keySet());
if (tokens.size() > 0)
{
Collections.sort(tokens);
@ -1122,7 +1122,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
*/
public EndPoint[] getNStorageEndPoint(String key)
{
BigInteger token = hash(key);
Token token = token(key);
return nodePicker_.getStorageEndPoints(token);
}
@ -1162,7 +1162,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
*/
public Map<EndPoint, EndPoint> getNStorageEndPointMap(String key)
{
BigInteger token = hash(key);
Token token = token(key);
return nodePicker_.getHintedStorageEndPoints(token);
}
@ -1172,7 +1172,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
*
* param @ token - position on the ring
*/
public EndPoint[] getNStorageEndPoint(BigInteger token)
public EndPoint[] getNStorageEndPoint(Token token)
{
return nodePicker_.getStorageEndPoints(token);
}
@ -1185,7 +1185,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
* param @ token - position on the ring
* param @ tokens - w/o the following tokens in the token list
*/
protected EndPoint[] getNStorageEndPoint(BigInteger token, Map<BigInteger, EndPoint> tokenToEndPointMap)
protected EndPoint[] getNStorageEndPoint(Token token, Map<Token, EndPoint> tokenToEndPointMap)
{
return nodePicker_.getStorageEndPoints(token, tokenToEndPointMap);
}

5
src/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -43,11 +43,6 @@ public interface StorageServiceMBean
*/
public void loadAll(String nodes);
/**
* This method is used only for debug purpose.
*/
public void updateToken(String token);
/**
*
*/

7
src/org/apache/cassandra/service/TokenUpdateVerbHandler.java
View File

@ -19,12 +19,13 @@
package org.apache.cassandra.service;
import java.io.IOException;
import java.math.BigInteger;
import org.apache.log4j.Logger;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.utils.LogUtil;
import org.apache.log4j.Logger;
/**
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
@ -37,7 +38,7 @@ public class TokenUpdateVerbHandler implements IVerbHandler
public void doVerb(Message message)
{
byte[] body = (byte[])message.getMessageBody()[0];
BigInteger token = new BigInteger(body);
Token token = StorageService.getPartitioner().getTokenFactory().fromByteArray(body);
try
{
logger_.info("Updating the token to [" + token + "]");

13
src/org/apache/cassandra/tools/MembershipCleanerVerbHandler.java
View File

@ -18,25 +18,20 @@
package org.apache.cassandra.tools;
import java.util.*;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.math.BigInteger;
import java.util.Map;
import java.util.Set;
import org.apache.log4j.Logger;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tools.TokenUpdater.TokenInfoMessage;
import org.apache.cassandra.utils.LogUtil;
import org.apache.log4j.Logger;
import org.apache.cassandra.io.*;
import org.apache.cassandra.config.*;
/**
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )

17
src/org/apache/cassandra/tools/TokenUpdateVerbHandler.java
View File

@ -21,20 +21,20 @@ package org.apache.cassandra.tools;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.math.BigInteger;
import java.util.Map;
import java.util.Set;
import org.apache.log4j.Logger;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tools.TokenUpdater.TokenInfoMessage;
import org.apache.cassandra.utils.LogUtil;
/**
@ -54,9 +54,8 @@ public class TokenUpdateVerbHandler implements IVerbHandler
DataInputBuffer bufIn = new DataInputBuffer();
bufIn.reset(body, body.length);
/* Deserialize to get the token for this endpoint. */
TokenUpdater.TokenInfoMessage tiMessage = TokenUpdater.TokenInfoMessage.serializer().deserialize(bufIn);
BigInteger token = tiMessage.getToken();
Token token = Token.serializer().deserialize(bufIn);
logger_.info("Updating the token to [" + token + "]");
StorageService.instance().updateToken(token);
@ -66,19 +65,19 @@ public class TokenUpdateVerbHandler implements IVerbHandler
logger_.debug("Number of nodes in the header " + headers.size());
Set<String> nodes = headers.keySet();
IPartitioner p = StorageService.getPartitioner();
for ( String node : nodes )
{
logger_.debug("Processing node " + node);
byte[] bytes = headers.remove(node);
/* Send a message to this node to update its token to the one retreived. */
EndPoint target = new EndPoint(node, DatabaseDescriptor.getStoragePort());
token = new BigInteger(bytes);
token = p.getTokenFactory().fromByteArray(bytes);
/* Reset the new TokenInfoMessage */
tiMessage = new TokenUpdater.TokenInfoMessage(target, token );
/* Reset the new Message */
ByteArrayOutputStream bos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(bos);
TokenInfoMessage.serializer().serialize(tiMessage, dos);
Token.serializer().serialize(token, dos);
message.setMessageBody(new Object[]{bos.toByteArray()});
logger_.debug("Sending a token update message to " + target + " to update it to " + token);

80
src/org/apache/cassandra/tools/TokenUpdater.java
View File

@ -20,16 +20,12 @@ package org.apache.cassandra.tools;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Serializable;
import java.math.BigInteger;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.cassandra.io.ICompactSerializer;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
@ -50,16 +46,17 @@ public class TokenUpdater
}
String ipPort = args[0];
String token = args[1];
IPartitioner p = StorageService.getPartitioner();
Token token = p.getTokenFactory().fromString(args[1]);
String file = args[2];
String[] ipPortPair = ipPort.split(":");
EndPoint target = new EndPoint(ipPortPair[0], Integer.valueOf(ipPortPair[1]));
TokenInfoMessage tiMessage = new TokenInfoMessage( target, new BigInteger(token) );
ByteArrayOutputStream bos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(bos);
TokenInfoMessage.serializer().serialize(tiMessage, dos);
Token.serializer().serialize(token, dos);
/* Construct the token update message to be sent */
Message tokenUpdateMessage = new Message( new EndPoint(FBUtilities.getHostAddress(), port_), "", StorageService.tokenVerbHandler_, new Object[]{bos.toByteArray()} );
@ -70,8 +67,8 @@ public class TokenUpdater
{
String[] nodeTokenPair = line.split(" ");
/* Add the node and the token pair into the header of this message. */
BigInteger nodeToken = new BigInteger(nodeTokenPair[1]);
tokenUpdateMessage.addHeader(nodeTokenPair[0], nodeToken.toByteArray());
Token nodeToken = p.getTokenFactory().fromString(nodeTokenPair[1]);
tokenUpdateMessage.addHeader(nodeTokenPair[0], p.getTokenFactory().toByteArray(nodeToken));
}
System.out.println("Sending a token update message to " + target);
@ -79,64 +76,5 @@ public class TokenUpdater
Thread.sleep(TokenUpdater.waitTime_);
System.out.println("Done sending the update message");
}
public static class TokenInfoMessage implements Serializable
{
private static ICompactSerializer<TokenInfoMessage> serializer_;
private static AtomicInteger idGen_ = new AtomicInteger(0);
static
{
serializer_ = new TokenInfoMessageSerializer();
}
static ICompactSerializer<TokenInfoMessage> serializer()
{
return serializer_;
}
private EndPoint target_;
private BigInteger token_;
TokenInfoMessage(EndPoint target, BigInteger token)
{
target_ = target;
token_ = token;
}
EndPoint getTarget()
{
return target_;
}
BigInteger getToken()
{
return token_;
}
}
public static class TokenInfoMessageSerializer implements ICompactSerializer<TokenInfoMessage>
{
public void serialize(TokenInfoMessage tiMessage, DataOutputStream dos) throws IOException
{
byte[] node = EndPoint.toBytes( tiMessage.getTarget() );
dos.writeInt(node.length);
dos.write(node);
byte[] token = tiMessage.getToken().toByteArray();
dos.writeInt( token.length );
dos.write(token);
}
public TokenInfoMessage deserialize(DataInputStream dis) throws IOException
{
byte[] target = new byte[dis.readInt()];
dis.readFully(target);
byte[] token = new byte[dis.readInt()];
dis.readFully(token);
return new TokenInfoMessage(EndPoint.fromBytes(target), new BigInteger(token));
}
}
}

2
src/org/apache/cassandra/utils/FastObjectHash.java
View File

@ -187,7 +187,7 @@ abstract public class FastObjectHash<T> extends FastHash
}
else
{ // already FULL or REMOVED, must probe
// compute the double hash
// compute the double token
final int probe = 1 + (hash % (length - 2));
// if the slot we landed on is FULL (but not removed), probe

2
test/unit/org/apache/cassandra/db/SystemTableTest.java
View File

@ -9,6 +9,6 @@ import java.io.IOException;
public class SystemTableTest extends ServerTest {
@Test
public void testMain() throws IOException {
SystemTable.openSystemTable(SystemTable.cfName_).updateToken( StorageService.hash("503545744:0") );
SystemTable.openSystemTable(SystemTable.cfName_).updateToken( StorageService.token("503545744:0") );
}
}

18
test/unit/org/apache/cassandra/dht/RangeTest.java
View File

@ -2,19 +2,17 @@ package org.apache.cassandra.dht;
import org.testng.annotations.Test;
import java.math.BigInteger;
public class RangeTest {
@Test
public void testRange() {
Range left = new Range(new BigInteger("0"), new BigInteger("100"));
assert left.contains(new BigInteger("10"));
assert !left.contains(new BigInteger("-1"));
assert !left.contains(new BigInteger("101"));
Range left = new Range(new BigIntegerToken("0"), new BigIntegerToken("100"));
assert left.contains(new BigIntegerToken("10"));
assert !left.contains(new BigIntegerToken("-1"));
assert !left.contains(new BigIntegerToken("101"));
Range right = new Range(new BigInteger("100"), new BigInteger("0"));
assert right.contains(new BigInteger("200"));
assert right.contains(new BigInteger("-10"));
assert !right.contains(new BigInteger("1"));
Range right = new Range(new BigIntegerToken("100"), new BigIntegerToken("0"));
assert right.contains(new BigIntegerToken("200"));
assert right.contains(new BigIntegerToken("-10"));
assert !right.contains(new BigIntegerToken("1"));
}
}