diff --git a/src/org/apache/cassandra/service/CassandraServer.java b/src/org/apache/cassandra/service/CassandraServer.java index 5fe97ba2e4..4f4efd0673 100644 --- a/src/org/apache/cassandra/service/CassandraServer.java +++ b/src/org/apache/cassandra/service/CassandraServer.java @@ -18,29 +18,27 @@ package org.apache.cassandra.service; -import java.io.BufferedInputStream; -import java.io.FileInputStream; -import java.io.IOException; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Iterator; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.Arrays; -import java.util.concurrent.TimeoutException; -import java.util.concurrent.TimeUnit; - -import org.apache.log4j.Logger; - +import com.facebook.thrift.*; +import com.facebook.thrift.server.*; +import com.facebook.thrift.server.TThreadPoolServer.Options; +import com.facebook.thrift.transport.*; +import com.facebook.thrift.protocol.*; import com.facebook.fb303.FacebookBase; import com.facebook.fb303.fb_status; -import com.facebook.thrift.TException; -import com.facebook.thrift.protocol.TBinaryProtocol; -import com.facebook.thrift.protocol.TProtocolFactory; -import com.facebook.thrift.server.TThreadPoolServer; -import com.facebook.thrift.server.TThreadPoolServer.Options; -import com.facebook.thrift.transport.TServerSocket; +import java.io.*; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import org.apache.cassandra.db.*; +import org.apache.cassandra.net.*; +import org.apache.cassandra.utils.*; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql.common.CqlResult; @@ -50,21 +48,18 @@ import org.apache.cassandra.db.IColumn; import org.apache.cassandra.db.Row; import org.apache.cassandra.db.RowMutation; import org.apache.cassandra.db.RowMutationMessage; +import org.apache.cassandra.gms.FailureDetector; +import org.apache.cassandra.io.DataInputBuffer; import org.apache.cassandra.net.EndPoint; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.net.IAsyncResult; import org.apache.cassandra.utils.LogUtil; -import org.apache.cassandra.io.DataInputBuffer; -import org.apache.cassandra.io.DataOutputBuffer; -import org.apache.cassandra.dht.OrderPreservingPartitioner; - +import org.apache.log4j.Logger; /** * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com ) */ -public class CassandraServer extends FacebookBase implements - Cassandra.Iface +public class CassandraServer extends FacebookBase implements Cassandra.Iface { private static Logger logger_ = Logger.getLogger(CassandraServer.class); @@ -81,7 +76,7 @@ public class CassandraServer extends FacebookBase implements storageService = StorageService.instance(); } - public CassandraServer() + public CassandraServer() throws Throwable { super("CassandraServer"); // Create the instance of the storage service @@ -93,7 +88,7 @@ public class CassandraServer extends FacebookBase implements * specified port. */ public void start() throws Throwable - { + { LogUtil.init(); //LogUtil.setLogLevel("com.facebook", "DEBUG"); // Start the storage service @@ -125,7 +120,7 @@ public class CassandraServer extends FacebookBase implements { throw new CassandraException("No row exists for key " + key); } - Map cfMap = row.getColumnFamilyMap(); + Map cfMap = row.getColumnFamilies(); if (cfMap == null || cfMap.size() == 0) { logger_ .info("ERROR ColumnFamily " + columnFamily + " map is missing.....: " + " key:" + key ); @@ -167,7 +162,7 @@ public class CassandraServer extends FacebookBase implements throw new CassandraException("ERROR No row for this key .....: " + key); } - Map cfMap = row.getColumnFamilyMap(); + Map cfMap = row.getColumnFamilies(); if (cfMap == null || cfMap.size() == 0) { logger_ .info("ERROR ColumnFamily " + columnFamily_column + " map is missing.....: " + " key:" + key); @@ -282,7 +277,7 @@ public class CassandraServer extends FacebookBase implements throw new CassandraException("ERROR No row for this key .....: " + key); } - Map cfMap = row.getColumnFamilyMap(); + Map cfMap = row.getColumnFamilies(); if (cfMap == null || cfMap.size() == 0) { logger_ .info("ERROR ColumnFamily " + columnFamily_column + " map is missing.....: " + " key:" + key); @@ -350,7 +345,7 @@ public class CassandraServer extends FacebookBase implements throw new CassandraException("ERROR No row for this key .....: " + key); } - Map cfMap = row.getColumnFamilyMap(); + Map cfMap = row.getColumnFamilies(); if (cfMap == null || cfMap.size() == 0) { logger_ .info("ERROR ColumnFamily map is missing.....: " @@ -422,7 +417,7 @@ public class CassandraServer extends FacebookBase implements throw new CassandraException("ERROR No row for this key .....: " + key); } - Map cfMap = row.getColumnFamilyMap(); + Map cfMap = row.getColumnFamilies(); if (cfMap == null || cfMap.size() == 0) { logger_ .info("ERROR ColumnFamily map is missing.....: " @@ -486,34 +481,132 @@ public class CassandraServer extends FacebookBase implements public boolean batch_insert_blocking(batch_mutation_t batchMutation) { - logger_.debug("batch_insert_blocking"); - RowMutation rm = RowMutation.getRowMutation(batchMutation); - return StorageProxy.insertBlocking(rm); - } + // 1. Get the N nodes from storage service where the data needs to be + // replicated + // 2. Construct a message for read\write + // 3. SendRR ( to all the nodes above ) + // 4. Wait for a response from atleast X nodes where X <= N + // 5. return success + boolean result = false; + try + { + logger_.warn(" batch_insert_blocking"); + validateTable(batchMutation.table); + IResponseResolver writeResponseResolver = new WriteResponseResolver(); + QuorumResponseHandler quorumResponseHandler = new QuorumResponseHandler( + DatabaseDescriptor.getReplicationFactor(), + writeResponseResolver); + EndPoint[] endpoints = storageService.getNStorageEndPoint(batchMutation.key); + // TODO: throw a thrift exception if we do not have N nodes + logger_.debug(" Creating the row mutation"); + RowMutation rm = new RowMutation(batchMutation.table, + batchMutation.key.trim()); + Set keys = batchMutation.cfmap.keySet(); + Iterator keyIter = keys.iterator(); + while (keyIter.hasNext()) + { + Object key = keyIter.next(); // Get the next key. + List list = batchMutation.cfmap.get(key); + for (column_t columnData : list) + { + rm.add(key.toString() + ":" + columnData.columnName, + columnData.value.getBytes(), columnData.timestamp); + + } + } + + RowMutationMessage rmMsg = new RowMutationMessage(rm); + Message message = new Message(StorageService.getLocalStorageEndPoint(), + StorageService.mutationStage_, + StorageService.mutationVerbHandler_, + new Object[]{ rmMsg } + ); + MessagingService.getMessagingInstance().sendRR(message, endpoints, + quorumResponseHandler); + logger_.debug(" Calling quorum response handler's get"); + result = quorumResponseHandler.get(); + + // TODO: if the result is false that means the writes to all the + // servers failed hence we need to throw an exception or return an + // error back to the client so that it can take appropriate action. + } + catch (Exception e) + { + logger_.info( LogUtil.throwableToString(e) ); + } + return result; + + } public void batch_insert(batch_mutation_t batchMutation) { - logger_.debug("batch_insert"); - RowMutation rm = RowMutation.getRowMutation(batchMutation); - StorageProxy.insert(rm); + // 1. Get the N nodes from storage service where the data needs to be + // replicated + // 2. Construct a message for read\write + // 3. SendRR ( to all the nodes above ) + // 4. Wait for a response from atleast X nodes where X <= N + // 5. return success + + try + { + logger_.debug(" batch_insert"); + logger_.debug(" Creating the row mutation"); + validateTable(batchMutation.table); + RowMutation rm = new RowMutation(batchMutation.table, + batchMutation.key.trim()); + if(batchMutation.cfmap != null) + { + Set keys = batchMutation.cfmap.keySet(); + Iterator keyIter = keys.iterator(); + while (keyIter.hasNext()) + { + Object key = keyIter.next(); // Get the next key. + List list = batchMutation.cfmap.get(key); + for (column_t columnData : list) + { + rm.add(key.toString() + ":" + columnData.columnName, + columnData.value.getBytes(), columnData.timestamp); + + } + } + } + if(batchMutation.cfmapdel != null) + { + Set keys = batchMutation.cfmapdel.keySet(); + Iterator keyIter = keys.iterator(); + while (keyIter.hasNext()) + { + Object key = keyIter.next(); // Get the next key. + List list = batchMutation.cfmapdel.get(key); + for (column_t columnData : list) + { + rm.delete(key.toString() + ":" + columnData.columnName); + } + } + } + StorageProxy.insert(rm); + } + catch (Exception e) + { + logger_.info( LogUtil.throwableToString(e) ); + } + return; } public void remove(String tablename, String key, String columnFamily_column) { - throw new UnsupportedOperationException("Remove is coming soon"); - } - - public boolean remove(String tablename, String key, String columnFamily_column, long timestamp, int block_for) - { - logger_.debug("remove"); - RowMutation rm = new RowMutation(tablename, key.trim()); - rm.delete(columnFamily_column, timestamp); - if (block_for > 0) { - return StorageProxy.insertBlocking(rm); - } else { + try + { + validateTable(tablename); + RowMutation rm = new RowMutation(tablename, key.trim()); + rm.delete(columnFamily_column); StorageProxy.insert(rm); - return true; - } + } + catch (Exception e) + { + logger_.debug( LogUtil.throwableToString(e) ); + } + return; } public List get_slice_super_by_names(String tablename, String key, String columnFamily, List superColumnNames) throws CassandraException, TException @@ -590,7 +683,7 @@ public class CassandraServer extends FacebookBase implements throw new CassandraException("ERROR No row for this key .....: " + key); } - Map cfMap = row.getColumnFamilyMap(); + Map cfMap = row.getColumnFamilies(); if (cfMap == null || cfMap.size() == 0) { logger_ .info("ERROR ColumnFamily map is missing.....: " @@ -665,7 +758,7 @@ public class CassandraServer extends FacebookBase implements throw new CassandraException("ERROR No row for this key .....: " + key); } - Map cfMap = row.getColumnFamilyMap(); + Map cfMap = row.getColumnFamilies(); if (cfMap == null || cfMap.size() == 0) { logger_ .info("ERROR ColumnFamily map is missing.....: " @@ -721,16 +814,110 @@ public class CassandraServer extends FacebookBase implements public boolean batch_insert_superColumn_blocking(batch_mutation_super_t batchMutationSuper) { - logger_.debug("batch_insert_SuperColumn_blocking"); - RowMutation rm = RowMutation.getRowMutation(batchMutationSuper); - return StorageProxy.insertBlocking(rm); + boolean result = false; + try + { + logger_.warn(" batch_insert_SuperColumn_blocking"); + logger_.debug(" Creating the row mutation"); + validateTable(batchMutationSuper.table); + RowMutation rm = new RowMutation(batchMutationSuper.table, + batchMutationSuper.key.trim()); + Set keys = batchMutationSuper.cfmap.keySet(); + Iterator keyIter = keys.iterator(); + while (keyIter.hasNext()) + { + Object key = keyIter.next(); // Get the next key. + List list = batchMutationSuper.cfmap.get(key); + for (superColumn_t superColumnData : list) + { + if(superColumnData.columns.size() != 0 ) + { + for (column_t columnData : superColumnData.columns) + { + rm.add(key.toString() + ":" + superColumnData.name +":" + columnData.columnName, + columnData.value.getBytes(), columnData.timestamp); + } + } + else + { + rm.add(key.toString() + ":" + superColumnData.name, new byte[0], 0); + } + } + } + StorageProxy.insert(rm); + } + catch (Exception e) + { + logger_.info( LogUtil.throwableToString(e) ); + } + return result; + } - public void batch_insert_superColumn(batch_mutation_super_t batchMutationSuper) { - logger_.debug("batch_insert_SuperColumn"); - RowMutation rm = RowMutation.getRowMutation(batchMutationSuper); - StorageProxy.insert(rm); + try + { + logger_.debug(" batch_insert"); + logger_.debug(" Creating the row mutation"); + validateTable(batchMutationSuper.table); + RowMutation rm = new RowMutation(batchMutationSuper.table, + batchMutationSuper.key.trim()); + if(batchMutationSuper.cfmap != null) + { + Set keys = batchMutationSuper.cfmap.keySet(); + Iterator keyIter = keys.iterator(); + while (keyIter.hasNext()) + { + Object key = keyIter.next(); // Get the next key. + List list = batchMutationSuper.cfmap.get(key); + for (superColumn_t superColumnData : list) + { + if(superColumnData.columns.size() != 0 ) + { + for (column_t columnData : superColumnData.columns) + { + rm.add(key.toString() + ":" + superColumnData.name +":" + columnData.columnName, + columnData.value.getBytes(), columnData.timestamp); + } + } + else + { + rm.add(key.toString() + ":" + superColumnData.name, new byte[0], 0); + } + } + } + } + if(batchMutationSuper.cfmapdel != null) + { + Set keys = batchMutationSuper.cfmapdel.keySet(); + Iterator keyIter = keys.iterator(); + while (keyIter.hasNext()) + { + Object key = keyIter.next(); // Get the next key. + List list = batchMutationSuper.cfmapdel.get(key); + for (superColumn_t superColumnData : list) + { + if(superColumnData.columns.size() != 0 ) + { + for (column_t columnData : superColumnData.columns) + { + rm.delete(key.toString() + ":" + superColumnData.name +":" + columnData.columnName); + } + } + else + { + rm.delete(key.toString() + ":" + superColumnData.name); + } + } + } + } + StorageProxy.insert(rm); + } + catch (Exception e) + { + logger_.info( LogUtil.throwableToString(e) ); + } + return; } public String getStringProperty(String propertyName) throws TException @@ -817,76 +1004,7 @@ public class CassandraServer extends FacebookBase implements } return result; } - - public List get_range(String tablename, final String startkey) throws CassandraException - { - if (!(StorageService.getPartitioner() instanceof OrderPreservingPartitioner)) { - throw new CassandraException("range queries may only be performed against an order-preserving partitioner"); - } - - logger_.debug("get_range"); - - // send request - Message message; - DataOutputBuffer dob = new DataOutputBuffer(); - try - { - dob.writeUTF(startkey); - } - catch (IOException e) - { - logger_.error("unable to write startkey", e); - throw new RuntimeException(e); - } - byte[] messageBody = Arrays.copyOf(dob.getData(), dob.getLength()); - message = new Message(StorageService.getLocalStorageEndPoint(), - StorageService.readStage_, - StorageService.rangeVerbHandler_, - messageBody); - EndPoint endPoint; - try - { - endPoint = StorageService.instance().findSuitableEndPoint(startkey); - } - catch (Exception e) - { - throw new CassandraException("Unable to find endpoint for " + startkey); - } - IAsyncResult iar = MessagingService.getMessagingInstance().sendRR(message, endPoint); - - // read response - // TODO send more requests if we need to span multiple nodes (or can we just let client worry about that, - // since they have to handle multiple requests anyway?) - byte[] responseBody; - try - { - responseBody = (byte[]) iar.get(2 * DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS)[0]; - } - catch (TimeoutException e) - { - throw new RuntimeException(e); - } - DataInputBuffer bufIn = new DataInputBuffer(); - bufIn.reset(responseBody, responseBody.length); - - // turn into List - List keys = new ArrayList(); - while (bufIn.getPosition() < responseBody.length) - { - try - { - keys.add(bufIn.readUTF()); - } - catch (IOException e) - { - logger_.error("bad utf", e); - throw new RuntimeException(e); - } - } - - return keys; - } - + /* * This method is used to ensure that all keys * prior to the specified key, as dtermined by @@ -924,15 +1042,6 @@ public class CassandraServer extends FacebookBase implements public static void main(String[] args) throws Throwable { int port = 9160; - - Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() - { - public void uncaughtException(Thread t, Throwable e) - { - logger_.error("Fatal exception in thread " + t, e); - } - }); - try { CassandraServer peerStorageServer = new CassandraServer(); diff --git a/src/org/apache/cassandra/service/ConsistencyManager.java b/src/org/apache/cassandra/service/ConsistencyManager.java index d0d9ca60e8..f8a87df28b 100644 --- a/src/org/apache/cassandra/service/ConsistencyManager.java +++ b/src/org/apache/cassandra/service/ConsistencyManager.java @@ -58,6 +58,11 @@ class ConsistencyManager implements Runnable if ( responses_.size() == ConsistencyManager.this.replicas_.size() ) handleDigestResponses(); } + + public void attachContext(Object o) + { + throw new UnsupportedOperationException("This operation is not currently supported."); + } private void handleDigestResponses() { @@ -91,7 +96,8 @@ class ConsistencyManager implements Runnable replicas_.add(StorageService.getLocalStorageEndPoint()); IAsyncCallback responseHandler = new DataRepairHandler(ConsistencyManager.this.replicas_.size(), readResponseResolver); String table = DatabaseDescriptor.getTables().get(0); - ReadMessage readMessage = new ReadMessage(table, row_.key(), columnFamily_); + ReadMessage readMessage = constructReadMessage(false); + // ReadMessage readMessage = new ReadMessage(table, row_.key(), columnFamily_); Message message = ReadMessage.makeReadMessage(readMessage); MessagingService.getMessagingInstance().sendRR(message, replicas_.toArray( new EndPoint[0] ), responseHandler); } @@ -116,10 +122,14 @@ class ConsistencyManager implements Runnable if ( responses_.size() == majority_ ) { String messageId = message.getMessageId(); - readRepairTable_.put(messageId, messageId, this); - // handleResponses(); + readRepairTable_.put(messageId, messageId, this); } } + + public void attachContext(Object o) + { + throw new UnsupportedOperationException("This operation is not currently supported."); + } public void callMe(String key, String value) { @@ -176,30 +186,8 @@ class ConsistencyManager implements Runnable public void run() { - logger_.debug(" Run the consistency checks for " + columnFamily_); - String table = DatabaseDescriptor.getTables().get(0); - ReadMessage readMessageDigestOnly = null; - if(columnNames_.size() == 0) - { - if( start_ >= 0 && count_ < Integer.MAX_VALUE) - { - readMessageDigestOnly = new ReadMessage(table, row_.key(), columnFamily_, start_, count_); - } - else if(sinceTimestamp_ > 0) - { - readMessageDigestOnly = new ReadMessage(table, row_.key(), columnFamily_, sinceTimestamp_); - } - else - { - readMessageDigestOnly = new ReadMessage(table, row_.key(), columnFamily_); - } - } - else - { - readMessageDigestOnly = new ReadMessage(table, row_.key(), columnFamily_, columnNames_); - - } - readMessageDigestOnly.setIsDigestQuery(true); + logger_.debug(" Run the consistency checks for " + columnFamily_); + ReadMessage readMessageDigestOnly = constructReadMessage(true); try { Message messageDigestOnly = ReadMessage.makeReadMessage(readMessageDigestOnly); @@ -211,4 +199,33 @@ class ConsistencyManager implements Runnable logger_.info(LogUtil.throwableToString(ex)); } } + + private ReadMessage constructReadMessage(boolean isDigestQuery) + { + ReadMessage readMessage = null; + String table = DatabaseDescriptor.getTables().get(0); + + if(columnNames_.size() == 0) + { + if( start_ >= 0 && count_ < Integer.MAX_VALUE) + { + readMessage = new ReadMessage(table, row_.key(), columnFamily_, start_, count_); + } + else if(sinceTimestamp_ > 0) + { + readMessage = new ReadMessage(table, row_.key(), columnFamily_, sinceTimestamp_); + } + else + { + readMessage = new ReadMessage(table, row_.key(), columnFamily_); + } + } + else + { + readMessage = new ReadMessage(table, row_.key(), columnFamily_, columnNames_); + + } + readMessage.setIsDigestQuery(isDigestQuery); + return readMessage; + } } diff --git a/src/org/apache/cassandra/service/HttpRequestVerbHandler.java b/src/org/apache/cassandra/service/HttpRequestVerbHandler.java index 4257ad5577..0c847fff7e 100644 --- a/src/org/apache/cassandra/service/HttpRequestVerbHandler.java +++ b/src/org/apache/cassandra/service/HttpRequestVerbHandler.java @@ -23,18 +23,15 @@ import java.lang.management.ManagementFactory; import java.lang.management.MemoryMXBean; import java.lang.management.MemoryUsage; import java.lang.management.RuntimeMXBean; +import java.math.BigInteger; import java.nio.ByteBuffer; -import java.util.Arrays; -import java.util.List; -import java.util.Map; -import java.util.Set; - -import org.apache.log4j.Logger; +import java.util.*; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.CalloutDeployMessage; import org.apache.cassandra.db.CalloutManager; import org.apache.cassandra.db.ColumnFamily; +import org.apache.cassandra.db.IColumn; import org.apache.cassandra.db.RowMutation; import org.apache.cassandra.db.Table; import org.apache.cassandra.dht.Range; @@ -51,6 +48,8 @@ import org.apache.cassandra.net.http.HttpRequest; import org.apache.cassandra.net.http.HttpWriteResponse; import org.apache.cassandra.procedures.GroovyScriptRunner; import org.apache.cassandra.utils.LogUtil; +import org.apache.log4j.Logger; +import org.apache.cassandra.net.*; /* * This class handles the incoming HTTP request after diff --git a/src/org/apache/cassandra/service/QuorumResponseHandler.java b/src/org/apache/cassandra/service/QuorumResponseHandler.java index c4c6177392..e47c57ff36 100644 --- a/src/org/apache/cassandra/service/QuorumResponseHandler.java +++ b/src/org/apache/cassandra/service/QuorumResponseHandler.java @@ -26,12 +26,13 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.WriteResponseMessage; import org.apache.cassandra.net.IAsyncCallback; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.utils.LogUtil; import org.apache.log4j.Logger; - +import org.apache.cassandra.utils.*; /** * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com ) */ @@ -122,4 +123,9 @@ public class QuorumResponseHandler implements IAsyncCallback lock_.unlock(); } } + + public void attachContext(Object o) + { + throw new UnsupportedOperationException("This operation is not supported in this version of the callback handler"); + } } diff --git a/src/org/apache/cassandra/service/RangeVerbHandler.java b/src/org/apache/cassandra/service/RangeVerbHandler.java deleted file mode 100644 index 3587907fc2..0000000000 --- a/src/org/apache/cassandra/service/RangeVerbHandler.java +++ /dev/null @@ -1,152 +0,0 @@ -package org.apache.cassandra.service; - -import java.util.ArrayList; -import java.util.List; -import java.util.Iterator; -import java.util.Comparator; -import java.util.Arrays; -import java.io.FileNotFoundException; -import java.io.IOException; - -import org.apache.commons.collections.IteratorUtils; -import org.apache.commons.collections.Predicate; - -import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.net.IVerbHandler; -import org.apache.cassandra.net.Message; -import org.apache.cassandra.db.IdentityFilter; -import org.apache.cassandra.db.ColumnFamily; -import org.apache.cassandra.db.ColumnFamilyStore; -import org.apache.cassandra.db.FileStruct; -import org.apache.cassandra.db.Table; -import org.apache.cassandra.db.Memtable; -import org.apache.cassandra.db.MemtableManager; -import org.apache.cassandra.io.SequenceFile; -import org.apache.cassandra.io.DataInputBuffer; -import org.apache.cassandra.io.DataOutputBuffer; -import org.apache.cassandra.config.DatabaseDescriptor; - -public class RangeVerbHandler implements IVerbHandler -{ - public static final Comparator STRING_COMPARATOR = new Comparator() - { - public int compare(String o1, String o2) - { - return o1.compareTo(o2); - } - }; - - public void doVerb(Message message) - { - byte[] bytes = (byte[]) message.getMessageBody()[0]; - final String startkey; - if (bytes.length == 0) - { - startkey = ""; - } - else - { - DataInputBuffer dib = new DataInputBuffer(); - dib.reset(bytes, bytes.length); - try - { - startkey = dib.readUTF(); - } - catch (IOException e) - { - throw new RuntimeException(e); - } - } - - List> iterators = new ArrayList>(); - Table table = Table.open(DatabaseDescriptor.getTableName()); - for (String cfName : table.getApplicationColumnFamilies()) - { - ColumnFamilyStore cfs = table.getColumnFamilyStore(cfName); - - // memtable keys: current and historical - Iterator it = (Iterator) IteratorUtils.chainedIterator( - IteratorUtils.singletonIterator(cfs.getMemtable()), - MemtableManager.instance().getUnflushedMemtables(cfName).iterator()); - while (it.hasNext()) - { - iterators.add(IteratorUtils.filteredIterator(it.next().sortedKeyIterator(), new Predicate() - { - public boolean evaluate(Object key) - { - return ((String) key).compareTo(startkey) >= 0; - } - })); - } - - // sstables - for (String filename : cfs.getSSTableFilenames()) - { - try - { - FileStruct fs = new FileStruct(SequenceFile.reader(filename)); - fs.seekTo(startkey); - iterators.add(fs.iterator()); - } - catch (FileNotFoundException e) - { - throw new RuntimeException(e); - } - catch (IOException e) - { - throw new RuntimeException(e); - } - } - } - - Iterator iter = IteratorUtils.collatedIterator(STRING_COMPARATOR, iterators); - List keys = new ArrayList(); - String last = null, current = null; - - while (keys.size() < 1000) - { - if (!iter.hasNext()) - { - break; - } - current = iter.next(); - if (!current.equals(last)) - { - last = current; - for (String cfName : table.getApplicationColumnFamilies()) - { - ColumnFamilyStore cfs = table.getColumnFamilyStore(cfName); - try - { - ColumnFamily cf = cfs.getColumnFamily(current, cfName, new IdentityFilter()); - if (cf != null && cf.getColumns().size() > 0) - { - keys.add(current); - break; - } - } - catch (IOException e) - { - throw new RuntimeException(); - } - } - } - } - - DataOutputBuffer dob = new DataOutputBuffer(); - for (String key : keys) - { - try - { - dob.writeUTF(key); - } - catch (IOException e) - { - throw new RuntimeException(e); - } - } - byte[] data = Arrays.copyOf(dob.getData(), dob.getLength()); - Message response = message.getReply(StorageService.getLocalStorageEndPoint(), data); - MessagingService.getMessagingInstance().sendOneWay(response, message.getFrom()); - } -} diff --git a/src/org/apache/cassandra/service/ReadResponseResolver.java b/src/org/apache/cassandra/service/ReadResponseResolver.java index 8969956b33..21ffd250be 100644 --- a/src/org/apache/cassandra/service/ReadResponseResolver.java +++ b/src/org/apache/cassandra/service/ReadResponseResolver.java @@ -37,8 +37,8 @@ import org.apache.cassandra.utils.LogUtil; import org.apache.log4j.Logger; -/* - * This class is used by all read functions and is called by the Qorum +/** + * This class is used by all read functions and is called by the Quorum * when atleast a few of the servers ( few is specified in Quorum) * have sent the response . The resolve fn then schedules read repair * and resolution of read data from the various servers. @@ -46,7 +46,6 @@ import org.apache.log4j.Logger; */ public class ReadResponseResolver implements IResponseResolver { - private static Logger logger_ = Logger.getLogger(WriteResponseResolver.class); /* @@ -101,15 +100,16 @@ public class ReadResponseResolver implements IResponseResolver logger_.info(LogUtil.throwableToString(ex)); } } - // If there was a digest query compare it withh all teh data digests - // If there is a mismatch then thwrow an exception so that read repair can happen. + // If there was a digest query compare it with all the data digests + // If there is a mismatch then throw an exception so that read repair can happen. if(isDigestQuery) { for(Row row: rowList) { if( !Arrays.equals(row.digest(), digest) ) { - throw new DigestMismatchException("The Digest does not match"); + /* Wrap the key as the context in this exception */ + throw new DigestMismatchException(row.key()); } } } @@ -141,13 +141,13 @@ public class ReadResponseResolver implements IResponseResolver continue; // create the row mutation message based on the diff and schedule a read repair RowMutation rowMutation = new RowMutation(table, key); - Map columnFamilies = diffRow.getColumnFamilyMap(); + Map columnFamilies = diffRow.getColumnFamilies(); Set cfNames = columnFamilies.keySet(); for ( String cfName : cfNames ) { ColumnFamily cf = columnFamilies.get(cfName); - rowMutation.add(cf); + rowMutation.add(cfName, cf); } RowMutationMessage rowMutationMessage = new RowMutationMessage(rowMutation); // schedule the read repair diff --git a/src/org/apache/cassandra/service/StorageLoadBalancer.java b/src/org/apache/cassandra/service/StorageLoadBalancer.java index 1c82777027..3e6f4ab223 100644 --- a/src/org/apache/cassandra/service/StorageLoadBalancer.java +++ b/src/org/apache/cassandra/service/StorageLoadBalancer.java @@ -19,30 +19,33 @@ package org.apache.cassandra.service; import java.io.Serializable; -import java.util.HashMap; -import java.util.Map; +import java.math.BigInteger; +import java.util.*; 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.Token; +import org.apache.cassandra.dht.LeaveJoinProtocolImpl; +import org.apache.cassandra.dht.Range; 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 @@ -161,6 +164,7 @@ 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); } @@ -392,18 +396,18 @@ final class StorageLoadBalancer implements IEndPointStateChangeSubscriber, IComp class MoveMessage implements Serializable { - private Token targetToken_; + private BigInteger targetToken_; private MoveMessage() { } - MoveMessage(Token targetToken) + MoveMessage(BigInteger targetToken) { targetToken_ = targetToken; } - Token getTargetToken() + BigInteger getTargetToken() { return targetToken_; } diff --git a/src/org/apache/cassandra/service/StorageProxy.java b/src/org/apache/cassandra/service/StorageProxy.java index 4225209495..1f1aba5ac1 100644 --- a/src/org/apache/cassandra/service/StorageProxy.java +++ b/src/org/apache/cassandra/service/StorageProxy.java @@ -26,9 +26,6 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; - -import org.apache.commons.lang.StringUtils; - import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ReadMessage; import org.apache.cassandra.db.ReadResponseMessage; @@ -39,6 +36,7 @@ import org.apache.cassandra.db.Table; import org.apache.cassandra.db.TouchMessage; import org.apache.cassandra.io.DataInputBuffer; import org.apache.cassandra.net.EndPoint; +import org.apache.cassandra.net.IAsyncCallback; import org.apache.cassandra.net.IAsyncResult; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; @@ -48,35 +46,35 @@ import org.apache.log4j.Logger; public class StorageProxy { - private static Logger logger_ = Logger.getLogger(StorageProxy.class); + private static Logger logger_ = Logger.getLogger(StorageProxy.class); /** * This method is responsible for creating Message to be * sent over the wire to N replicas where some of the replicas * may be hints. */ - private static Map createWriteMessages(RowMutation rm, Map endpointMap) throws IOException + private static Map createWriteMessages(RowMutationMessage rmMessage, Map endpointMap) throws IOException { - Map messageMap = new HashMap(); - Message message = rm.makeRowMutationMessage(); - - for (Map.Entry entry : endpointMap.entrySet()) - { - EndPoint target = entry.getKey(); - EndPoint hint = entry.getValue(); + Map messageMap = new HashMap(); + Message message = RowMutationMessage.makeRowMutationMessage(rmMessage); + + Set targets = endpointMap.keySet(); + for( EndPoint target : targets ) + { + EndPoint hint = endpointMap.get(target); if ( !target.equals(hint) ) - { - Message hintedMessage = rm.makeRowMutationMessage(); - hintedMessage.addHeader(RowMutation.HINT, EndPoint.toBytes(hint) ); - logger_.debug("Sending the hint of " + target.getHost() + " to " + hint.getHost()); - messageMap.put(target, hintedMessage); - } - else - { - messageMap.put(target, message); - } - } - return messageMap; + { + Message hintedMessage = RowMutationMessage.makeRowMutationMessage(rmMessage); + hintedMessage.addHeader(RowMutationMessage.hint_, EndPoint.toBytes(hint) ); + logger_.debug("Sending the hint of " + target.getHost() + " to " + hint.getHost()); + messageMap.put(target, hintedMessage); + } + else + { + messageMap.put(target, message); + } + } + return messageMap; } /** @@ -84,96 +82,124 @@ public class StorageProxy * across all replicas. This method will take care * of the possibility of a replica being down and hint * the data across to some other replica. - * @param rm the mutation to be applied across the replicas + * @param RowMutation the mutation to be applied + * across the replicas */ public static void insert(RowMutation rm) - { + { /* * Get the N nodes from storage service where the data needs to be * replicated * Construct a message for write * Send them asynchronously to the replicas. */ - assert rm.key() != null; - - try - { - Map endpointMap = StorageService.instance().getNStorageEndPointMap(rm.key()); - // TODO: throw a thrift exception if we do not have N nodes - Map messageMap = createWriteMessages(rm, endpointMap); - logger_.debug("insert writing to [" + StringUtils.join(messageMap.keySet(), ", ") + "]"); - for (Map.Entry entry : messageMap.entrySet()) - { - MessagingService.getMessagingInstance().sendOneWay(entry.getValue(), entry.getKey()); - } - } + try + { + logger_.debug(" insert"); + Map endpointMap = StorageService.instance().getNStorageEndPointMap(rm.key()); + // TODO: throw a thrift exception if we do not have N nodes + RowMutationMessage rmMsg = new RowMutationMessage(rm); + /* Create the write messages to be sent */ + Map messageMap = createWriteMessages(rmMsg, endpointMap); + Set endpoints = messageMap.keySet(); + for(EndPoint endpoint : endpoints) + { + MessagingService.getMessagingInstance().sendOneWay(messageMap.get(endpoint), endpoint); + } + } catch (Exception e) { - logger_.error( LogUtil.throwableToString(e) ); + logger_.info( LogUtil.throwableToString(e) ); } return; } - public static boolean insertBlocking(RowMutation rm) + + private static Map constructMessages(Map readMessages) throws IOException { - assert rm.key() != null; - - try + Map messages = new HashMap(); + Set keys = readMessages.keySet(); + for ( String key : keys ) { - Message message = rm.makeRowMutationMessage(); - - IResponseResolver writeResponseResolver = new WriteResponseResolver(); - QuorumResponseHandler quorumResponseHandler = new QuorumResponseHandler( - DatabaseDescriptor.getReplicationFactor(), - writeResponseResolver); - EndPoint[] endpoints = StorageService.instance().getNStorageEndPoint(rm.key()); - logger_.debug("insertBlocking writing to [" + StringUtils.join(endpoints, ", ") + "]"); - // TODO: throw a thrift exception if we do not have N nodes - - MessagingService.getMessagingInstance().sendRR(message, endpoints, quorumResponseHandler); - return quorumResponseHandler.get(); - - // TODO: if the result is false that means the writes to all the - // servers failed hence we need to throw an exception or return an - // error back to the client so that it can take appropriate action. - } - catch (Exception e) + Message message = ReadMessage.makeReadMessage( readMessages.get(key) ); + messages.put(key, message); + } + return messages; + } + + private static IAsyncResult dispatchMessages(Map endPoints, Map messages) + { + Set keys = endPoints.keySet(); + EndPoint[] eps = new EndPoint[keys.size()]; + Message[] msgs = new Message[keys.size()]; + + int i = 0; + for ( String key : keys ) { - logger_.error( LogUtil.throwableToString(e) ); - return false; + eps[i] = endPoints.get(key); + msgs[i] = messages.get(key); + ++i; } + + IAsyncResult iar = MessagingService.getMessagingInstance().sendRR(msgs, eps); + return iar; + } + + /** + * This is an implementation for the multiget version. + * @param readMessages map of key --> ReadMessage to be sent + * @return map of key --> Row + * @throws IOException + * @throws TimeoutException + */ + public static Map doReadProtocol(Map readMessages) throws IOException,TimeoutException + { + Map rows = new HashMap(); + Set keys = readMessages.keySet(); + /* Find all the suitable endpoints for the keys */ + Map endPoints = StorageService.instance().findSuitableEndPoints(keys.toArray( new String[0] )); + /* Construct the messages to be sent out */ + Map messages = constructMessages(readMessages); + /* Dispatch the messages to the respective endpoints */ + IAsyncResult iar = dispatchMessages(endPoints, messages); + List results = iar.multiget(2*DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS); + + for ( Object[] result : results ) + { + byte[] body = (byte[])result[0]; + DataInputBuffer bufIn = new DataInputBuffer(); + bufIn.reset(body, body.length); + ReadResponseMessage responseMessage = ReadResponseMessage.serializer().deserialize(bufIn); + Row row = responseMessage.row(); + rows.put(row.key(), row); + } + return rows; } public static Row doReadProtocol(String key, ReadMessage readMessage) throws IOException,TimeoutException { - EndPoint endPoint = null; - try - { - endPoint = StorageService.instance().findSuitableEndPoint(key); - } - catch( Throwable ex) - { - ex.printStackTrace(); - } + Row row = null; + EndPoint endPoint = StorageService.instance().findSuitableEndPoint(key); if(endPoint != null) { Message message = ReadMessage.makeReadMessage(readMessage); + message.addHeader(ReadMessage.doRepair_, ReadMessage.doRepair_.getBytes()); IAsyncResult iar = MessagingService.getMessagingInstance().sendRR(message, endPoint); Object[] result = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS); byte[] body = (byte[])result[0]; DataInputBuffer bufIn = new DataInputBuffer(); bufIn.reset(body, body.length); ReadResponseMessage responseMessage = ReadResponseMessage.serializer().deserialize(bufIn); - return responseMessage.row(); + row = responseMessage.row(); } else { logger_.warn(" Alert : Unable to find a suitable end point for the key : " + key ); } - return null; + return row; } - static void touch_local (String tablename, String key, boolean fData ) throws IOException + static void touch_local(String tablename, String key, boolean fData ) throws IOException { Table table = Table.open( tablename ); table.touch(key, fData); @@ -237,10 +263,8 @@ public class StorageProxy weakTouchProtocol(tablename, key, fData); break; } - } + } - - public static Row readProtocol(String tablename, String key, String columnFamily, List columnNames, StorageService.ConsistencyLevel consistencyLevel) throws Exception { Row row = null; @@ -277,9 +301,7 @@ public class StorageProxy break; } } - return row; - - + return row; } public static Row readProtocol(String tablename, String key, String columnFamily, int start, int count, StorageService.ConsistencyLevel consistencyLevel) throws Exception @@ -321,6 +343,26 @@ public class StorageProxy return row; } + public static Map readProtocol(String tablename, String[] keys, String columnFamily, int start, int count, StorageService.ConsistencyLevel consistencyLevel) throws Exception + { + Map rows = new HashMap(); + switch ( consistencyLevel ) + { + case WEAK: + rows = weakReadProtocol(tablename, keys, columnFamily, start, count); + break; + + case STRONG: + rows = strongReadProtocol(tablename, keys, columnFamily, start, count); + break; + + default: + rows = weakReadProtocol(tablename, keys, columnFamily, start, count); + break; + } + return rows; + } + public static Row readProtocol(String tablename, String key, String columnFamily, long sinceTimestamp, StorageService.ConsistencyLevel consistencyLevel) throws Exception { Row row = null; @@ -374,18 +416,21 @@ public class StorageProxy return row; } - /* - * This function executes the read protocol. - // 1. Get the N nodes from storage service where the data needs to be - // replicated - // 2. Construct a message for read\write - * 3. Set one of teh messages to get teh data and teh rest to get teh digest - // 4. SendRR ( to all the nodes above ) - // 5. Wait for a response from atleast X nodes where X <= N and teh data node - * 6. If the digest matches return teh data. - * 7. else carry out read repair by getting data from all the nodes. - // 5. return success - * + /** + * This function executes the read protocol. + * 1. Get the N nodes from storage service where the data needs to be replicated + * 2. Construct a message for read\write + * 3. Set one of teh messages to get teh data and teh rest to get teh digest + * 4. SendRR ( to all the nodes above ) + * 5. Wait for a response from atleast X nodes where X <= N and teh data node + * 6. If the digest matches return teh data. + * 7. else carry out read repair by getting data from all the nodes. + * @param tablename the name of the table + * @param key the row key identifier + * @param columnFamily the column in Cassandra format + * @start the start position + * @count the number of columns we are interested in + * @throws IOException, TimeoutException */ public static Row strongReadProtocol(String tablename, String key, String columnFamily, int start, int count) throws IOException, TimeoutException { @@ -416,6 +461,49 @@ public class StorageProxy return row; } + /** + * This is a multiget version of the above method. + * @param tablename + * @param keys + * @param columnFamily + * @param start + * @param count + * @return + * @throws IOException + * @throws TimeoutException + */ + public static Map strongReadProtocol(String tablename, String[] keys, String columnFamily, int start, int count) throws IOException, TimeoutException + { + Map rows = new HashMap(); + long startTime = System.currentTimeMillis(); + // TODO: throw a thrift exception if we do not have N nodes + Map readMessages = new HashMap(); + for (String key : keys ) + { + ReadMessage[] readMessage = new ReadMessage[2]; + if( start >= 0 && count < Integer.MAX_VALUE) + { + readMessage[0] = new ReadMessage(tablename, key, columnFamily, start, count); + } + else + { + readMessage[0] = new ReadMessage(tablename, key, columnFamily); + } + if( start >= 0 && count < Integer.MAX_VALUE) + { + readMessage[1] = new ReadMessage(tablename, key, columnFamily, start, count); + } + else + { + readMessage[1] = new ReadMessage(tablename, key, columnFamily); + } + readMessage[1].setIsDigestQuery(true); + } + rows = doStrongReadProtocol(readMessages); + logger_.debug("readProtocol: " + (System.currentTimeMillis() - startTime) + " ms."); + return rows; + } + public static Row strongReadProtocol(String tablename, String key, String columnFamily, long sinceTimestamp) throws IOException, TimeoutException { long startTime = System.currentTimeMillis(); @@ -431,8 +519,8 @@ public class StorageProxy return row; } - /* - * This method performs the actual read from the replicas. + /** + * This method performs the read from the replicas. * param @ key - key for which the data is required. * param @ readMessage - the read message to get the actual data * param @ readMessageDigest - the read message to get the digest. @@ -454,11 +542,14 @@ public class StorageProxy EndPoint[] endPoints = new EndPoint[endpointList.size() + 1]; Message messages[] = new Message[endpointList.size() + 1]; - // first message is the data Point + /* + * First message is sent to the node that will actually get + * the data for us. The other two replicas are only sent a + * digest query. + */ endPoints[0] = dataPoint; - messages[0] = message; - - for(int i=1; i < endPoints.length ; i++) + messages[0] = message; + for (int i=1; i < endPoints.length ; i++) { endPoints[i] = endpointList.get(i-1); messages[i] = messageDigestOnly; @@ -466,8 +557,7 @@ public class StorageProxy try { - MessagingService.getMessagingInstance().sendRR(messages, endPoints, quorumResponseHandler); - + MessagingService.getMessagingInstance().sendRR(messages, endPoints, quorumResponseHandler); long startTime2 = System.currentTimeMillis(); row = quorumResponseHandler.get(); logger_.debug("quorumResponseHandler: " + (System.currentTimeMillis() - startTime2) @@ -490,8 +580,7 @@ public class StorageProxy readMessage.setIsDigestQuery(false); logger_.info("DigestMismatchException: " + key); Message messageRepair = ReadMessage.makeReadMessage(readMessage); - MessagingService.getMessagingInstance().sendRR(messageRepair, endPoints, - quorumResponseHandlerRepair); + MessagingService.getMessagingInstance().sendRR(messageRepair, endPoints, quorumResponseHandlerRepair); try { row = quorumResponseHandlerRepair.get(); @@ -509,6 +598,111 @@ public class StorageProxy return row; } + private static Map constructReplicaMessages(Map readMessages) throws IOException + { + Map messages = new HashMap(); + Set keys = readMessages.keySet(); + + for ( String key : keys ) + { + Message[] msg = new Message[DatabaseDescriptor.getReplicationFactor()]; + ReadMessage[] readMessage = readMessages.get(key); + msg[0] = ReadMessage.makeReadMessage( readMessage[0] ); + for ( int i = 1; i < msg.length; ++i ) + { + msg[i] = ReadMessage.makeReadMessage( readMessage[1] ); + } + } + return messages; + } + + private static MultiQuorumResponseHandler dispatchMessages(Map readMessages, Map messages) throws IOException + { + Set keys = messages.keySet(); + /* This maps the keys to the original data read messages */ + Map readMessage = new HashMap(); + /* This maps the keys to their respective endpoints/replicas */ + Map endpoints = new HashMap(); + /* Groups the messages that need to be sent to the individual keys */ + Message[][] msgList = new Message[messages.size()][DatabaseDescriptor.getReplicationFactor()]; + /* Respects the above grouping and provides the endpoints for the above messages */ + EndPoint[][] epList = new EndPoint[messages.size()][DatabaseDescriptor.getReplicationFactor()]; + + int i = 0; + for ( String key : keys ) + { + /* This is the primary */ + EndPoint dataPoint = StorageService.instance().findSuitableEndPoint(key); + List replicas = new ArrayList( StorageService.instance().getNLiveStorageEndPoint(key) ); + replicas.remove(dataPoint); + /* Get the messages to be sent index 0 is the data messages and index 1 is the digest message */ + Message[] message = messages.get(key); + msgList[i][0] = message[0]; + int N = DatabaseDescriptor.getReplicationFactor(); + for ( int j = 1; j < N; ++j ) + { + msgList[i][j] = message[1]; + } + /* Get the endpoints to which the above messages need to be sent */ + epList[i][0] = dataPoint; + for ( int j = 1; i < N; ++i ) + { + epList[i][j] = replicas.get(j - 1); + } + /* Data ReadMessage associated with this key */ + readMessage.put( key, readMessages.get(key)[0] ); + /* EndPoints for this specific key */ + endpoints.put(key, epList[i]); + ++i; + } + + /* Handles the read semantics for this entire set of keys */ + MultiQuorumResponseHandler quorumResponseHandlers = new MultiQuorumResponseHandler(readMessage, endpoints); + MessagingService.getMessagingInstance().sendRR(msgList, epList, quorumResponseHandlers); + return quorumResponseHandlers; + } + + /** + * This method performs the read from the replicas for a bunch of keys. + * @param readMessages map of key --> readMessage[] of two entries where + * the first entry is the readMessage for the data and the second + * is the entry for the digest + * @return map containing key ---> Row + * @throws IOException, TimeoutException + */ + private static Map doStrongReadProtocol(Map readMessages) throws IOException + { + Map rows = new HashMap(); + /* Construct the messages to be sent to the replicas */ + Map replicaMessages = constructReplicaMessages(readMessages); + /* Dispatch the messages to the different replicas */ + MultiQuorumResponseHandler cb = dispatchMessages(readMessages, replicaMessages); + try + { + Row[] rows2 = cb.get(); + for ( Row row : rows2 ) + { + rows.put(row.key(), row); + } + } + catch ( TimeoutException ex ) + { + logger_.info("Operation timed out waiting for responses ..."); + logger_.info(LogUtil.throwableToString(ex)); + } + return rows; + } + + /** + * This version is used to retrieve the row associated with + * the specified key + * @param tablename name of the table that needs to be queried + * @param keys keys whose values we are interested in + * @param columnFamily name of the "column" we are interested in + * @param columns the columns we are interested in + * @return the interested row + * @throws Exception + */ public static Row weakReadProtocol(String tablename, String key, String columnFamily, List columns) throws Exception { long startTime = System.currentTimeMillis(); @@ -530,11 +724,56 @@ public class StorageProxy return row; } - /* + /** + * This version is used when results for multiple keys needs to be + * retrieved. + * + * @param tablename name of the table that needs to be queried + * @param keys keys whose values we are interested in + * @param columnFamily name of the "column" we are interested in + * @param columns the columns we are interested in + * @return a mapping of key --> Row + * @throws Exception + */ + public static Map weakReadProtocol(String tablename, String[] keys, String columnFamily, List columns) throws Exception + { + Row row = null; + long startTime = System.currentTimeMillis(); + Map readMessages = new HashMap(); + for ( String key : keys ) + { + ReadMessage readMessage = new ReadMessage(tablename, key, columnFamily, columns); + readMessages.put(key, readMessage); + } + /* Performs the multiget in parallel */ + Map rows = doReadProtocol(readMessages); + /* + * Do the consistency checks for the keys that are being queried + * in the background. + */ + for ( String key : keys ) + { + List endpoints = StorageService.instance().getNLiveStorageEndPoint(key); + /* Remove the local storage endpoint from the list. */ + endpoints.remove( StorageService.getLocalStorageEndPoint() ); + if ( endpoints.size() > 0 && DatabaseDescriptor.getConsistencyCheck()) + StorageService.instance().doConsistencyCheck(row, endpoints, columnFamily, columns); + } + return rows; + } + + /** * This function executes the read protocol locally and should be used only if consistency is not a concern. * Read the data from the local disk and return if the row is NOT NULL. If the data is NULL do the read from * one of the other replicas (in the same data center if possible) till we get the data. In the event we get - * the data we perform consistency checks and figure out if any repairs need to be done to the replicas. + * the data we perform consistency checks and figure out if any repairs need to be done to the replicas. + * @param tablename name of the table that needs to be queried + * @param key key whose we are interested in + * @param columnFamily name of the "column" we are interested in + * @param start start index + * @param count the number of columns we are interested in + * @return the row associated with this key + * @throws Exception */ public static Row weakReadProtocol(String tablename, String key, String columnFamily, int start, int count) throws Exception { @@ -565,6 +804,55 @@ public class StorageProxy return row; } + /** + * This version is used when results for multiple keys needs to be + * retrieved. + * + * @param tablename name of the table that needs to be queried + * @param keys keys whose values we are interested in + * @param columnFamily name of the "column" we are interested in + * @param start start index + * @param count the number of columns we are interested in + * @return a mapping of key --> Row + * @throws Exception + */ + public static Map weakReadProtocol(String tablename, String[] keys, String columnFamily, int start, int count) throws Exception + { + Row row = null; + long startTime = System.currentTimeMillis(); + Map readMessages = new HashMap(); + for ( String key : keys ) + { + ReadMessage readMessage = new ReadMessage(tablename, key, columnFamily, start, count); + readMessages.put(key, readMessage); + } + /* Performs the multiget in parallel */ + Map rows = doReadProtocol(readMessages); + /* + * Do the consistency checks for the keys that are being queried + * in the background. + */ + for ( String key : keys ) + { + List endpoints = StorageService.instance().getNLiveStorageEndPoint(key); + /* Remove the local storage endpoint from the list. */ + endpoints.remove( StorageService.getLocalStorageEndPoint() ); + if ( endpoints.size() > 0 && DatabaseDescriptor.getConsistencyCheck()) + StorageService.instance().doConsistencyCheck(row, endpoints, columnFamily, start, count); + } + return rows; + } + + /** + * This version is used when retrieving a single key. + * + * @param tablename name of the table that needs to be queried + * @param key key whose we are interested in + * @param columnFamily name of the "column" we are interested in + * @param sinceTimestamp this is lower bound of the timestamp + * @return the row associated with this key + * @throws Exception + */ public static Row weakReadProtocol(String tablename, String key, String columnFamily, long sinceTimestamp) throws Exception { Row row = null; @@ -585,5 +873,42 @@ public class StorageProxy StorageService.instance().doConsistencyCheck(row, endpoints, columnFamily, sinceTimestamp); return row; } - + + /** + * This version is used when results for multiple keys needs to be + * retrieved. + * + * @param tablename name of the table that needs to be queried + * @param keys keys whose values we are interested in + * @param columnFamily name of the "column" we are interested in + * @param sinceTimestamp this is lower bound of the timestamp + * @return a mapping of key --> Row + * @throws Exception + */ + public static Map weakReadProtocol(String tablename, String[] keys, String columnFamily, long sinceTimestamp) throws Exception + { + Row row = null; + long startTime = System.currentTimeMillis(); + Map readMessages = new HashMap(); + for ( String key : keys ) + { + ReadMessage readMessage = new ReadMessage(tablename, key, columnFamily, sinceTimestamp); + readMessages.put(key, readMessage); + } + /* Performs the multiget in parallel */ + Map rows = doReadProtocol(readMessages); + /* + * Do the consistency checks for the keys that are being queried + * in the background. + */ + for ( String key : keys ) + { + List endpoints = StorageService.instance().getNLiveStorageEndPoint(key); + /* Remove the local storage endpoint from the list. */ + endpoints.remove( StorageService.getLocalStorageEndPoint() ); + if ( endpoints.size() > 0 && DatabaseDescriptor.getConsistencyCheck()) + StorageService.instance().doConsistencyCheck(row, endpoints, columnFamily, sinceTimestamp); + } + return rows; + } } diff --git a/src/org/apache/cassandra/service/StorageService.java b/src/org/apache/cassandra/service/StorageService.java index 4b9dffd6f0..aee3c46b6f 100644 --- a/src/org/apache/cassandra/service/StorageService.java +++ b/src/org/apache/cassandra/service/StorageService.java @@ -18,32 +18,20 @@ package org.apache.cassandra.service; -import java.io.File; -import java.io.IOException; -import java.lang.management.ManagementFactory; -import java.lang.reflect.InvocationTargetException; -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.*; 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 java.net.UnknownHostException; 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; @@ -68,22 +56,21 @@ import org.apache.cassandra.db.TouchVerbHandler; import org.apache.cassandra.dht.BootStrapper; import org.apache.cassandra.dht.BootstrapInitiateMessage; import org.apache.cassandra.dht.BootstrapMetadataVerbHandler; -import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.dht.OrderPreservingPartitioner; -import org.apache.cassandra.dht.RandomPartitioner; 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.FailureDetector; import org.apache.cassandra.gms.Gossiper; import org.apache.cassandra.gms.IEndPointStateChangeSubscriber; +import org.apache.cassandra.io.DataInputBuffer; +import org.apache.cassandra.io.ICompactSerializer; 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.CompactEndPointSerializationHelper; import org.apache.cassandra.net.EndPoint; import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.Message; @@ -94,13 +81,22 @@ import org.apache.cassandra.tools.MembershipCleanerVerbHandler; import org.apache.cassandra.tools.TokenUpdateVerbHandler; import org.apache.cassandra.utils.FileUtils; import org.apache.cassandra.utils.LogUtil; +import org.apache.commons.math.linear.RealMatrix; +import org.apache.commons.math.linear.RealMatrixImpl; +import org.apache.log4j.Logger; +import org.apache.cassandra.concurrent.*; +import org.apache.cassandra.db.*; +import org.apache.cassandra.net.io.*; +import org.apache.cassandra.gms.*; +import org.apache.cassandra.utils.*; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; -import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.data.Stat; +import org.apache.zookeeper.proto.WatcherEvent; /* * This abstraction contains the token/identifier of this node @@ -137,10 +133,14 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto public final static String dataFileVerbHandler_ = "DATA-FILE-VERB-HANDLER"; public final static String mbrshipCleanerVerbHandler_ = "MBRSHIP-CLEANER-VERB-HANDLER"; public final static String bsMetadataVerbHandler_ = "BS-METADATA-VERB-HANDLER"; + public final static String jobConfigurationVerbHandler_ = "JOB-CONFIGURATION-VERB-HANDLER"; + public final static String taskMetricVerbHandler_ = "TASK-METRIC-VERB-HANDLER"; + public final static String mapAssignmentVerbHandler_ = "MAP-ASSIGNMENT-VERB-HANDLER"; + public final static String reduceAssignmentVerbHandler_ = "REDUCE-ASSIGNMENT-VERB-HANDLER"; + public final static String mapCompletionVerbHandler_ = "MAP-COMPLETION-VERB-HANDLER"; public final static String calloutDeployVerbHandler_ = "CALLOUT-DEPLOY-VERB-HANDLER"; public final static String touchVerbHandler_ = "TOUCH-VERB-HANDLER"; - public static String rangeVerbHandler_ = "RANGE-VERB-HANDLER"; - + public static enum ConsistencyLevel { WEAK, @@ -168,19 +168,20 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto { return "http://" + tcpAddr_.getHost() + ":" + DatabaseDescriptor.getHttpPort(); } + + public static PartitionerType getPartitionerType() + { + return (DatabaseDescriptor.getHashingStrategy().equalsIgnoreCase(DatabaseDescriptor.ophf_)) ? PartitionerType.OPHF : PartitionerType.RANDOM; + } /** * This is a facade for the hashing * function used by the system for * partitioning. */ - public static Token token(String key) + public static BigInteger hash(String key) { - return partitioner_.getTokenForKey(key); - } - - public static IPartitioner getPartitioner() { - return partitioner_; + return partitioner_.hash(key); } public static enum BootstrapMode @@ -248,6 +249,8 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto * */ private IEndPointSnitch endPointSnitch_; + /* Uptime of this node - we use this to determine if a bootstrap can be performed by this node */ + private long uptime_ = 0L; /* This abstraction maintains the token/endpoint metadata information */ private TokenMetadata tokenMetadata_ = new TokenMetadata(); @@ -258,6 +261,11 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto * for a clean exit. */ private Set components_ = new HashSet(); + /* + * This boolean indicates if we are in loading state. If we are then we do not want any + * distributed algorithms w.r.t change in token state to kick in. + */ + private boolean isLoadState_ = false; /* Timer is used to disseminate load information */ private Timer loadTimer_ = new Timer(false); @@ -304,6 +312,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto public StorageService() throws Throwable { init(); + uptime_ = System.currentTimeMillis(); storageLoadBalancer_ = new StorageLoadBalancer(this); endPointSnitch_ = new EndPointSnitch(); @@ -318,6 +327,8 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.bootStrapInitiateDoneVerbHandler_, new StorageService.BootstrapInitiateDoneVerbHandler()); MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.bootStrapTerminateVerbHandler_, new StreamManager.BootstrapTerminateVerbHandler()); MessagingService.getMessagingInstance().registerVerbHandlers(HttpConnection.httpRequestVerbHandler_, new HttpRequestVerbHandler(this) ); + MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.tokenInfoVerbHandler_, new TokenInfoVerbHandler() ); + MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.locationInfoVerbHandler_, new LocationInfoVerbHandler() ); MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.dataFileVerbHandler_, new DataFileVerbHandler() ); MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.mbrshipCleanerVerbHandler_, new MembershipCleanerVerbHandler() ); MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.bsMetadataVerbHandler_, new BootstrapMetadataVerbHandler() ); @@ -433,21 +444,23 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto components_.add(component); } - static + private void initPartitioner() { - try + String hashingStrategy = DatabaseDescriptor.getHashingStrategy(); + if ( hashingStrategy.equalsIgnoreCase(DatabaseDescriptor.ophf_) ) { - Class cls = Class.forName(DatabaseDescriptor.getPartitionerClass()); - partitioner_ = (IPartitioner) cls.getConstructor().newInstance(); - } - catch (Exception e) + partitioner_ = new OrderPreservingHashPartitioner(); + } + else { - throw new RuntimeException(e); + partitioner_ = new RandomPartitioner(); } } public void start() throws Throwable - { + { + /* Set up the partitioner */ + initPartitioner(); /* Start the DB */ storageMetadata_ = DBManager.instance().start(); /* Set up TCP endpoint */ @@ -487,6 +500,23 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto Gossiper.instance().addApplicationState(StorageService.nodeId_, new ApplicationState(storageMetadata_.getStorageId().toString())); } + private void startMapReduceFramework() + { + // TODO: This is a null pointer exception if JobTrackerHost is not in + // the config file. Also, shouldn't this comparison be done by IP + // instead of host name? We could have a match but not a textual + // match (e.g. somehost.vip vs somehost.vip.domain.com) + if ( DatabaseDescriptor.getJobTrackerAddress().equals( StorageService.tcpAddr_.getHost() ) ) + { +// JobTracker.instance().start(); +// TaskTracker.instance().start(); + } + else + { +// TaskTracker.instance().start(); + } + } + public void killMe() throws Throwable { isShutdown_.set(true); @@ -539,7 +569,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto } /* TODO: remove later */ - public void updateTokenMetadata(Token token, EndPoint endpoint) + public void updateTokenMetadata(BigInteger token, EndPoint endpoint) { tokenMetadata_.update(token, endpoint); } @@ -582,11 +612,47 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto consistencyManager_.submit(consistencySentinel); } + /* + * This method displays all the ranges and the replicas + * that are responsible for the individual ranges. The + * format of this string is the following: + * + * R1 : A B C + * R2 : D E F + * R3 : G H I + */ + public String showTheRing() + { + StringBuilder sb = new StringBuilder(); + /* Get the token to endpoint map. */ + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + Set tokens = tokenToEndPointMap.keySet(); + /* All the ranges for the tokens */ + Range[] ranges = getAllRanges(tokens); + Map> oldRangeToEndPointMap = constructRangeToEndPointMap(ranges); + + Set rangeSet = oldRangeToEndPointMap.keySet(); + for ( Range range : rangeSet ) + { + sb.append(range); + sb.append(" : "); + + List replicas = oldRangeToEndPointMap.get(range); + for ( EndPoint replica : replicas ) + { + sb.append(replica); + sb.append(" "); + } + sb.append(System.getProperty("line.separator")); + } + return sb.toString(); + } + public Map> getRangeToEndPointMap() { /* Get the token to endpoint map. */ - Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - Set tokens = tokenToEndPointMap.keySet(); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + Set tokens = tokenToEndPointMap.keySet(); /* All the ranges for the tokens */ Range[] ranges = getAllRanges(tokens); Map> oldRangeToEndPointMap = constructRangeToEndPointMap(ranges); @@ -619,7 +685,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>(); @@ -641,7 +707,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 ) { @@ -649,6 +715,77 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto } return endPointToRangesMap; } + + /** + * Get the estimated disk space of the target endpoint in its + * primary range. + * @param target whose primary range we are interested in. + * @return disk space of the target in the primary range. + */ + private double getDiskSpaceForPrimaryRange(EndPoint target) + { + double primaryDiskSpace = 0d; + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + Set tokens = tokenToEndPointMap.keySet(); + Range[] allRanges = getAllRanges(tokens); + Arrays.sort(allRanges); + /* Mapping from Range to its ordered position on the ring */ + Map rangeIndex = new HashMap(); + for ( int i = 0; i < allRanges.length; ++i ) + { + rangeIndex.put(allRanges[i], i); + } + /* Get the coefficients for the equations */ + List equations = new ArrayList(); + /* Get the endpoint to range map */ + Map> endPointToRangesMap = constructEndPointToRangesMap(); + Set eps = endPointToRangesMap.keySet(); + + for ( EndPoint ep : eps ) + { + List ranges = endPointToRangesMap.get(ep); + double[] equation = new double[allRanges.length]; + for ( Range range : ranges ) + { + int index = rangeIndex.get(range); + equation[index] = 1; + } + equations.add(equation); + } + double[][] coefficients = equations.toArray( new double[0][0] ); + + /* Get the constants which are the aggregate disk space for each endpoint */ + double[] constants = new double[allRanges.length]; + int index = 0; + for ( EndPoint ep : eps ) + { + /* reset the port back to control port */ + ep.setPort(DatabaseDescriptor.getControlPort()); + String lInfo = null; + if ( ep.equals(StorageService.udpAddr_) ) + lInfo = getLoadInfo(); + else + lInfo = getLoadInfo(ep); + LoadInfo li = new LoadInfo(lInfo); + constants[index++] = FileUtils.stringToFileSize(li.diskSpace()); + } + + RealMatrix matrix = new RealMatrixImpl(coefficients); + double[] solutions = matrix.solve(constants); + Range primaryRange = getPrimaryRangeForEndPoint(target); + primaryDiskSpace = solutions[rangeIndex.get(primaryRange)]; + return primaryDiskSpace; + } + + /** + * This is very dangerous. This is used only on the client + * side to set up the client library. This is then used to + * find the appropriate nodes to route the key to. + */ + public void setTokenMetadata(TokenMetadata tokenMetadata) + { + tokenMetadata_ = tokenMetadata; + } /** * Called when there is a change in application state. In particular @@ -662,9 +799,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto ApplicationState nodeIdState = epState.getApplicationState(StorageService.nodeId_); if (nodeIdState != null) { - Token newToken = getPartitioner().getTokenFactory().fromString(nodeIdState.getState()); + BigInteger newToken = new BigInteger(nodeIdState.getState()); logger_.debug("CHANGE IN STATE FOR " + endpoint + " - has token " + nodeIdState.getState()); - Token oldToken = tokenMetadata_.getToken(ep); + BigInteger oldToken = tokenMetadata_.getToken(ep); if ( oldToken != null ) { @@ -722,6 +859,17 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto } } + public static BigInteger generateRandomToken() + { + byte[] randomBytes = new byte[24]; + Random random = new Random(); + for ( int i = 0 ; i < 24 ; i++) + { + randomBytes[i] = (byte)(31 + random.nextInt(256 - 31)); + } + return hash(new String(randomBytes)); + } + /** * Get the count of primary keys from the sampler. */ @@ -741,12 +889,43 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto LoadInfo li = storageLoadBalancer_.getLoad(ep); return ( li == null ) ? "N/A" : li.toString(); } + + /** + * Get the endpoint that has the largest primary count. + * @return + */ + EndPoint getEndPointWithLargestPrimaryCount() + { + Set allMbrs = Gossiper.instance().getAllMembers(); + Map loadInfoToEndPointMap = new HashMap(); + List lInfos = new ArrayList(); + + for ( EndPoint mbr : allMbrs ) + { + mbr.setPort(DatabaseDescriptor.getStoragePort()); + LoadInfo li = null; + if ( mbr.equals(StorageService.tcpAddr_) ) + { + li = new LoadInfo( getLoadInfo() ); + lInfos.add( li ); + } + else + { + li = storageLoadBalancer_.getLoad(mbr); + lInfos.add( li ); + } + loadInfoToEndPointMap.put(li, mbr); + } + + Collections.sort(lInfos, new LoadInfo.DiskSpaceComparator()); + return loadInfoToEndPointMap.get( lInfos.get(lInfos.size() - 1) ); + } /* * This method updates the token on disk and modifies the cached * StorageMetadata instance. This is only for the local endpoint. */ - public void updateToken(Token token) throws IOException + public void updateToken(BigInteger token) throws IOException { /* update the token on disk */ SystemTable.openSystemTable(SystemTable.name_).updateToken(token); @@ -787,12 +966,13 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto { if ( keys.length > 0 ) { - Token token = tokenMetadata_.getToken(StorageService.tcpAddr_); - Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - Token[] tokens = tokenToEndPointMap.keySet().toArray(new Token[tokenToEndPointMap.keySet().size()]); + isLoadState_ = true; + BigInteger token = tokenMetadata_.getToken(StorageService.tcpAddr_); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + BigInteger[] tokens = tokenToEndPointMap.keySet().toArray( new BigInteger[0] ); Arrays.sort(tokens); int index = Arrays.binarySearch(tokens, token) * (keys.length/tokens.length); - Token newToken = token( keys[index] ); + BigInteger newToken = hash( keys[index] ); /* update the token */ updateToken(newToken); } @@ -804,6 +984,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto */ public void resetLoadState() { + isLoadState_ = false; } /** @@ -830,7 +1011,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto } String[] allNodes = nodesToLoad.split(":"); EndPoint[] endpoints = new EndPoint[allNodes.length]; - Token[] tokens = new Token[allNodes.length]; + BigInteger[] tokens = new BigInteger[allNodes.length]; for ( int i = 0; i < allNodes.length; ++i ) { @@ -866,8 +1047,8 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto switch ( mode ) { case FULL: - Token token = tokenMetadata_.getToken(endpoint); - bootStrapper_.submit(new BootStrapper(new EndPoint[]{endpoint}, token)); + BigInteger token = tokenMetadata_.getToken(endpoint); + bootStrapper_.submit( new BootStrapper(new EndPoint[]{endpoint}, new BigInteger[]{token}) ); break; case HINT: @@ -885,14 +1066,26 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto public String getToken(EndPoint ep) { EndPoint ep2 = new EndPoint(ep.getHost(), DatabaseDescriptor.getStoragePort()); - Token token = tokenMetadata_.getToken(ep2); - return ( token == null ) ? "" : token.toString(); + BigInteger token = tokenMetadata_.getToken(ep2); + return ( token == null ) ? BigInteger.ZERO.toString() : 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() { @@ -921,6 +1114,12 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto doBootstrap(nodes); } + public String getAppropriateToken(int count) + { + BigInteger token = BootstrapAndLbHelper.getTokenBasedOnPrimaryCount(count); + return token.toString(); + } + public void doGC() { List tables = DatabaseDescriptor.getTables(); @@ -975,9 +1174,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto */ EndPoint getPredecessor(EndPoint ep) { - Token token = tokenMetadata_.getToken(ep); - Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - List tokens = new ArrayList(tokenToEndPointMap.keySet()); + BigInteger token = tokenMetadata_.getToken(ep); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + List tokens = new ArrayList(tokenToEndPointMap.keySet()); Collections.sort(tokens); int index = Collections.binarySearch(tokens, token); EndPoint predecessor = (index == 0) ? tokenToEndPointMap.get(tokens @@ -992,9 +1191,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto */ public EndPoint getSuccessor(EndPoint ep) { - Token token = tokenMetadata_.getToken(ep); - Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - List tokens = new ArrayList(tokenToEndPointMap.keySet()); + BigInteger token = tokenMetadata_.getToken(ep); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + List tokens = new ArrayList(tokenToEndPointMap.keySet()); Collections.sort(tokens); int index = Collections.binarySearch(tokens, token); EndPoint successor = (index == (tokens.size() - 1)) ? tokenToEndPointMap @@ -1003,6 +1202,35 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto return successor; } + /** + * This method returns the range handled by this node. + */ + public Range getMyRange() + { + BigInteger myToken = tokenMetadata_.getToken(StorageService.tcpAddr_); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + List allTokens = new ArrayList(tokenToEndPointMap.keySet()); + Collections.sort(allTokens); + int index = Collections.binarySearch(allTokens, myToken); + /* Calculate the lhs for the range */ + BigInteger lhs = (index == 0) ? allTokens.get(allTokens.size() - 1) : allTokens.get( index - 1); + return new Range( lhs, myToken ); + } + + /** + * Get the primary for the given range. Use the replica placement + * strategies to determine which are the replicas. The first replica + * in the list is the primary. + * + * @param range on the ring. + * @return endpoint responsible for the range. + */ + public EndPoint getPrimaryStorageEndPointForRange(Range range) + { + EndPoint[] replicas = nodePicker_.getStorageEndPoints(range.left()); + return replicas[0]; + } + /** * Get the primary range for the specified endpoint. * @param ep endpoint we are interested in. @@ -1010,9 +1238,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto */ public Range getPrimaryRangeForEndPoint(EndPoint ep) { - Token right = tokenMetadata_.getToken(ep); + BigInteger right = tokenMetadata_.getToken(ep); EndPoint predecessor = getPredecessor(ep); - Token left = tokenMetadata_.getToken(predecessor); + BigInteger left = tokenMetadata_.getToken(predecessor); return new Range(left, right); } @@ -1036,17 +1264,28 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto return ranges; } - + + /** + * Get all ranges that span the ring as per + * current snapshot of the token distribution. + * @return all ranges in sorted order. + */ + public Range[] getAllRanges() + { + Set allTokens = tokenMetadata_.cloneTokenEndPointMap().keySet(); + return getAllRanges( allTokens ); + } + /** * Get all ranges that span the ring given a set * of tokens. All ranges are in sorted order of * 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 ) @@ -1059,6 +1298,20 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto return ranges.toArray( new Range[0] ); } + /** + * Get all ranges that span the ring given a set + * of endpoints. + */ + public Range[] getPrimaryRangesForEndPoints(Set endpoints) + { + List allRanges = new ArrayList(); + for ( EndPoint endpoint : endpoints ) + { + allRanges.add( getPrimaryRangeForEndPoint( endpoint) ); + } + return allRanges.toArray(new Range[0]); + } + /** * This method returns the endpoint that is responsible for storing the * specified key. @@ -1069,9 +1322,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto public EndPoint getPrimary(String key) { EndPoint endpoint = StorageService.tcpAddr_; - Token token = token(key); - Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - List tokens = new ArrayList(tokenToEndPointMap.keySet()); + BigInteger token = hash(key); + Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); + List tokens = new ArrayList(tokenToEndPointMap.keySet()); if (tokens.size() > 0) { Collections.sort(tokens); @@ -1107,7 +1360,63 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto EndPoint endpoint = getPrimary(key); return StorageService.tcpAddr_.equals(endpoint); } - + + /** + * This method determines whether the target endpoint is the + * primary for the given key. + * @param key + * @param target the target enpoint + * @return true if the local endpoint is the primary replica. + */ + public boolean isPrimary(String key, EndPoint target) + { + EndPoint endpoint = getPrimary(key); + return target.equals(endpoint); + } + + /** + * This method determines whether the local endpoint is the + * seondary replica for the given key. + * @param key + * @return true if the local endpoint is the secondary replica. + */ + public boolean isSecondary(String key) + { + EndPoint[] topN = getNStorageEndPoint(key); + if ( topN.length < DatabaseDescriptor.getReplicationFactor() ) + return false; + return topN[1].equals(StorageService.tcpAddr_); + } + + /** + * This method determines whether the local endpoint is the + * seondary replica for the given key. + * @param key + * @return true if the local endpoint is the tertiary replica. + */ + public boolean isTertiary(String key) + { + EndPoint[] topN = getNStorageEndPoint(key); + if ( topN.length < DatabaseDescriptor.getReplicationFactor() ) + return false; + return topN[2].equals(StorageService.tcpAddr_); + } + + /** + * This method determines if the local endpoint is + * in the topN of N nodes passed in. + */ + public boolean isInTopN(String key) + { + EndPoint[] topN = getNStorageEndPoint(key); + for ( EndPoint ep : topN ) + { + if ( ep.equals( StorageService.tcpAddr_ ) ) + return true; + } + return false; + } + /** * This method returns the N endpoints that are responsible for storing the * specified key i.e for replication. @@ -1117,11 +1426,16 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto */ public EndPoint[] getNStorageEndPoint(String key) { - Token token = token(key); + BigInteger token = hash(key); return nodePicker_.getStorageEndPoints(token); } - - + + private Map getNStorageEndPoints(String[] keys) + { + return nodePicker_.getStorageEndPoints(keys); + } + + /** * This method attempts to return N endpoints that are responsible for storing the * specified key i.e for replication. @@ -1152,7 +1466,22 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto */ public Map getNStorageEndPointMap(String key) { - Token token = token(key); + BigInteger token = hash(key); + return nodePicker_.getHintedStorageEndPoints(token); + } + + /** + * This method returns the N endpoints that are responsible for storing the + * specified key i.e for replication. But it makes sure that the N endpoints + * that are returned are live as reported by the FD. It returns the hint information + * if some nodes in the top N are not live. + * + * param @ key - key for which we need to find the endpoint return value - + * the endpoint responsible for this key + */ + public Map getNHintedStorageEndPoint(String key) + { + BigInteger token = hash(key); return nodePicker_.getHintedStorageEndPoints(token); } @@ -1162,7 +1491,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto * * param @ token - position on the ring */ - public EndPoint[] getNStorageEndPoint(Token token) + public EndPoint[] getNStorageEndPoint(BigInteger token) { return nodePicker_.getStorageEndPoints(token); } @@ -1175,11 +1504,24 @@ 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(Token token, Map tokenToEndPointMap) + protected EndPoint[] getNStorageEndPoint(BigInteger token, Map tokenToEndPointMap) { return nodePicker_.getStorageEndPoints(token, tokenToEndPointMap); } + /** + * This method returns the N endpoints that are responsible for storing the + * specified key i.e for replication. But it makes sure that the N endpoints + * that are returned are live as reported by the FD. It returns the hint information + * if some nodes in the top N are not live. + * + * param @ token - position on the ring + */ + public Map getNHintedStorageEndPoint(BigInteger token) + { + return nodePicker_.getHintedStorageEndPoints(token); + } + /** * This function finds the most suitable endpoint given a key. * It checks for loclity and alive test. @@ -1217,5 +1559,58 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto } return null; } - + + public Map findSuitableEndPoints(String[] keys) throws IOException + { + Map suitableEndPoints = new HashMap(); + Map results = getNStorageEndPoints(keys); + for ( String key : keys ) + { + EndPoint[] endpoints = results.get(key); + /* indicates if we have to move on to the next key */ + boolean moveOn = false; + for(EndPoint endPoint: endpoints) + { + if(endPoint.equals(StorageService.getLocalStorageEndPoint())) + { + suitableEndPoints.put(key, endPoint); + moveOn = true; + break; + } + } + + if ( moveOn ) + continue; + + int j = 0; + for ( ; j < endpoints.length; ++j ) + { + if ( StorageService.instance().isInSameDataCenter(endpoints[j]) && FailureDetector.instance().isAlive(endpoints[j]) ) + { + logger_.debug("EndPoint " + endpoints[j] + " is in the same data center as local storage endpoint."); + suitableEndPoints.put(key, endpoints[j]); + moveOn = true; + break; + } + } + + if ( moveOn ) + continue; + + // We have tried to be really nice but looks like theer are no servers + // in the local data center that are alive and can service this request so + // just send it to the first alive guy and see if we get anything. + j = 0; + for ( ; j < endpoints.length; ++j ) + { + if ( FailureDetector.instance().isAlive(endpoints[j]) ) + { + logger_.debug("EndPoint " + endpoints[j] + " is alive so get data from it."); + suitableEndPoints.put(key, endpoints[j]); + break; + } + } + } + return suitableEndPoints; + } } diff --git a/src/org/apache/cassandra/service/StorageServiceMBean.java b/src/org/apache/cassandra/service/StorageServiceMBean.java index 1acbcfdb3a..444113eb9e 100644 --- a/src/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/org/apache/cassandra/service/StorageServiceMBean.java @@ -19,6 +19,9 @@ package org.apache.cassandra.service; import java.io.IOException; +import java.math.BigInteger; + +import org.apache.cassandra.net.EndPoint; /** @@ -43,11 +46,25 @@ public interface StorageServiceMBean */ public void loadAll(String nodes); + /** + * This method is used only for debug purpose. + */ + public void updateToken(String token); + /** * */ public void doGC(); - + + /** + * Get the token such that the range of this node + * is split after count number of keys. + * @param count number of keys after which to generate + * token. + * @return appropriate token + */ + public String getAppropriateToken(int count); + /** * Stream the files in the bootstrap directory over to the * node being bootstrapped. This is used in case of normal diff --git a/src/org/apache/cassandra/service/TokenUpdateVerbHandler.java b/src/org/apache/cassandra/service/TokenUpdateVerbHandler.java index 6dcaa31f40..90384e366c 100644 --- a/src/org/apache/cassandra/service/TokenUpdateVerbHandler.java +++ b/src/org/apache/cassandra/service/TokenUpdateVerbHandler.java @@ -19,13 +19,12 @@ 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 ) @@ -38,7 +37,7 @@ public class TokenUpdateVerbHandler implements IVerbHandler public void doVerb(Message message) { byte[] body = (byte[])message.getMessageBody()[0]; - Token token = StorageService.getPartitioner().getTokenFactory().fromByteArray(body); + BigInteger token = new BigInteger(body); try { logger_.info("Updating the token to [" + token + "]"); diff --git a/src/org/apache/cassandra/service/WriteResponseResolver.java b/src/org/apache/cassandra/service/WriteResponseResolver.java index 06c53eca2a..ac93c60a33 100644 --- a/src/org/apache/cassandra/service/WriteResponseResolver.java +++ b/src/org/apache/cassandra/service/WriteResponseResolver.java @@ -20,7 +20,7 @@ package org.apache.cassandra.service; import java.util.List; -import org.apache.cassandra.db.WriteResponse; +import org.apache.cassandra.db.WriteResponseMessage; import org.apache.cassandra.net.Message; import org.apache.log4j.Logger; @@ -47,11 +47,11 @@ public class WriteResponseResolver implements IResponseResolver { boolean returnValue = false; for (Message response : responses) { Object[] body = response.getMessageBody(); - WriteResponse writeResponse = (WriteResponse) body[0]; - boolean result = writeResponse.isSuccess(); + WriteResponseMessage writeResponseMessage = (WriteResponseMessage) body[0]; + boolean result = writeResponseMessage.isSuccess(); if (!result) { logger_.debug("Write at " + response.getFrom() - + " may have failed for the key " + writeResponse.key()); + + " may have failed for the key " + writeResponseMessage.key()); } returnValue |= result; }