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

Basic implementation of multiget() functionality. Fix to how read-repair is done in the ConsistencyManager. 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@759028 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Avinash Lakshman 2009-03-27 05:41:59 +00:00
parent 52062900ca
commit 1cc4294d34
12 changed files with 1257 additions and 538 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

395
src/org/apache/cassandra/service/CassandraServer.java
View File

@ -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<String, ColumnFamily> cfMap = row.getColumnFamilyMap();
Map<String, ColumnFamily> 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<String, ColumnFamily> cfMap = row.getColumnFamilyMap();
Map<String, ColumnFamily> 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<String, ColumnFamily> cfMap = row.getColumnFamilyMap();
Map<String, ColumnFamily> 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<String, ColumnFamily> cfMap = row.getColumnFamilyMap();
Map<String, ColumnFamily> 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<String, ColumnFamily> cfMap = row.getColumnFamilyMap();
Map<String, ColumnFamily> 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<Boolean> writeResponseResolver = new WriteResponseResolver();
QuorumResponseHandler<Boolean> quorumResponseHandler = new QuorumResponseHandler<Boolean>(
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<column_t> 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<column_t> 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<column_t> 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<superColumn_t> get_slice_super_by_names(String tablename, String key, String columnFamily, List<String> 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<String, ColumnFamily> cfMap = row.getColumnFamilyMap();
Map<String, ColumnFamily> 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<String, ColumnFamily> cfMap = row.getColumnFamilyMap();
Map<String, ColumnFamily> 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<superColumn_t> 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<superColumn_t> 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<superColumn_t> 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<String> 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<String> keys = new ArrayList<String>();
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();

71
src/org/apache/cassandra/service/ConsistencyManager.java
View File

@ -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;
}
}

11
src/org/apache/cassandra/service/HttpRequestVerbHandler.java
View File

@ -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

8
src/org/apache/cassandra/service/QuorumResponseHandler.java
View File

@ -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<T> implements IAsyncCallback
lock_.unlock();
}
}
public void attachContext(Object o)
{
throw new UnsupportedOperationException("This operation is not supported in this version of the callback handler");
}
}

152
src/org/apache/cassandra/service/RangeVerbHandler.java
View File

@ -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> STRING_COMPARATOR = new Comparator<String>()
{
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<Iterator<String>> iterators = new ArrayList<Iterator<String>>();
Table table = Table.open(DatabaseDescriptor.getTableName());
for (String cfName : table.getApplicationColumnFamilies())
{
ColumnFamilyStore cfs = table.getColumnFamilyStore(cfName);
// memtable keys: current and historical
Iterator<Memtable> it = (Iterator<Memtable>) 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<String> iter = IteratorUtils.collatedIterator(STRING_COMPARATOR, iterators);
List<String> keys = new ArrayList<String>();
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());
}
}

16
src/org/apache/cassandra/service/ReadResponseResolver.java
View File

@ -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<Row>
{
private static Logger logger_ = Logger.getLogger(WriteResponseResolver.class);
/*
@ -101,15 +100,16 @@ public class ReadResponseResolver implements IResponseResolver<Row>
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<Row>
continue;
// create the row mutation message based on the diff and schedule a read repair
RowMutation rowMutation = new RowMutation(table, key);
Map<String, ColumnFamily> columnFamilies = diffRow.getColumnFamilyMap();
Map<String, ColumnFamily> columnFamilies = diffRow.getColumnFamilies();
Set<String> 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

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

@ -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_;
}

539
src/org/apache/cassandra/service/StorageProxy.java
View File

@ -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<EndPoint, Message> createWriteMessages(RowMutation rm, Map<EndPoint, EndPoint> endpointMap) throws IOException
private static Map<EndPoint, Message> createWriteMessages(RowMutationMessage rmMessage, Map<EndPoint, EndPoint> endpointMap) throws IOException
{
Map<EndPoint, Message> messageMap = new HashMap<EndPoint, Message>();
Message message = rm.makeRowMutationMessage();
for (Map.Entry<EndPoint, EndPoint> entry : endpointMap.entrySet())
{
EndPoint target = entry.getKey();
EndPoint hint = entry.getValue();
Map<EndPoint, Message> messageMap = new HashMap<EndPoint, Message>();
Message message = RowMutationMessage.makeRowMutationMessage(rmMessage);
Set<EndPoint> 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<EndPoint, EndPoint> endpointMap = StorageService.instance().getNStorageEndPointMap(rm.key());
// TODO: throw a thrift exception if we do not have N nodes
Map<EndPoint, Message> messageMap = createWriteMessages(rm, endpointMap);
logger_.debug("insert writing to [" + StringUtils.join(messageMap.keySet(), ", ") + "]");
for (Map.Entry<EndPoint, Message> entry : messageMap.entrySet())
{
MessagingService.getMessagingInstance().sendOneWay(entry.getValue(), entry.getKey());
}
}
try
{
logger_.debug(" insert");
Map<EndPoint, EndPoint> 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<EndPoint, Message> messageMap = createWriteMessages(rmMsg, endpointMap);
Set<EndPoint> 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<String, Message> constructMessages(Map<String, ReadMessage> readMessages) throws IOException
{
assert rm.key() != null;
try
Map<String, Message> messages = new HashMap<String, Message>();
Set<String> keys = readMessages.keySet();
for ( String key : keys )
{
Message message = rm.makeRowMutationMessage();
IResponseResolver<Boolean> writeResponseResolver = new WriteResponseResolver();
QuorumResponseHandler<Boolean> quorumResponseHandler = new QuorumResponseHandler<Boolean>(
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<String, EndPoint> endPoints, Map<String, Message> messages)
{
Set<String> 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<String, Row> doReadProtocol(Map<String, ReadMessage> readMessages) throws IOException,TimeoutException
{
Map<String, Row> rows = new HashMap<String, Row>();
Set<String> keys = readMessages.keySet();
/* Find all the suitable endpoints for the keys */
Map<String, EndPoint> endPoints = StorageService.instance().findSuitableEndPoints(keys.toArray( new String[0] ));
/* Construct the messages to be sent out */
Map<String, Message> messages = constructMessages(readMessages);
/* Dispatch the messages to the respective endpoints */
IAsyncResult iar = dispatchMessages(endPoints, messages);
List<Object[]> 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<String> 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<String, Row> readProtocol(String tablename, String[] keys, String columnFamily, int start, int count, StorageService.ConsistencyLevel consistencyLevel) throws Exception
{
Map<String, Row> rows = new HashMap<String, Row>();
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<String, Row> strongReadProtocol(String tablename, String[] keys, String columnFamily, int start, int count) throws IOException, TimeoutException
{
Map<String, Row> rows = new HashMap<String, Row>();
long startTime = System.currentTimeMillis();
// TODO: throw a thrift exception if we do not have N nodes
Map<String, ReadMessage[]> readMessages = new HashMap<String, ReadMessage[]>();
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<String, Message[]> constructReplicaMessages(Map<String, ReadMessage[]> readMessages) throws IOException
{
Map<String, Message[]> messages = new HashMap<String, Message[]>();
Set<String> 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<String, ReadMessage[]> readMessages, Map<String, Message[]> messages) throws IOException
{
Set<String> keys = messages.keySet();
/* This maps the keys to the original data read messages */
Map<String, ReadMessage> readMessage = new HashMap<String, ReadMessage>();
/* This maps the keys to their respective endpoints/replicas */
Map<String, EndPoint[]> endpoints = new HashMap<String, EndPoint[]>();
/* 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<EndPoint> replicas = new ArrayList<EndPoint>( 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<String, Row> doStrongReadProtocol(Map<String, ReadMessage[]> readMessages) throws IOException
{
Map<String, Row> rows = new HashMap<String, Row>();
/* Construct the messages to be sent to the replicas */
Map<String, Message[]> 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<String> 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<String, Row> weakReadProtocol(String tablename, String[] keys, String columnFamily, List<String> columns) throws Exception
{
Row row = null;
long startTime = System.currentTimeMillis();
Map<String, ReadMessage> readMessages = new HashMap<String, ReadMessage>();
for ( String key : keys )
{
ReadMessage readMessage = new ReadMessage(tablename, key, columnFamily, columns);
readMessages.put(key, readMessage);
}
/* Performs the multiget in parallel */
Map<String, Row> rows = doReadProtocol(readMessages);
/*
* Do the consistency checks for the keys that are being queried
* in the background.
*/
for ( String key : keys )
{
List<EndPoint> 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<String, Row> weakReadProtocol(String tablename, String[] keys, String columnFamily, int start, int count) throws Exception
{
Row row = null;
long startTime = System.currentTimeMillis();
Map<String, ReadMessage> readMessages = new HashMap<String, ReadMessage>();
for ( String key : keys )
{
ReadMessage readMessage = new ReadMessage(tablename, key, columnFamily, start, count);
readMessages.put(key, readMessage);
}
/* Performs the multiget in parallel */
Map<String, Row> rows = doReadProtocol(readMessages);
/*
* Do the consistency checks for the keys that are being queried
* in the background.
*/
for ( String key : keys )
{
List<EndPoint> 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<String, Row> weakReadProtocol(String tablename, String[] keys, String columnFamily, long sinceTimestamp) throws Exception
{
Row row = null;
long startTime = System.currentTimeMillis();
Map<String, ReadMessage> readMessages = new HashMap<String, ReadMessage>();
for ( String key : keys )
{
ReadMessage readMessage = new ReadMessage(tablename, key, columnFamily, sinceTimestamp);
readMessages.put(key, readMessage);
}
/* Performs the multiget in parallel */
Map<String, Row> rows = doReadProtocol(readMessages);
/*
* Do the consistency checks for the keys that are being queried
* in the background.
*/
for ( String key : keys )
{
List<EndPoint> 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;
}
}

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

@ -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<IComponentShutdown> components_ = new HashSet<IComponentShutdown>();
/*
* 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<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Set<BigInteger> tokens = tokenToEndPointMap.keySet();
/* All the ranges for the tokens */
Range[] ranges = getAllRanges(tokens);
Map<Range, List<EndPoint>> oldRangeToEndPointMap = constructRangeToEndPointMap(ranges);
Set<Range> rangeSet = oldRangeToEndPointMap.keySet();
for ( Range range : rangeSet )
{
sb.append(range);
sb.append(" : ");
List<EndPoint> replicas = oldRangeToEndPointMap.get(range);
for ( EndPoint replica : replicas )
{
sb.append(replica);
sb.append(" ");
}
sb.append(System.getProperty("line.separator"));
}
return sb.toString();
}
public Map<Range, List<EndPoint>> getRangeToEndPointMap()
{
/* Get the token to endpoint map. */
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Set<Token> tokens = tokenToEndPointMap.keySet();
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Set<BigInteger> tokens = tokenToEndPointMap.keySet();
/* All the ranges for the tokens */
Range[] ranges = getAllRanges(tokens);
Map<Range, List<EndPoint>> 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<Range, List<EndPoint>> constructRangeToEndPointMap(Range[] ranges, Map<Token, EndPoint> tokenToEndPointMap)
public Map<Range, List<EndPoint>> constructRangeToEndPointMap(Range[] ranges, Map<BigInteger, EndPoint> tokenToEndPointMap)
{
logger_.debug("Constructing range to endpoint map ...");
Map<Range, List<EndPoint>> rangeToEndPointMap = new HashMap<Range, List<EndPoint>>();
@ -641,7 +707,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
public Map<EndPoint, List<Range>> constructEndPointToRangesMap()
{
Map<EndPoint, List<Range>> endPointToRangesMap = new HashMap<EndPoint, List<Range>>();
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Collection<EndPoint> 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<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Set<BigInteger> tokens = tokenToEndPointMap.keySet();
Range[] allRanges = getAllRanges(tokens);
Arrays.sort(allRanges);
/* Mapping from Range to its ordered position on the ring */
Map<Range, Integer> rangeIndex = new HashMap<Range, Integer>();
for ( int i = 0; i < allRanges.length; ++i )
{
rangeIndex.put(allRanges[i], i);
}
/* Get the coefficients for the equations */
List<double[]> equations = new ArrayList<double[]>();
/* Get the endpoint to range map */
Map<EndPoint, List<Range>> endPointToRangesMap = constructEndPointToRangesMap();
Set<EndPoint> eps = endPointToRangesMap.keySet();
for ( EndPoint ep : eps )
{
List<Range> 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<EndPoint> allMbrs = Gossiper.instance().getAllMembers();
Map<LoadInfo, EndPoint> loadInfoToEndPointMap = new HashMap<LoadInfo, EndPoint>();
List<LoadInfo> lInfos = new ArrayList<LoadInfo>();
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<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
Token[] tokens = tokenToEndPointMap.keySet().toArray(new Token[tokenToEndPointMap.keySet().size()]);
isLoadState_ = true;
BigInteger token = tokenMetadata_.getToken(StorageService.tcpAddr_);
Map<BigInteger, EndPoint> 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<String> tables = DatabaseDescriptor.getTables();
@ -975,9 +1174,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
*/
EndPoint getPredecessor(EndPoint ep)
{
Token token = tokenMetadata_.getToken(ep);
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List tokens = new ArrayList<Token>(tokenToEndPointMap.keySet());
BigInteger token = tokenMetadata_.getToken(ep);
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List<BigInteger> tokens = new ArrayList<BigInteger>(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<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List tokens = new ArrayList<Token>(tokenToEndPointMap.keySet());
BigInteger token = tokenMetadata_.getToken(ep);
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List<BigInteger> tokens = new ArrayList<BigInteger>(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<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List<BigInteger> allTokens = new ArrayList<BigInteger>(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<BigInteger> 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<Token> tokens)
public Range[] getAllRanges(Set<BigInteger> tokens)
{
List<Range> ranges = new ArrayList<Range>();
List<Token> allTokens = new ArrayList<Token>(tokens);
List<BigInteger> allTokens = new ArrayList<BigInteger>(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<EndPoint> endpoints)
{
List<Range> allRanges = new ArrayList<Range>();
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<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List tokens = new ArrayList<Token>(tokenToEndPointMap.keySet());
BigInteger token = hash(key);
Map<BigInteger, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
List<BigInteger> tokens = new ArrayList<BigInteger>(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<String, EndPoint[]> 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<EndPoint, EndPoint> 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<EndPoint, EndPoint> 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<Token, EndPoint> tokenToEndPointMap)
protected EndPoint[] getNStorageEndPoint(BigInteger token, Map<BigInteger, EndPoint> 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<EndPoint, EndPoint> 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<String, EndPoint> findSuitableEndPoints(String[] keys) throws IOException
{
Map<String, EndPoint> suitableEndPoints = new HashMap<String, EndPoint>();
Map<String, EndPoint[]> 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;
}
}

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

@ -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 <i>count</i> 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

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

@ -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 + "]");

8
src/org/apache/cassandra/service/WriteResponseResolver.java
View File

@ -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> {
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;
}