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

send back response so blocking calls can work 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@758977 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-03-27 02:19:35 +00:00
parent 8f56a96a69
commit def1edd1fe
2 changed files with 61 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -46,17 +46,13 @@ public class RowMutationVerbHandler implements IVerbHandler
protected Row row_ = new Row();
protected DataInputBuffer buffer_ = new DataInputBuffer();
}
private static Logger logger_ = Logger.getLogger(RowMutationVerbHandler.class);
private static Logger logger_ = Logger.getLogger(RowMutationVerbHandler.class);
/* We use this so that we can reuse the same row mutation context for the mutation. */
private static ThreadLocal<RowMutationContext> tls_ = new InheritableThreadLocal<RowMutationContext>();
public void doVerb(Message message)
{
/* For DEBUG only. Printing queue length */
logger_.info( "ROW MUTATION STAGE: " + StageManager.getStageTaskCount(StorageService.mutationStage_) );
/* END DEBUG */
byte[] bytes = (byte[])message.getMessageBody()[0];
/* Obtain a Row Mutation Context from TLS */
RowMutationContext rowMutationCtx = tls_.get();
@ -65,51 +61,46 @@ public class RowMutationVerbHandler implements IVerbHandler
rowMutationCtx = new RowMutationContext();
tls_.set(rowMutationCtx);
}
rowMutationCtx.buffer_.reset(bytes, bytes.length);
rowMutationCtx.buffer_.reset(bytes, bytes.length);
try
{
RowMutationMessage rmMsg = RowMutationMessage.serializer().deserialize(rowMutationCtx.buffer_);
RowMutation rm = rmMsg.getRowMutation();
RowMutation rm = RowMutation.serializer().deserialize(rowMutationCtx.buffer_);
logger_.debug("Applying " + rm);
/* Check if there were any hints in this message */
byte[] hintedBytes = message.getHeader(RowMutationMessage.hint_);
byte[] hintedBytes = message.getHeader(RowMutationMessage.hint_);
if ( hintedBytes != null && hintedBytes.length > 0 )
{
EndPoint hint = EndPoint.fromBytes(hintedBytes);
logger_.debug("Adding hint for " + hint);
/* add necessary hints to this mutation */
try
{
RowMutation hintedMutation = new RowMutation(rm.table(), HintedHandOffManager.key_);
hintedMutation.addHints(rm.key() + ":" + hint.getHost());
hintedMutation.apply();
}
catch ( ColumnFamilyNotDefinedException ex )
{
logger_.debug(LogUtil.throwableToString(ex));
}
RowMutation hintedMutation = new RowMutation(rm.table(), HintedHandOffManager.key_);
hintedMutation.addHints(rm.key() + ":" + hint.getHost());
hintedMutation.apply();
}
long start = System.currentTimeMillis();
long start = System.currentTimeMillis();
rowMutationCtx.row_.key(rm.key());
rm.apply(rowMutationCtx.row_);
long end = System.currentTimeMillis();
logger_.info("ROW MUTATION APPLY: " + (end - start) + " ms.");
/*WriteResponseMessage writeResponseMessage = new WriteResponseMessage(rm.table(), rm.key(), true);
Message response = message.getReply( StorageService.getLocalStorageEndPoint(), new Object[]{writeResponseMessage} );
logger_.debug("Sending teh response to " + message.getFrom() + " for key :" + rm.key());
MessagingService.getMessagingInstance().sendOneWay(response, message.getFrom()); */
}
catch( ColumnFamilyNotDefinedException ex )
long end = System.currentTimeMillis();
WriteResponse response = new WriteResponse(rm.table(), rm.key(), true);
Message responseMessage = WriteResponse.makeWriteResponseMessage(message, response);
logger_.debug("Mutation applied in " + (end - start) + "ms. Sending response to " + message.getFrom() + " for key :" + rm.key());
MessagingService.getMessagingInstance().sendOneWay(responseMessage, message.getFrom());
}
catch(ColumnFamilyNotDefinedException ex)
{
logger_.debug(LogUtil.throwableToString(ex));
}
catch ( IOException e )
// TODO shouldn't this be checked before it's sent to us?
logger_.warn("column family not defined, and no way to tell the client", ex);
}
catch (IOException e)
{
logger_.debug(LogUtil.throwableToString(e));
}
logger_.error("Error in row mutation", e);
}
}
}

72
src/org/apache/cassandra/db/WriteResponse.java
View File

@ -38,40 +38,26 @@ import org.apache.cassandra.service.StorageService;
* key in a table
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
*/
public class WriteResponse implements Serializable
public class WriteResponse
{
private static ICompactSerializer<WriteResponse> serializer_;
private static WriteResponseSerializer serializer_ = new WriteResponseSerializer();
static
{
serializer_ = new WriteResponseSerializer();
}
static ICompactSerializer<WriteResponse> serializer()
public static WriteResponseSerializer serializer()
{
return serializer_;
}
public static Message makeWriteResponseMessage(WriteResponse writeResponse) throws IOException
public static Message makeWriteResponseMessage(Message original, WriteResponse writeResponseMessage) throws IOException
{
ByteArrayOutputStream bos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream( bos );
WriteResponse.serializer().serialize(writeResponse, dos);
Message message = new Message(StorageService.getLocalStorageEndPoint(), MessagingService.responseStage_, MessagingService.responseVerbHandler_, new Object[]{bos.toByteArray()});
return message;
WriteResponse.serializer().serialize(writeResponseMessage, dos);
return original.getReply(StorageService.getLocalStorageEndPoint(), bos.toByteArray());
}
@XmlElement(name = "Table")
private String table_;
@XmlElement(name = "key")
private String key_;
@XmlElement(name = "Status")
private boolean status_;
private WriteResponse() {
}
private final String table_;
private final String key_;
private final boolean status_;
public WriteResponse(String table, String key, boolean bVal) {
table_ = table;
@ -79,36 +65,36 @@ private static ICompactSerializer<WriteResponse> serializer_;
status_ = bVal;
}
public String table()
public String table()
{
return table_;
}
public String key()
public String key()
{
return key_;
}
public boolean isSuccess()
{
return status_;
}
}
class WriteResponseSerializer implements ICompactSerializer<WriteResponse>
{
public void serialize(WriteResponse wm, DataOutputStream dos) throws IOException
{
dos.writeUTF(wm.table());
dos.writeUTF(wm.key());
dos.writeBoolean(wm.isSuccess());
}
public WriteResponse deserialize(DataInputStream dis) throws IOException
public static class WriteResponseSerializer implements ICompactSerializer<WriteResponse>
{
String table = dis.readUTF();
String key = dis.readUTF();
boolean status = dis.readBoolean();
return new WriteResponse(table, key, status);
public void serialize(WriteResponse wm, DataOutputStream dos) throws IOException
{
dos.writeUTF(wm.table());
dos.writeUTF(wm.key());
dos.writeBoolean(wm.isSuccess());
}
public WriteResponse deserialize(DataInputStream dis) throws IOException
{
String table = dis.readUTF();
String key = dis.readUTF();
boolean status = dis.readBoolean();
return new WriteResponse(table, key, status);
}
}
}
}