diff --git a/src/org/apache/cassandra/db/ColumnFamilyStore.java b/src/org/apache/cassandra/db/ColumnFamilyStore.java index 612722b7d9..8727a46f1a 100644 --- a/src/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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; diff --git a/src/org/apache/cassandra/db/DBManager.java b/src/org/apache/cassandra/db/DBManager.java index fa7849b609..009ca06968 100644 --- a/src/org/apache/cassandra/db/DBManager.java +++ b/src/org/apache/cassandra/db/DBManager.java @@ -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); diff --git a/src/org/apache/cassandra/db/FileStruct.java b/src/org/apache/cassandra/db/FileStruct.java index 84e60324cf..69c6dc0ee8 100644 --- a/src/org/apache/cassandra/db/FileStruct.java +++ b/src/org/apache/cassandra/db/FileStruct.java @@ -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 diff --git a/src/org/apache/cassandra/db/SystemTable.java b/src/org/apache/cassandra/db/SystemTable.java index 23999a0ff1..a7d6fa2682 100644 --- a/src/org/apache/cassandra/db/SystemTable.java +++ b/src/org/apache/cassandra/db/SystemTable.java @@ -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 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 ranges = new ArrayList(); - ranges.add( new Range(new BigInteger("1218069462158869448693347920504606362273788442553"), new BigInteger("1092770595533781724218060956188429069")) ); - if ( Range.isKeyInRanges(ranges, "304700067:0") ) - { - System.out.println("Done"); - } - */ } } diff --git a/src/org/apache/cassandra/dht/BigIntegerToken.java b/src/org/apache/cassandra/dht/BigIntegerToken.java new file mode 100644 index 0000000000..2615b2d9b5 --- /dev/null +++ b/src/org/apache/cassandra/dht/BigIntegerToken.java @@ -0,0 +1,16 @@ +package org.apache.cassandra.dht; + +import java.math.BigInteger; + +public class BigIntegerToken extends Token +{ + public BigIntegerToken(BigInteger token) + { + super(token); + } + + // convenience method for testing + public BigIntegerToken(String token) { + this(new BigInteger(token)); + } +} diff --git a/src/org/apache/cassandra/dht/BootStrapper.java b/src/org/apache/cassandra/dht/BootStrapper.java index 7dc29da322..618f0532ae 100644 --- a/src/org/apache/cassandra/dht/BootStrapper.java +++ b/src/org/apache/cassandra/dht/BootStrapper.java @@ -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 filters_ = new ArrayList(); - 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 tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); /* remove the tokens associated with the endpoints being bootstrapped */ - for ( BigInteger token : tokens_ ) + for (Token token : tokens_) { tokenToEndPointMap.remove(token); } - Set oldTokens = new HashSet( tokenToEndPointMap.keySet() ); + Set oldTokens = new HashSet( tokenToEndPointMap.keySet() ); Range[] oldRanges = StorageService.instance().getAllRanges(oldTokens); logger_.debug("Total number of old ranges " + oldRanges.length); /* diff --git a/src/org/apache/cassandra/dht/IPartitioner.java b/src/org/apache/cassandra/dht/IPartitioner.java index e298b227d7..fbffce3364 100644 --- a/src/org/apache/cassandra/dht/IPartitioner.java +++ b/src/org/apache/cassandra/dht/IPartitioner.java @@ -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 getDecoratedKeyComparator(); public Comparator getReverseDecoratedKeyComparator(); + + public Token getTokenForKey(String key); + + public Token getDefaultToken(); + + public Token.TokenFactory getTokenFactory(); } diff --git a/src/org/apache/cassandra/dht/LeaveJoinProtocolHelper.java b/src/org/apache/cassandra/dht/LeaveJoinProtocolHelper.java index 1faeb88535..5f3fb1dc20 100644 --- a/src/org/apache/cassandra/dht/LeaveJoinProtocolHelper.java +++ b/src/org/apache/cassandra/dht/LeaveJoinProtocolHelper.java @@ -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> getRangeSplitRangeMapping(Range[] oldRanges, BigInteger[] allTokens) + protected static Map> getRangeSplitRangeMapping(Range[] oldRanges, Token[] allTokens) { Map> splitRanges = new HashMap>(); - 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 subRanges = splitRanges.get(prevRange); @@ -65,7 +65,7 @@ class LeaveJoinProtocolHelper prevRange = oldRange; prevToken = oldRange.left(); - for ( BigInteger token : tokens ) + for (Token token : tokens) { List subRanges = splitRanges.get(oldRange); if ( oldRange.contains(token) ) diff --git a/src/org/apache/cassandra/dht/LeaveJoinProtocolImpl.java b/src/org/apache/cassandra/dht/LeaveJoinProtocolImpl.java index f803ec78ce..08c8429065 100644 --- a/src/org/apache/cassandra/dht/LeaveJoinProtocolImpl.java +++ b/src/org/apache/cassandra/dht/LeaveJoinProtocolImpl.java @@ -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 tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); /* copy the endpoint to token map */ - Map endpointToTokenMap = tokenMetadata_.cloneEndPointTokenMap(); + Map endpointToTokenMap = tokenMetadata_.cloneEndPointTokenMap(); - Set oldTokens = new HashSet( tokenToEndPointMap.keySet() ); + Set oldTokens = new HashSet( 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> oldRangeToEndPointMap = StorageService.instance().constructRangeToEndPointMap(oldRanges); /* Remove the tokens of the nodes leaving the ring */ - Set tokens = getTokensForLeavingNodes(); + Set tokens = getTokensForLeavingNodes(); oldTokens.removeAll(tokens); Range[] rangesAfterNodesLeave = StorageService.instance().getAllRanges(oldTokens); /* Get expanded range to initial range mapping */ Map> 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 getTokensForLeavingNodes() + private Set getTokensForLeavingNodes() { - Set tokens = new HashSet(); + Set tokens = new HashSet(); 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(); } } diff --git a/src/org/apache/cassandra/dht/OrderPreservingHashPartitioner.java b/src/org/apache/cassandra/dht/OrderPreservingHashPartitioner.java deleted file mode 100644 index 9af1584cfb..0000000000 --- a/src/org/apache/cassandra/dht/OrderPreservingHashPartitioner.java +++ /dev/null @@ -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 comparator = new Comparator() - { - public int compare(String o1, String o2) - { - return o1.compareTo(o2); - } - }; - private static final Comparator rcomparator = new Comparator() - { - 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 getReverseDecoratedKeyComparator() - { - return rcomparator; - } - - public Comparator getDecoratedKeyComparator() - { - return comparator; - } -} diff --git a/src/org/apache/cassandra/dht/OrderPreservingPartitioner.java b/src/org/apache/cassandra/dht/OrderPreservingPartitioner.java new file mode 100644 index 0000000000..833a6329d7 --- /dev/null +++ b/src/org/apache/cassandra/dht/OrderPreservingPartitioner.java @@ -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 comparator = new Comparator() { + public int compare(String o1, String o2) + { + return collator.compare(o1, o2); + } + }; + private static final Comparator reverseComparator = new Comparator() { + 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 getDecoratedKeyComparator() + { + return comparator; + } + + public Comparator 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 tokenFactory = new Token.TokenFactory() { + public byte[] toByteArray(Token stringToken) + { + try + { + return stringToken.token.getBytes("UTF-8"); + } + catch (UnsupportedEncodingException e) + { + throw new RuntimeException(e); + } + } + + public Token fromByteArray(byte[] bytes) + { + try + { + return new StringToken(new String(bytes, "UTF-8")); + } + catch (UnsupportedEncodingException e) + { + throw new RuntimeException(e); + } + } + + public Token fromString(String string) + { + return new StringToken(string); + } + }; + + public Token.TokenFactory getTokenFactory() + { + return tokenFactory; + } + + public Token getTokenForKey(String key) + { + return new StringToken(key); + } +} diff --git a/src/org/apache/cassandra/dht/RandomPartitioner.java b/src/org/apache/cassandra/dht/RandomPartitioner.java index 9234068fb3..f95e556e88 100644 --- a/src/org/apache/cassandra/dht/RandomPartitioner.java +++ b/src/org/apache/cassandra/dht/RandomPartitioner.java @@ -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 tokenFactory = new Token.TokenFactory() { + public byte[] toByteArray(Token bigIntegerToken) + { + return bigIntegerToken.token.toByteArray(); + } + + public Token fromByteArray(byte[] bytes) + { + return new BigIntegerToken(new BigInteger(bytes)); + } + + public Token fromString(String string) + { + return new BigIntegerToken(new BigInteger(string)); + } + }; + + public Token.TokenFactory getTokenFactory() + { + return tokenFactory; + } + + public Token getTokenForKey(String key) + { + return new BigIntegerToken(FBUtilities.hash(key)); + } } \ No newline at end of file diff --git a/src/org/apache/cassandra/dht/Range.java b/src/org/apache/cassandra/dht/Range.java index b4313f65f6..738fff5de4 100644 --- a/src/org/apache/cassandra/dht/Range.java +++ b/src/org/apache/cassandra/dht/Range.java @@ -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 return serializer_; } - public static boolean isKeyInRanges(List 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 * 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 * @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 * (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 * @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 return right_.compareTo(rhs.right_); } + + public static boolean isKeyInRanges(String key, List 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 class RangeSerializer implements ICompactSerializer { 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)); } } diff --git a/src/org/apache/cassandra/dht/StringToken.java b/src/org/apache/cassandra/dht/StringToken.java new file mode 100644 index 0000000000..a1939ddac7 --- /dev/null +++ b/src/org/apache/cassandra/dht/StringToken.java @@ -0,0 +1,14 @@ +package org.apache.cassandra.dht; + +public class StringToken extends Token +{ + protected StringToken(String token) + { + super(token); + } + + public int compareTo(Token o) + { + return OrderPreservingPartitioner.collator.compare(this.token, o.token); + } +} diff --git a/src/org/apache/cassandra/dht/Token.java b/src/org/apache/cassandra/dht/Token.java new file mode 100644 index 0000000000..03e3c334de --- /dev/null +++ b/src/org/apache/cassandra/dht/Token.java @@ -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 implements Comparable> +{ + 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 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 + { + public abstract byte[] toByteArray(Token token); + public abstract Token fromByteArray(byte[] bytes); + public abstract Token fromString(String string); + } + + public static class TokenSerializer implements ICompactSerializer + { + 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); + } + } +} diff --git a/src/org/apache/cassandra/io/SSTable.java b/src/org/apache/cassandra/io/SSTable.java index 71b7171587..2a8070a777 100644 --- a/src/org/apache/cassandra/io/SSTable.java +++ b/src/org/apache/cassandra/io/SSTable.java @@ -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 diff --git a/src/org/apache/cassandra/locator/AbstractStrategy.java b/src/org/apache/cassandra/locator/AbstractStrategy.java index 51da847642..62ac217838 100644 --- a/src/org/apache/cassandra/locator/AbstractStrategy.java +++ b/src/org/apache/cassandra/locator/AbstractStrategy.java @@ -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 topN, List liveNodes) { EndPoint endPoint = null; - Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - List tokens = new ArrayList(tokenToEndPointMap.keySet()); + Map 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 getHintedStorageEndPoints(BigInteger token) + public Map getHintedStorageEndPoints(Token token) { List liveList = new ArrayList(); Map map = new HashMap(); @@ -107,6 +107,6 @@ public abstract class AbstractStrategy implements IReplicaPlacementStrategy return map; } - public abstract EndPoint[] getStorageEndPoints(BigInteger token); + public abstract EndPoint[] getStorageEndPoints(Token token); } diff --git a/src/org/apache/cassandra/locator/IReplicaPlacementStrategy.java b/src/org/apache/cassandra/locator/IReplicaPlacementStrategy.java index b2913ef511..f550683cf2 100644 --- a/src/org/apache/cassandra/locator/IReplicaPlacementStrategy.java +++ b/src/org/apache/cassandra/locator/IReplicaPlacementStrategy.java @@ -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 getStorageEndPoints(String[] keys); - public EndPoint[] getStorageEndPoints(BigInteger token, Map tokenToEndPointMap); - public Map getHintedStorageEndPoints(BigInteger token); + public EndPoint[] getStorageEndPoints(Token token); + public Map getStorageEndPoints(String[] keys); + public EndPoint[] getStorageEndPoints(Token token, Map tokenToEndPointMap); + public Map getHintedStorageEndPoints(Token token); } diff --git a/src/org/apache/cassandra/locator/RackAwareStrategy.java b/src/org/apache/cassandra/locator/RackAwareStrategy.java index 986bd16103..460dc30686 100644 --- a/src/org/apache/cassandra/locator/RackAwareStrategy.java +++ b/src/org/apache/cassandra/locator/RackAwareStrategy.java @@ -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 list = new ArrayList(); @@ -36,8 +37,8 @@ public class RackAwareStrategy extends AbstractStrategy boolean bOtherRack = false; int foundCount = 0; int N = DatabaseDescriptor.getReplicationFactor(); - Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - List tokens = new ArrayList(tokenToEndPointMap.keySet()); + Map 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 tokenToEndPointMap) + + public EndPoint[] getStorageEndPoints(Token token, Map tokenToEndPointMap) { throw new UnsupportedOperationException("This operation is not currently supported"); } diff --git a/src/org/apache/cassandra/locator/RackUnawareStrategy.java b/src/org/apache/cassandra/locator/RackUnawareStrategy.java index a88986c0a8..084ba52639 100644 --- a/src/org/apache/cassandra/locator/RackUnawareStrategy.java +++ b/src/org/apache/cassandra/locator/RackUnawareStrategy.java @@ -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> 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 tokenToEndPointMap) + public EndPoint[] getStorageEndPoints(Token token, Map tokenToEndPointMap) { int startIndex; List list = new ArrayList(); int foundCount = 0; int N = DatabaseDescriptor.getReplicationFactor(); - List tokens = new ArrayList(tokenToEndPointMap.keySet()); + List tokens = new ArrayList(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; diff --git a/src/org/apache/cassandra/locator/TokenMetadata.java b/src/org/apache/cassandra/locator/TokenMetadata.java index 45e4408f9b..82129cfc61 100644 --- a/src/org/apache/cassandra/locator/TokenMetadata.java +++ b/src/org/apache/cassandra/locator/TokenMetadata.java @@ -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 tokenToEndPointMap_ = new HashMap(); + private Map tokenToEndPointMap_ = new HashMap(); /* Maintains a reverse index of endpoint to token in the cluster. */ - private Map endPointToTokenMap_ = new HashMap(); + private Map endPointToTokenMap_ = new HashMap(); /* 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 tokenToEndPointMap, Map endPointToTokenMap) + private TokenMetadata(Map tokenToEndPointMap, Map endPointToTokenMap) { tokenToEndPointMap_ = tokenToEndPointMap; endPointToTokenMap_ = endPointToTokenMap; @@ -59,20 +54,18 @@ public class TokenMetadata public TokenMetadata cloneMe() { - Map tokenToEndPointMap = cloneTokenEndPointMap(); - Map 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 cloneTokenEndPointMap() + public Map cloneTokenEndPointMap() { lock_.readLock().lock(); try { - return new HashMap( tokenToEndPointMap_ ); + return new HashMap( tokenToEndPointMap_ ); } finally { @@ -149,12 +142,12 @@ public class TokenMetadata /* * Returns a safe clone of endPointTokenMap_. */ - public Map cloneEndPointTokenMap() + public Map cloneEndPointTokenMap() { lock_.readLock().lock(); try { - return new HashMap( endPointToTokenMap_ ); + return new HashMap( endPointToTokenMap_ ); } finally { diff --git a/src/org/apache/cassandra/service/StorageLoadBalancer.java b/src/org/apache/cassandra/service/StorageLoadBalancer.java index 3e6f4ab223..1c82777027 100644 --- a/src/org/apache/cassandra/service/StorageLoadBalancer.java +++ b/src/org/apache/cassandra/service/StorageLoadBalancer.java @@ -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_; } diff --git a/src/org/apache/cassandra/service/StorageService.java b/src/org/apache/cassandra/service/StorageService.java index cccbb1f84e..9988762a9b 100644 --- a/src/org/apache/cassandra/service/StorageService.java +++ b/src/org/apache/cassandra/service/StorageService.java @@ -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> getRangeToEndPointMap() { /* Get the token to endpoint map. */ - Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - Set tokens = tokenToEndPointMap.keySet(); + Map 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> constructRangeToEndPointMap(Range[] ranges, Map tokenToEndPointMap) + public Map> constructRangeToEndPointMap(Range[] ranges, Map tokenToEndPointMap) { logger_.debug("Constructing range to endpoint map ..."); Map> rangeToEndPointMap = new HashMap>(); @@ -623,7 +635,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto public Map> constructEndPointToRangesMap() { Map> endPointToRangesMap = new HashMap>(); - Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); Collection 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 tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - BigInteger[] tokens = tokenToEndPointMap.keySet().toArray( new BigInteger[0] ); + Token token = tokenMetadata_.getToken(StorageService.tcpAddr_); + Map 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 tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - List tokens = new ArrayList(tokenToEndPointMap.keySet()); + Token token = tokenMetadata_.getToken(ep); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + List tokens = new ArrayList(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 tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - List tokens = new ArrayList(tokenToEndPointMap.keySet()); + Token token = tokenMetadata_.getToken(ep); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + List tokens = new ArrayList(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 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 tokens) + public Range[] getAllRanges(Set tokens) { List ranges = new ArrayList(); - List allTokens = new ArrayList(tokens); + List allTokens = new ArrayList(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 tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - List tokens = new ArrayList(tokenToEndPointMap.keySet()); + Token token = token(key); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + List tokens = new ArrayList(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 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 tokenToEndPointMap) + protected EndPoint[] getNStorageEndPoint(Token token, Map tokenToEndPointMap) { return nodePicker_.getStorageEndPoints(token, tokenToEndPointMap); } diff --git a/src/org/apache/cassandra/service/StorageServiceMBean.java b/src/org/apache/cassandra/service/StorageServiceMBean.java index 3995af7707..1acbcfdb3a 100644 --- a/src/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/org/apache/cassandra/service/StorageServiceMBean.java @@ -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); - /** * */ diff --git a/src/org/apache/cassandra/service/TokenUpdateVerbHandler.java b/src/org/apache/cassandra/service/TokenUpdateVerbHandler.java index 90384e366c..6dcaa31f40 100644 --- a/src/org/apache/cassandra/service/TokenUpdateVerbHandler.java +++ b/src/org/apache/cassandra/service/TokenUpdateVerbHandler.java @@ -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 + "]"); diff --git a/src/org/apache/cassandra/tools/MembershipCleanerVerbHandler.java b/src/org/apache/cassandra/tools/MembershipCleanerVerbHandler.java index 24bf152509..9b3972fb36 100644 --- a/src/org/apache/cassandra/tools/MembershipCleanerVerbHandler.java +++ b/src/org/apache/cassandra/tools/MembershipCleanerVerbHandler.java @@ -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 ) diff --git a/src/org/apache/cassandra/tools/TokenUpdateVerbHandler.java b/src/org/apache/cassandra/tools/TokenUpdateVerbHandler.java index 94029b00b7..9d52a1d4ef 100644 --- a/src/org/apache/cassandra/tools/TokenUpdateVerbHandler.java +++ b/src/org/apache/cassandra/tools/TokenUpdateVerbHandler.java @@ -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 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); diff --git a/src/org/apache/cassandra/tools/TokenUpdater.java b/src/org/apache/cassandra/tools/TokenUpdater.java index 391df99a04..f78c625eed 100644 --- a/src/org/apache/cassandra/tools/TokenUpdater.java +++ b/src/org/apache/cassandra/tools/TokenUpdater.java @@ -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 serializer_; - private static AtomicInteger idGen_ = new AtomicInteger(0); - - static - { - serializer_ = new TokenInfoMessageSerializer(); - } - - static ICompactSerializer 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 - { - 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)); - } - } } diff --git a/src/org/apache/cassandra/utils/FastObjectHash.java b/src/org/apache/cassandra/utils/FastObjectHash.java index c46e4031f0..b434c6a499 100644 --- a/src/org/apache/cassandra/utils/FastObjectHash.java +++ b/src/org/apache/cassandra/utils/FastObjectHash.java @@ -187,7 +187,7 @@ abstract public class FastObjectHash 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 diff --git a/test/unit/org/apache/cassandra/db/SystemTableTest.java b/test/unit/org/apache/cassandra/db/SystemTableTest.java index 5adaabbcb0..7572bbc433 100644 --- a/test/unit/org/apache/cassandra/db/SystemTableTest.java +++ b/test/unit/org/apache/cassandra/db/SystemTableTest.java @@ -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") ); } } diff --git a/test/unit/org/apache/cassandra/dht/RangeTest.java b/test/unit/org/apache/cassandra/dht/RangeTest.java index 38c47b0499..203ee5bb9d 100644 --- a/test/unit/org/apache/cassandra/dht/RangeTest.java +++ b/test/unit/org/apache/cassandra/dht/RangeTest.java @@ -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")); } }