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

Merge branch 'cassandra-2.0' into trunk

This commit is contained in:
Sylvain Lebresne 2013-11-04 16:13:06 +01:00
parent a552b305f3 4a439d22c5
commit 2bbabe154e
6 changed files with 116 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -25,6 +25,7 @@
* Require Permission.SELECT for CAS updates (CASSANDRA-6247)
* New CQL-aware SSTableWriter (CASSANDRA-5894)
* Reject CAS operation when the protocol v1 is used (CASSANDRA-6270)
* Correctly throw error when frame too large (CASSANDRA-5981)
Merged from 1.2:
* Require logging in for Thrift CQL2/3 statement preparation (CASSANDRA-6254)
* restrict max_num_tokens to 1536 (CASSANDRA-6267)

6
NEWS.txt
View File

@ -15,6 +15,12 @@ using the provided 'sstableupgrade' tool.
2.0.3
=====
New features
------------
- It's now possible to configure the maximum allowed size of the native
protocol frames (native_transport_max_frame_size_in_mb in the yaml file).
Upgrading
---------
- The IEndpointStateChangeSubscriber has a new method, beforeChange, that

4
conf/cassandra.yaml
View File

@ -315,6 +315,10 @@ native_transport_port: 9042
# there is no native_transport_min_threads, idle threads will always be stopped
# after 30 seconds).
# native_transport_max_threads: 128
#
# The maximum size of allowed frame. Frame (requests) larger than this will
# be rejected as invalid. The default is 256MB.
# native_transport_max_frame_size_in_mb: 256
# Whether to start the thrift rpc server.
start_rpc: true

1
src/java/org/apache/cassandra/config/Config.java
View File

@ -96,6 +96,7 @@ public class Config
public Boolean start_native_transport = false;
public Integer native_transport_port = 9042;
public Integer native_transport_max_threads = 128;
public Integer native_transport_max_frame_size_in_mb = 256;
@Deprecated
public Integer thrift_max_message_length_in_mb = 16;

8
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -323,6 +323,9 @@ public class DatabaseDescriptor
if (conf.thrift_framed_transport_size_in_mb <= 0)
throw new ConfigurationException("thrift_framed_transport_size_in_mb must be positive");
if (conf.native_transport_max_frame_size_in_mb <= 0)
throw new ConfigurationException("native_transport_max_frame_size_in_mb must be positive");
/* end point snitch */
if (conf.endpoint_snitch == null)
{
@ -1022,6 +1025,11 @@ public class DatabaseDescriptor
return conf.native_transport_max_threads;
}
public static int getNativeTransportMaxFrameSize()
{
return conf.native_transport_max_frame_size_in_mb * 1024 * 1024;
}
public static double getCommitLogSyncBatchWindow()
{
return conf.commitlog_sync_batch_window_in_ms;

160
src/java/org/apache/cassandra/transport/Frame.java
View File

@ -28,6 +28,9 @@ import org.jboss.netty.handler.codec.oneone.OneToOneDecoder;
import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;
import org.jboss.netty.handler.codec.frame.*;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.transport.messages.ErrorMessage;
import org.apache.cassandra.utils.ByteBufferUtil;
public class Frame
@ -52,27 +55,6 @@ public class Frame
this.body = body;
}
public static Frame create(ChannelBuffer fullFrame)
{
assert fullFrame.readableBytes() >= Header.LENGTH : String.format("Frame too short (%d bytes = %s)",
fullFrame.readableBytes(),
ByteBufferUtil.bytesToHex(fullFrame.toByteBuffer()));
int version = fullFrame.readByte();
int flags = fullFrame.readByte();
int streamId = fullFrame.readByte();
int opcode = fullFrame.readByte();
int length = fullFrame.readInt();
assert length == fullFrame.readableBytes();
// version first byte is the "direction" of the frame (request or response)
Message.Direction direction = Message.Direction.extractFromVersion(version);
version = version & 0x7F;
Header header = new Header(version, flags, streamId, Message.Type.fromOpcode(opcode, direction));
return new Frame(header, fullFrame);
}
public static Frame create(Message.Type type, int streamId, int version, EnumSet<Header.Flag> flags, ChannelBuffer body)
{
Header header = new Header(version, flags, streamId, type);
@ -83,6 +65,9 @@ public class Frame
{
public static final int LENGTH = 8;
public static final int BODY_LENGTH_OFFSET = 4;
public static final int BODY_LENGTH_SIZE = 4;
public final int version;
public final EnumSet<Flag> flags;
public final int streamId;
@ -134,15 +119,19 @@ public class Frame
return new Frame(header, newBody);
}
public static class Decoder extends LengthFieldBasedFrameDecoder
public static class Decoder extends FrameDecoder
{
private static final int MAX_FRAME_LENTH = 256 * 1024 * 1024; // 256 MB
private static final int MAX_FRAME_LENGTH = DatabaseDescriptor.getNativeTransportMaxFrameSize();
private boolean discardingTooLongFrame;
private long tooLongFrameLength;
private long bytesToDiscard;
private int tooLongStreamId;
private final Connection.Factory factory;
public Decoder(Connection.Factory factory)
{
super(MAX_FRAME_LENTH, 4, 4, 0, 0, true);
this.factory = factory;
}
@ -150,53 +139,96 @@ public class Frame
protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer)
throws Exception
{
try
if (discardingTooLongFrame)
{
// We must at least validate that the frame version is something we support/know and it doesn't hurt to
// check the opcode is not garbage. And we should do that indenpently of what is the the bytes corresponding
// to the frame length are, i.e. we shouldn't wait for super.decode() to return non-null.
if (buffer.readableBytes() == 0)
return null;
int firstByte = buffer.getByte(0);
Message.Direction direction = Message.Direction.extractFromVersion(firstByte);
int version = firstByte & 0x7F;
if (version > Server.CURRENT_VERSION)
throw new ProtocolException("Invalid or unsupported protocol version: " + version);
// Validate the opcode
if (buffer.readableBytes() >= 4)
Message.Type.fromOpcode(buffer.getByte(3), direction);
ChannelBuffer frame = (ChannelBuffer) super.decode(ctx, channel, buffer);
if (frame == null)
{
return null;
}
Connection connection = (Connection)channel.getAttachment();
if (connection == null)
{
// First message seen on this channel, attach the connection object
connection = factory.newConnection(channel, version);
channel.setAttachment(connection);
}
else if (connection.getVersion() != version)
{
throw new ProtocolException(String.format("Invalid message version. Got %d but previous messages on this connection had version %d", version, connection.getVersion()));
}
return Frame.create(frame);
bytesToDiscard = discard(buffer, bytesToDiscard);
// If we have discarded everything, throw the exception
if (bytesToDiscard <= 0)
fail();
return null;
}
catch (CorruptedFrameException e)
// Wait until we have read at least the header
if (buffer.readableBytes() < Header.LENGTH)
return null;
int idx = buffer.readerIndex();
int firstByte = buffer.getByte(idx);
Message.Direction direction = Message.Direction.extractFromVersion(firstByte);
int version = firstByte & 0x7F;
if (version > Server.CURRENT_VERSION)
throw new ProtocolException("Invalid or unsupported protocol version: " + version);
int flags = buffer.getByte(idx + 1);
int streamId = buffer.getByte(idx + 2);
// This throws a protocol exceptions if the opcode is unknown
Message.Type type = Message.Type.fromOpcode(buffer.getByte(idx + 3), direction);
long bodyLength = buffer.getUnsignedInt(idx + Header.BODY_LENGTH_OFFSET);
if (bodyLength < 0)
{
throw new ProtocolException(e.getMessage());
buffer.skipBytes(Header.LENGTH);
throw new ProtocolException("Invalid frame body length: " + bodyLength);
}
catch (TooLongFrameException e)
long frameLength = bodyLength + Header.LENGTH;
if (frameLength > MAX_FRAME_LENGTH)
{
throw new ProtocolException(e.getMessage());
// Enter the discard mode and discard everything received so far.
discardingTooLongFrame = true;
tooLongStreamId = streamId;
tooLongFrameLength = frameLength;
bytesToDiscard = discard(buffer, frameLength);
if (bytesToDiscard <= 0)
fail();
return null;
}
// never overflows because it's less than the max frame length
int frameLengthInt = (int) frameLength;
if (buffer.readableBytes() < frameLengthInt)
return null;
// extract body
ChannelBuffer body = extractFrame(buffer, idx + Header.LENGTH, (int)bodyLength);
buffer.readerIndex(idx + frameLengthInt);
Connection connection = (Connection)channel.getAttachment();
if (connection == null)
{
// First message seen on this channel, attach the connection object
connection = factory.newConnection(channel, version);
channel.setAttachment(connection);
}
else if (connection.getVersion() != version)
{
throw new ProtocolException(String.format("Invalid message version. Got %d but previous messages on this connection had version %d", version, connection.getVersion()));
}
return new Frame(new Header(version, flags, streamId, type), body);
}
private void fail()
{
// Reset to the initial state and throw the exception
long tooLongFrameLength = this.tooLongFrameLength;
this.tooLongFrameLength = 0;
discardingTooLongFrame = false;
String msg = String.format("Request is too big: length %d exceeds maximum allowed length %d.", tooLongFrameLength, MAX_FRAME_LENGTH);
throw ErrorMessage.wrap(new InvalidRequestException(msg), tooLongStreamId);
}
}
// How much remains to be discarded
private static long discard(ChannelBuffer buffer, long remainingToDiscard)
{
int availableToDiscard = (int) Math.min(remainingToDiscard, buffer.readableBytes());
buffer.skipBytes(availableToDiscard);
return remainingToDiscard - availableToDiscard;
}
public static class Encoder extends OneToOneEncoder