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

move id out of Message so we can do cross-DC forwarding again 

patch by jbellis; reviewed by gdusbabek for CASSANDRA-1530

git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1068101 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-02-07 20:28:10 +00:00
parent 55c153f2f6
commit 7f4e3ebc31
45 changed files with 134 additions and 162 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/java/org/apache/cassandra/db/BinaryVerbHandler.java
View File

@ -32,7 +32,7 @@ public class BinaryVerbHandler implements IVerbHandler
{
private static Logger logger_ = LoggerFactory.getLogger(BinaryVerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
byte[] bytes = message.getMessageBody();
ByteArrayInputStream buffer = new ByteArrayInputStream(bytes);
@ -45,8 +45,8 @@ public class BinaryVerbHandler implements IVerbHandler
WriteResponse response = new WriteResponse(rm.getTable(), rm.key(), true);
Message responseMessage = WriteResponse.makeWriteResponseMessage(message, response);
if (logger_.isDebugEnabled())
logger_.debug("binary " + rm + " applied. Sending response to " + message.getMessageId() + "@" + message.getFrom());
MessagingService.instance().sendOneWay(responseMessage, message.getFrom());
logger_.debug("binary " + rm + " applied. Sending response to " + id + "@" + message.getFrom());
MessagingService.instance().sendReply(responseMessage, id, message.getFrom());
}
catch (Exception e)
{

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

@ -28,7 +28,7 @@ public class DefinitionsAnnounceVerbHandler implements IVerbHandler
{
/** someone is announcing their schema version. */
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
UUID theirVersion = UUID.fromString(new String(message.getMessageBody()));
MigrationManager.rectify(theirVersion, message.getFrom());

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

@ -42,7 +42,7 @@ public class DefinitionsUpdateResponseVerbHandler implements IVerbHandler
private static final Logger logger = LoggerFactory.getLogger(DefinitionsUpdateResponseVerbHandler.class);
/** someone sent me their data definitions */
public void doVerb(final Message message)
public void doVerb(final Message message, String id)
{
try
{

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

@ -33,7 +33,7 @@ public class ReadRepairVerbHandler implements IVerbHandler
{
private static Logger logger_ = LoggerFactory.getLogger(ReadRepairVerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
byte[] body = message.getMessageBody();
ByteArrayInputStream buffer = new ByteArrayInputStream(body);

6
src/java/org/apache/cassandra/db/ReadVerbHandler.java
View File

@ -45,7 +45,7 @@ public class ReadVerbHandler implements IVerbHandler
/* We use this so that we can reuse readcontext objects */
private static ThreadLocal<ReadVerbHandler.ReadContext> tls_ = new InheritableThreadLocal<ReadVerbHandler.ReadContext>();
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
byte[] body = message.getMessageBody();
/* Obtain a Read Context from TLS */
@ -79,8 +79,8 @@ public class ReadVerbHandler implements IVerbHandler
Message response = message.getReply(FBUtilities.getLocalAddress(), bytes);
if (logger_.isDebugEnabled())
logger_.debug(String.format("Read key %s; sending response to %s@%s",
ByteBufferUtil.bytesToHex(command.key), message.getMessageId(), message.getFrom()));
MessagingService.instance().sendOneWay(response, message.getFrom());
ByteBufferUtil.bytesToHex(command.key), id, message.getFrom()));
MessagingService.instance().sendReply(response, id, message.getFrom());
}
catch (IOException ex)
{

6
src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
View File

@ -40,7 +40,7 @@ public class RowMutationVerbHandler implements IVerbHandler
{
private static Logger logger_ = LoggerFactory.getLogger(RowMutationVerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
byte[] bytes = message.getMessageBody();
ByteArrayInputStream buffer = new ByteArrayInputStream(bytes);
@ -78,8 +78,8 @@ public class RowMutationVerbHandler implements IVerbHandler
WriteResponse response = new WriteResponse(rm.getTable(), rm.key(), true);
Message responseMessage = WriteResponse.makeWriteResponseMessage(message, response);
if (logger_.isDebugEnabled())
logger_.debug(rm + " applied. Sending response to " + message.getMessageId() + "@" + message.getFrom());
MessagingService.instance().sendOneWay(responseMessage, message.getFrom());
logger_.debug(rm + " applied. Sending response to " + id + "@" + message.getFrom());
MessagingService.instance().sendReply(responseMessage, id, message.getFrom());
}
catch (IOException e)
{

4
src/java/org/apache/cassandra/db/SchemaCheckVerbHandler.java
View File

@ -31,10 +31,10 @@ public class SchemaCheckVerbHandler implements IVerbHandler
{
private final Logger logger = LoggerFactory.getLogger(SchemaCheckVerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
logger.debug("Received schema check request.");
Message response = message.getInternalReply(DatabaseDescriptor.getDefsVersion().toString().getBytes());
MessagingService.instance().sendOneWay(response, message.getFrom());
MessagingService.instance().sendReply(response, id, message.getFrom());
}
}

7
src/java/org/apache/cassandra/db/TruncateVerbHandler.java
View File

@ -40,7 +40,7 @@ public class TruncateVerbHandler implements IVerbHandler
{
private static Logger logger = LoggerFactory.getLogger(TruncateVerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
byte[] bytes = message.getMessageBody();
ByteArrayInputStream buffer = new ByteArrayInputStream(bytes);
@ -76,9 +76,8 @@ public class TruncateVerbHandler implements IVerbHandler
TruncateResponse response = new TruncateResponse(t.keyspace, t.columnFamily, true);
Message responseMessage = TruncateResponse.makeTruncateResponseMessage(message, response);
logger.debug("{} applied. Sending response to {}@{} ",
new Object[]{t, message.getMessageId(), message.getFrom()});
MessagingService.instance().sendOneWay(responseMessage, message.getFrom());
logger.debug("{} applied. Sending response to {}@{} ", new Object[]{ t, id, message.getFrom()});
MessagingService.instance().sendReply(responseMessage, id, message.getFrom());
}
catch (IOException e)
{

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

@ -250,12 +250,12 @@ public class BootStrapper
public static class BootstrapTokenVerbHandler implements IVerbHandler
{
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
StorageService ss = StorageService.instance;
String tokenString = StorageService.getPartitioner().getTokenFactory().toString(ss.getBootstrapToken());
Message response = message.getInternalReply(tokenString.getBytes(Charsets.UTF_8));
MessagingService.instance().sendOneWay(response, message.getFrom());
MessagingService.instance().sendReply(response, id, message.getFrom());
}
}

2
src/java/org/apache/cassandra/gms/GossipDigestAck2VerbHandler.java
View File

@ -37,7 +37,7 @@ public class GossipDigestAck2VerbHandler implements IVerbHandler
{
private static Logger logger_ = LoggerFactory.getLogger(GossipDigestAck2VerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
InetAddress from = message.getFrom();
if (logger_.isTraceEnabled())

2
src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java
View File

@ -40,7 +40,7 @@ public class GossipDigestAckVerbHandler implements IVerbHandler
{
private static Logger logger_ = LoggerFactory.getLogger(GossipDigestAckVerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
InetAddress from = message.getFrom();
if (logger_.isTraceEnabled())

2
src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java
View File

@ -39,7 +39,7 @@ public class GossipDigestSynVerbHandler implements IVerbHandler
{
private static Logger logger_ = LoggerFactory.getLogger( GossipDigestSynVerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
InetAddress from = message.getFrom();
if (logger_.isTraceEnabled())

44
src/java/org/apache/cassandra/net/Header.java
View File

@ -33,8 +33,7 @@ import org.apache.cassandra.service.StorageService;
public class Header
{
private static ICompactSerializer<Header> serializer_;
private static AtomicInteger idGen_ = new AtomicInteger(0);
static
{
serializer_ = new HeaderSerializer();
@ -48,30 +47,22 @@ public class Header
private final InetAddress from_;
// TODO STAGE can be determined from verb
private final StorageService.Verb verb_;
private final String messageId_;
protected Map<String, byte[]> details_ = new Hashtable<String, byte[]>();
Header(String id, InetAddress from, StorageService.Verb verb)
{
assert id != null;
assert from != null;
assert verb != null;
messageId_ = id;
from_ = from;
verb_ = verb;
}
Header(String id, InetAddress from, StorageService.Verb verb, Map<String, byte[]> details)
{
this(id, from, verb);
details_ = details;
}
Header(InetAddress from, StorageService.Verb verb)
{
this(Integer.toString(idGen_.incrementAndGet()), from, verb);
}
assert from != null;
assert verb != null;
from_ = from;
verb_ = verb;
}
Header(InetAddress from, StorageService.Verb verb, Map<String, byte[]> details)
{
this(from, verb);
details_ = details;
}
InetAddress getFrom()
{
@ -83,11 +74,6 @@ public class Header
return verb_;
}
String getMessageId()
{
return messageId_;
}
byte[] getDetail(String key)
{
return details_.get(key);
@ -108,7 +94,6 @@ class HeaderSerializer implements ICompactSerializer<Header>
{
public void serialize(Header t, DataOutputStream dos) throws IOException
{
dos.writeUTF(t.getMessageId());
CompactEndpointSerializationHelper.serialize(t.getFrom(), dos);
dos.writeInt(t.getVerb().ordinal());
@ -128,7 +113,6 @@ class HeaderSerializer implements ICompactSerializer<Header>
public Header deserialize(DataInputStream dis) throws IOException
{
String id = dis.readUTF();
InetAddress from = CompactEndpointSerializationHelper.deserialize(dis);
int verbOrdinal = dis.readInt();
@ -144,7 +128,7 @@ class HeaderSerializer implements ICompactSerializer<Header>
details.put(key, bytes);
}
return new Header(id, from, StorageService.VERBS[verbOrdinal], details);
return new Header(from, StorageService.VERBS[verbOrdinal], details);
}
}

7
src/java/org/apache/cassandra/net/IVerbHandler.java
View File

@ -31,8 +31,9 @@ public interface IVerbHandler
* class was registered by a call to MessagingService.registerVerbHandlers).
* Note that the caller should not be holding any locks when calling this method
* because the implementation may be synchronized.
*
* @param message - incoming message that needs handling.
*
* @param message - incoming message that needs handling.
* @param id
*/
public void doVerb(Message message);
public void doVerb(Message message, String id);
}

6
src/java/org/apache/cassandra/net/IncomingTcpConnection.java
View File

@ -101,8 +101,10 @@ public class IncomingTcpConnection extends Thread
logger.info("Received connection from newer protocol version. Ignorning message.");
else
{
Message message = Message.serializer().deserialize(new DataInputStream(new ByteArrayInputStream(contentBytes)));
MessagingService.instance().receive(message);
DataInputStream dis = new DataInputStream(new ByteArrayInputStream(contentBytes));
String id = dis.readUTF();
Message message = Message.serializer().deserialize(dis);
MessagingService.instance().receive(message, id);
}
}
// prepare to read the next message

14
src/java/org/apache/cassandra/net/Message.java
View File

@ -94,21 +94,17 @@ public class Message
return header_.getVerb();
}
public String getMessageId()
{
return header_.getMessageId();
}
// TODO should take byte[] + length so we don't have to copy to a byte[] of exactly the right len
// TODO make static
public Message getReply(InetAddress from, byte[] args)
{
Header header = new Header(getMessageId(), from, StorageService.Verb.REQUEST_RESPONSE);
Header header = new Header(from, StorageService.Verb.REQUEST_RESPONSE);
return new Message(header, args);
}
public Message getInternalReply(byte[] body)
{
Header header = new Header(getMessageId(), FBUtilities.getLocalAddress(), StorageService.Verb.INTERNAL_RESPONSE);
Header header = new Header(FBUtilities.getLocalAddress(), StorageService.Verb.INTERNAL_RESPONSE);
return new Message(header, body);
}
@ -116,9 +112,7 @@ public class Message
{
StringBuilder sbuf = new StringBuilder("");
String separator = System.getProperty("line.separator");
sbuf.append("ID:" + getMessageId())
.append(separator)
.append("FROM:" + getFrom())
sbuf.append("FROM:" + getFrom())
.append(separator)
.append("TYPE:" + getMessageType())
.append(separator)

16
src/java/org/apache/cassandra/net/MessageDeliveryTask.java
View File

@ -28,18 +28,20 @@ public class MessageDeliveryTask implements Runnable
{
private static final Logger logger_ = LoggerFactory.getLogger(MessageDeliveryTask.class);
private Message message_;
private final long constructionTime_ = System.currentTimeMillis();
private Message message;
private final long constructionTime = System.currentTimeMillis();
private final String id;
public MessageDeliveryTask(Message message)
public MessageDeliveryTask(Message message, String id)
{
assert message != null;
message_ = message;
this.message = message;
this.id = id;
}
public void run()
{
StorageService.Verb verb = message_.getVerb();
StorageService.Verb verb = message.getVerb();
switch (verb)
{
case BINARY:
@ -48,7 +50,7 @@ public class MessageDeliveryTask implements Runnable
case RANGE_SLICE:
case READ_REPAIR:
case REQUEST_RESPONSE:
if (System.currentTimeMillis() > constructionTime_ + DatabaseDescriptor.getRpcTimeout())
if (System.currentTimeMillis() > constructionTime + DatabaseDescriptor.getRpcTimeout())
{
MessagingService.instance().incrementDroppedMessages(verb);
return;
@ -67,6 +69,6 @@ public class MessageDeliveryTask implements Runnable
IVerbHandler verbHandler = MessagingService.instance().getVerbHandler(verb);
assert verbHandler != null : "unknown verb " + verb;
verbHandler.doVerb(message_);
verbHandler.doVerb(message, id);
}
}

46
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -34,8 +34,6 @@ import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.base.Function;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -53,11 +51,10 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.FileStreamTask;
import org.apache.cassandra.streaming.StreamHeader;
import org.apache.cassandra.utils.ExpiringMap;
import org.apache.cassandra.utils.GuidGenerator;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.SimpleCondition;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import org.cliffc.high_scale_lib.NonBlockingHashSet;
public final class MessagingService implements MessagingServiceMBean
{
@ -262,6 +259,13 @@ public final class MessagingService implements MessagingServiceMBean
assert previous == null;
}
private static AtomicInteger idGen = new AtomicInteger(0);
// TODO make these integers to avoid unnecessary int -> string -> int conversions
private static String nextId()
{
return Integer.toString(idGen.incrementAndGet());
}
/**
* Send a message to a given endpoint. This method specifies a callback
* which is invoked with the actual response.
@ -272,12 +276,24 @@ public final class MessagingService implements MessagingServiceMBean
* suggest that a timeout occurred to the invoker of the send().
* @return an reference to message id used to match with the result
*/
public String sendRR(Message message, InetAddress to, IAsyncCallback cb)
public String sendRR(Message message, InetAddress to, IMessageCallback cb)
{
String messageId = message.getMessageId();
addCallback(cb, messageId, to);
sendOneWay(message, to);
return messageId;
String id = nextId();
if (logger_.isDebugEnabled())
logger_.debug("Sending " + message.getVerb() + " to " + id + "@" + to);
addCallback(cb, id, to);
sendOneWay(message, id, to);
return id;
}
public void sendOneWay(Message message, InetAddress to)
{
sendOneWay(message, nextId(), to);
}
public void sendReply(Message message, String id, InetAddress to)
{
sendOneWay(message, id, to);
}
/**
@ -286,12 +302,12 @@ public final class MessagingService implements MessagingServiceMBean
* @param message messages to be sent.
* @param to endpoint to which the message needs to be sent
*/
public void sendOneWay(Message message, InetAddress to)
private void sendOneWay(Message message, String id, InetAddress to)
{
// do local deliveries
if ( message.getFrom().equals(to) )
{
receive(message);
receive(message, id);
return;
}
@ -310,6 +326,7 @@ public final class MessagingService implements MessagingServiceMBean
try
{
DataOutputBuffer buffer = new DataOutputBuffer();
buffer.writeUTF(id);
Message.serializer().serialize(message, buffer);
data = buffer.getData();
}
@ -327,8 +344,7 @@ public final class MessagingService implements MessagingServiceMBean
public IAsyncResult sendRR(Message message, InetAddress to)
{
IAsyncResult iar = new AsyncResult();
addCallback(iar, message.getMessageId(), to);
sendOneWay(message, to);
sendRR(message, to, iar);
return iar;
}
@ -376,13 +392,13 @@ public final class MessagingService implements MessagingServiceMBean
logger_.info("Shutdown complete (no further commands will be processed)");
}
public void receive(Message message)
public void receive(Message message, String id)
{
message = SinkManager.processServerMessage(message);
if (message == null)
return;
Runnable runnable = new MessageDeliveryTask(message);
Runnable runnable = new MessageDeliveryTask(message, id);
ExecutorService stage = StageManager.getStage(message.getMessageType());
assert stage != null : "No stage for message type " + message.getMessageType();
stage.execute(runnable);

15
src/java/org/apache/cassandra/net/ResponseVerbHandler.java
View File

@ -19,8 +19,6 @@
package org.apache.cassandra.net;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -31,14 +29,13 @@ public class ResponseVerbHandler implements IVerbHandler
{
private static final Logger logger_ = LoggerFactory.getLogger( ResponseVerbHandler.class );
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
String messageId = message.getMessageId();
double age = System.currentTimeMillis() - MessagingService.instance().getRegisteredCallbackAge(messageId);
Pair<InetAddress, IMessageCallback> pair = MessagingService.instance().removeRegisteredCallback(messageId);
double age = System.currentTimeMillis() - MessagingService.instance().getRegisteredCallbackAge(id);
Pair<InetAddress, IMessageCallback> pair = MessagingService.instance().removeRegisteredCallback(id);
if (pair == null)
{
logger_.debug("Callback already removed for {}", messageId);
logger_.debug("Callback already removed for {}", id);
return;
}
@ -48,13 +45,13 @@ public class ResponseVerbHandler implements IVerbHandler
if (cb instanceof IAsyncCallback)
{
if (logger_.isDebugEnabled())
logger_.debug("Processing response on a callback from " + message.getMessageId() + "@" + message.getFrom());
logger_.debug("Processing response on a callback from " + id + "@" + message.getFrom());
((IAsyncCallback) cb).response(message);
}
else
{
if (logger_.isDebugEnabled())
logger_.debug("Processing response on an async result from " + message.getMessageId() + "@" + message.getFrom());
logger_.debug("Processing response on an async result from " + id + "@" + message.getFrom());
((IAsyncResult) cb).result(message);
}
}

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

@ -23,7 +23,6 @@ import java.net.InetAddress;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
@ -578,7 +577,7 @@ public class AntiEntropyService
/**
* Trigger a validation compaction which will return the tree upon completion.
*/
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
byte[] bytes = message.getMessageBody();
@ -645,7 +644,7 @@ public class AntiEntropyService
}
}
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
byte[] bytes = message.getMessageBody();
DataInputStream buffer = new DataInputStream(new ByteArrayInputStream(bytes));

6
src/java/org/apache/cassandra/service/IndexScanVerbHandler.java
View File

@ -33,7 +33,7 @@ public class IndexScanVerbHandler implements IVerbHandler
{
private static final Logger logger = LoggerFactory.getLogger(IndexScanVerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
try
{
@ -43,8 +43,8 @@ public class IndexScanVerbHandler implements IVerbHandler
RangeSliceReply reply = new RangeSliceReply(rows);
Message response = reply.getReply(message);
if (logger.isDebugEnabled())
logger.debug("Sending " + reply+ " to " + message.getMessageId() + "@" + message.getFrom());
MessagingService.instance().sendOneWay(response, message.getFrom());
logger.debug("Sending " + reply+ " to " + id + "@" + message.getFrom());
MessagingService.instance().sendReply(response, id, message.getFrom());
}
catch (Exception ex)
{

6
src/java/org/apache/cassandra/service/RangeSliceVerbHandler.java
View File

@ -35,7 +35,7 @@ public class RangeSliceVerbHandler implements IVerbHandler
private static final Logger logger = LoggerFactory.getLogger(RangeSliceVerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
try
{
@ -52,8 +52,8 @@ public class RangeSliceVerbHandler implements IVerbHandler
QueryFilter.getFilter(command.predicate, cfs.getComparator())));
Message response = reply.getReply(message);
if (logger.isDebugEnabled())
logger.debug("Sending " + reply+ " to " + message.getMessageId() + "@" + message.getFrom());
MessagingService.instance().sendOneWay(response, message.getFrom());
logger.debug("Sending " + reply+ " to " + id + "@" + message.getFrom());
MessagingService.instance().sendReply(response, id, message.getFrom());
}
catch (Exception ex)
{

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

@ -144,21 +144,6 @@ public class StorageLoadBalancer implements IEndpointStateChangeSubscriber
*/
}
class MoveMessageVerbHandler implements IVerbHandler
{
public void doVerb(Message message)
{
Message reply = message.getInternalReply(new byte[] {(byte)(isMoveable_.get() ? 1 : 0)});
MessagingService.instance().sendOneWay(reply, message.getFrom());
if ( isMoveable_.get() )
{
// MoveMessage moveMessage = (MoveMessage)message.getMessageBody()[0];
/* Start the leave operation and join the ring at the position specified */
isMoveable_.set(false);
}
}
}
private static final int BROADCAST_INTERVAL = 60 * 1000;
public static final StorageLoadBalancer instance = new StorageLoadBalancer();

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

@ -131,6 +131,7 @@ public class StorageProxy implements StorageProxyMBean
// Multimap that holds onto all the messages and addresses meant for a specific datacenter
Map<String, Multimap<Message, InetAddress>> dcMessages = new HashMap<String, Multimap<Message, InetAddress>>(hintedEndpoints.size());
Message unhintedMessage = null;
for (Map.Entry<InetAddress, Collection<InetAddress>> entry : hintedEndpoints.asMap().entrySet())
{
@ -149,9 +150,10 @@ public class StorageProxy implements StorageProxyMBean
else
{
// belongs on a different server
Message unhintedMessage = rm.makeRowMutationMessage();
if (unhintedMessage == null)
unhintedMessage = rm.makeRowMutationMessage();
if (logger.isDebugEnabled())
logger.debug("insert writing key " + ByteBufferUtil.bytesToHex(rm.key()) + " to " + unhintedMessage.getMessageId() + "@" + destination);
logger.debug("insert writing key " + ByteBufferUtil.bytesToHex(rm.key()) + " to " + destination);
Multimap<Message, InetAddress> messages = dcMessages.get(dc);
if (messages == null)
@ -165,7 +167,8 @@ public class StorageProxy implements StorageProxyMBean
}
else
{
// hinted
// hinted messages are unique, so there is no point to adding a hop by forwarding via another node.
// thus, we use sendRR/sendOneWay directly here.
Message hintedMessage = rm.makeRowMutationMessage();
for (InetAddress target : targets)
{
@ -173,25 +176,14 @@ public class StorageProxy implements StorageProxyMBean
{
addHintHeader(hintedMessage, target);
if (logger.isDebugEnabled())
logger.debug("insert writing key " + ByteBufferUtil.bytesToHex(rm.key()) + " to " + hintedMessage.getMessageId() + "@" + destination + " for " + target);
logger.debug("insert writing key " + ByteBufferUtil.bytesToHex(rm.key()) + " to " + destination + " for " + target);
}
}
// (non-destination hints are part of the callback and count towards consistency only under CL.ANY)
// (non-destination hints are part of the callback and count towards consistency only under CL.ANY)
// non-destination hints are part of the callback and count towards consistency only under CL.ANY
if (writeEndpoints.contains(destination) || consistency_level == ConsistencyLevel.ANY)
MessagingService.instance().sendRR(hintedMessage, destination, responseHandler);
else
MessagingService.instance().sendOneWay(hintedMessage, destination);
Multimap<Message, InetAddress> messages = dcMessages.get(dc);
if (messages == null)
{
messages = HashMultimap.create();
dcMessages.put(dc, messages);
}
messages.put(hintedMessage, destination);
}
}
@ -229,6 +221,9 @@ public class StorageProxy implements StorageProxyMBean
for (Map.Entry<Message, Collection<InetAddress>> messages: entry.getValue().asMap().entrySet())
{
Message message = messages.getKey();
// a single message object is used for unhinted writes, so clean out any forwards
// from previous loop iterations
message.removeHeader(RowMutation.FORWARD_HEADER);
if (dataCenter.equals(localDataCenter))
{
@ -375,12 +370,13 @@ public class StorageProxy implements StorageProxyMBean
{
Message message = command.makeReadMessage();
if (logger.isDebugEnabled())
logger.debug("reading data for " + command + " from " + message.getMessageId() + "@" + dataPoint);
logger.debug("reading data for " + command + " from " + dataPoint);
MessagingService.instance().sendRR(message, dataPoint, handler);
}
// We lazy-construct the digest Message object since it may not be necessary if we
// are doing a local digest read, or no digest reads at all.
Message digestMessage = null;
for (InetAddress digestPoint : endpoints.subList(1, endpoints.size()))
{
if (digestPoint.equals(FBUtilities.getLocalAddress()))
@ -391,9 +387,10 @@ public class StorageProxy implements StorageProxyMBean
}
else
{
Message digestMessage = digestCommand.makeReadMessage();
if (digestMessage == null)
digestMessage = digestCommand.makeReadMessage();
if (logger.isDebugEnabled())
logger.debug("reading digest for " + command + " from " + digestMessage.getMessageId() + "@" + digestPoint);
logger.debug("reading digest for " + command + " from " + digestPoint);
MessagingService.instance().sendRR(digestMessage, digestPoint, handler);
}
}
@ -493,11 +490,9 @@ public class StorageProxy implements StorageProxyMBean
{
ReadResponseResolver resolver = new ReadResponseResolver(command.table, command.key);
RepairCallback<Row> handler = new RepairCallback<Row>(resolver, endpoints);
Message messageRepair = command.makeReadMessage();
for (InetAddress endpoint : endpoints)
{
Message messageRepair = command.makeReadMessage();
MessagingService.instance().sendRR(messageRepair, endpoint, handler);
}
return handler;
}
@ -546,18 +541,18 @@ public class StorageProxy implements StorageProxyMBean
{
DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getLocalAddress(), liveEndpoints);
RangeSliceCommand c2 = new RangeSliceCommand(command.keyspace, command.column_family, command.super_column, command.predicate, range, command.max_keys);
Message message = c2.getMessage();
// collect replies and resolve according to consistency level
RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints);
AbstractReplicationStrategy rs = Table.open(command.keyspace).getReplicationStrategy();
ReadCallback<List<Row>> handler = getReadCallback(resolver, command.keyspace, consistency_level);
// TODO bail early if live endpoints can't satisfy requested consistency level
for (InetAddress endpoint : liveEndpoints)
for (InetAddress endpoint : liveEndpoints)
{
Message message = c2.getMessage();
MessagingService.instance().sendRR(message, endpoint, handler);
if (logger.isDebugEnabled())
logger.debug("reading " + c2 + " from " + message.getMessageId() + "@" + endpoint);
logger.debug("reading " + c2 + " from " + endpoint);
}
// TODO read repair on remaining replicas?
@ -811,12 +806,12 @@ public class StorageProxy implements StorageProxyMBean
throw new UnavailableException();
IndexScanCommand command = new IndexScanCommand(keyspace, column_family, index_clause, column_predicate, range);
Message message = command.getMessage();
for (InetAddress endpoint : liveEndpoints)
{
Message message = command.getMessage();
MessagingService.instance().sendRR(message, endpoint, handler);
if (logger.isDebugEnabled())
logger.debug("reading " + command + " from " + message.getMessageId() + "@" + endpoint);
logger.debug("reading " + command + " from " + endpoint);
}
List<Row> theseRows;
@ -899,11 +894,9 @@ public class StorageProxy implements StorageProxyMBean
// Send out the truncate calls and track the responses with the callbacks.
logger.debug("Starting to send truncate messages to hosts {}", allEndpoints);
Truncation truncation = new Truncation(keyspace, cfname);
Message message = truncation.makeTruncationMessage();
for (InetAddress endpoint : allEndpoints)
{
Message message = truncation.makeTruncationMessage();
MessagingService.instance().sendRR(message, endpoint, responseHandler);
}
// Wait for all
logger.debug("Sent all truncate messages, now waiting for {} responses", blockFor);

6
src/java/org/apache/cassandra/streaming/ReplicationFinishedVerbHandler.java
View File

@ -33,12 +33,12 @@ public class ReplicationFinishedVerbHandler implements IVerbHandler
{
private static Logger logger = LoggerFactory.getLogger(ReplicationFinishedVerbHandler.class);
public void doVerb(Message msg)
public void doVerb(Message msg, String id)
{
StorageService.instance.confirmReplication(msg.getFrom());
Message response = msg.getInternalReply(ArrayUtils.EMPTY_BYTE_ARRAY);
if (logger.isDebugEnabled())
logger.debug("Replying to " + msg.getMessageId() + "@" + msg.getFrom());
MessagingService.instance().sendOneWay(response, msg.getFrom());
logger.debug("Replying to " + id + "@" + msg.getFrom());
MessagingService.instance().sendReply(response, id, msg.getFrom());
}
}

2
src/java/org/apache/cassandra/streaming/StreamReplyVerbHandler.java
View File

@ -36,7 +36,7 @@ public class StreamReplyVerbHandler implements IVerbHandler
{
private static Logger logger = LoggerFactory.getLogger(StreamReplyVerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
byte[] body = message.getMessageBody();
ByteArrayInputStream bufIn = new ByteArrayInputStream(body);

2
src/java/org/apache/cassandra/streaming/StreamRequestVerbHandler.java
View File

@ -37,7 +37,7 @@ public class StreamRequestVerbHandler implements IVerbHandler
{
private static Logger logger = LoggerFactory.getLogger(StreamRequestVerbHandler.class);
public void doVerb(Message message)
public void doVerb(Message message, String id)
{
if (logger.isDebugEnabled())
logger.debug("Received a StreamRequestMessage from {}", message.getFrom());

BIN
test/data/serialization/0.7/db.RangeSliceCommand.bin
View File

Binary file not shown.

BIN
test/data/serialization/0.7/db.Row.bin
View File

Binary file not shown.

BIN
test/data/serialization/0.7/db.RowMutation.bin
View File

Binary file not shown.

BIN
test/data/serialization/0.7/db.SliceByNamesReadCommand.bin
View File

Binary file not shown.

BIN
test/data/serialization/0.7/db.SliceFromReadCommand.bin
View File

Binary file not shown.

BIN
test/data/serialization/0.7/db.Truncation.bin
View File

Binary file not shown.

2
test/data/serialization/0.7/db.migration.Keyspace1.bin
View File

File diff suppressed because one or more lines are too long

2
test/data/serialization/0.7/db.migration.Keyspace2.bin
View File

File diff suppressed because one or more lines are too long

BIN
test/data/serialization/0.7/db.migration.Keyspace3.bin
View File

Binary file not shown.

2
test/data/serialization/0.7/db.migration.Keyspace4.bin
View File

File diff suppressed because one or more lines are too long

2
test/data/serialization/0.7/db.migration.Keyspace5.bin
View File

File diff suppressed because one or more lines are too long

BIN
test/data/serialization/0.7/gms.EndpointState.bin
View File

Binary file not shown.

BIN
test/data/serialization/0.7/service.TreeRequest.bin
View File

Binary file not shown.

BIN
test/data/serialization/0.7/service.TreeResponse.bin
View File

Binary file not shown.

BIN
test/data/serialization/0.7/streaming.StreamReply.bin
View File

Binary file not shown.

BIN
test/data/serialization/0.7/streaming.StreamRequestMessage.bin
View File

Binary file not shown.

BIN
test/data/serialization/0.7/utils.BloomFilter.bin
View File

Binary file not shown.

BIN
test/data/serialization/0.7/utils.LegacyBloomFilter.bin
View File

Binary file not shown.