diff --git a/CHANGES.txt b/CHANGES.txt
index 8e195225e8..6e96738268 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.0-beta4
+ * Improve checksumming and compression in protocol V5 (CASSANDRA-15299)
* Optimised repair streaming improvements (CASSANDRA-16274)
* Update jctools dependency to 3.1.0 (CASSANDRA-16255)
* 'SSLEngine closed already' exception on failed outbound connection (CASSANDRA-16277)
diff --git a/bin/cqlsh.py b/bin/cqlsh.py
index 3f6009413b..003515dba5 100644
--- a/bin/cqlsh.py
+++ b/bin/cqlsh.py
@@ -475,6 +475,8 @@ class Shell(cmd.Cmd):
kwargs = {}
if protocol_version is not None:
kwargs['protocol_version'] = protocol_version
+ else:
+ kwargs['protocol_version'] = 4
self.conn = Cluster(contact_points=(self.hostname,), port=self.port, cql_version=cqlver,
auth_provider=self.auth_provider,
ssl_options=sslhandling.ssl_settings(hostname, CONFIG_FILE) if ssl else None,
diff --git a/build.xml b/build.xml
index 7b17409ed9..436dffcb79 100644
--- a/build.xml
+++ b/build.xml
@@ -644,7 +644,7 @@
-
+
@@ -939,6 +939,7 @@
+
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index 066d22e76a..7953379222 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -699,10 +699,6 @@ native_transport_port: 9042
# you may want to adjust max_value_size_in_mb accordingly. This should be positive and less than 2048.
# native_transport_max_frame_size_in_mb: 256
-# If checksumming is enabled as a protocol option, denotes the size of the chunks into which frame
-# are bodies will be broken and checksummed.
-# native_transport_frame_block_size_in_kb: 32
-
# The maximum number of concurrent client connections.
# The default is -1, which means unlimited.
# native_transport_max_concurrent_connections: -1
diff --git a/cql_protocol_V5_framing.asc b/cql_protocol_V5_framing.asc
new file mode 100644
index 0000000000..47591a7d22
--- /dev/null
+++ b/cql_protocol_V5_framing.asc
@@ -0,0 +1,101 @@
+Framing Format Changes For Native Protocol V5
+---------------------------------------------
+
+The framing format for V5 is based on the internode messaging system as it was redesigned for 4.0 in https://issues.apache.org/jira/browse/CASSANDRA-15066[CASSANDRA-15066]. In prior versions, the relationship between `Message` and `Frame` is 1:1, each CQL message being contained in exactly one frame and every frame containing only a single message. The changes introduced here enable frames to contain multiple small messages and for oversized messages to be broken into multiple frames.
+
+The introduction of this new format, brings an unfortunate naming clash between frames from the previous format, with a 1:1 relationship to messages, and the new outer frames. Taking inspiration from SMTP (RFC-5321), what was previously referred to as a `Frame` is better described as an `Envelope` which comprises a `CQL message` along with some attendant metadata.
+
+
+
+New Frame Format
+----------------
+
+In general, a `frame` comprises a `header`, `payload` and `trailer`; this proposal introduces two specific `frame` formats, `compressed` and `uncompressed`. In both cases, the `payload` is a stream of `CQL Envelopes`, each containing a single `CQL Message`. In effect, the new framing format is a simple wrapper around the previous protocol.
+
+In all cases, a `frame` may or may not be `self contained`. If `self contained`, the `payload` includes
+one or more complete `CQL envelopes` and can be fully processed immediately. Otherwise, the `payload` contains some part of a large `CQL envelope`, which has been split into its own sequence of `outer frames`. These are expected to be transmitted/received in order, so a processor can accumulate them as they arrive and process them once all have been received.
+
+The header contains length information for the `payload`, whether or not the frame is `self contained` and a `CRC` to protect the integrity of the `header` itself. There are slight variations in the `header` format between the compressed and uncompressed variants.
+
+The payload is opaque as far as the framing format is concerned, modulo the `self contained` variation.
+
+The `trailer` contains a `CRC` to protect the integrity of the `payload`. As the `payload` includes the `CQL envelopes` including both the message and the attached metadata, such as the `stream ID`, `flags` and `opcode`, this adds a level of protection that previously wasn't available (for example from the implementation in https://issues.apache.org/jira/browse/CASSANDRA-13304[CASSANDRA-13304]).
+
+
+Uncompressed Format
+-------------------
+
+The uncompressed variant uses a 6 byte `header` containing `payload length`, `self contained flag` and `CRC24` for the `header`. The max size for the payload is `128KiB`, and is followed by its `CRC32`.
+
+....
+ 1. Payload length (17 bits)
+ 2. isSelfContained flag (1 bit)
+ 3. Header padding (6 bits)
+ 4. CRC24 of the header (24 bits)
+ 5. Payload (up to 2 ^ 17 - 1 bits)
+ 6. Payload CRC32 (32 bits)
+
+ 0 1 2 3
+ 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ | Payload Length |C| |
+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ CRC24 of Header | |
+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +
+ | |
+ + +
+ | Payload |
+ + +
+ | |
+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ | CRC32 of Payload |
+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+....
+
+LZ4 Compressed Format
+---------------------
+
+The variant with `LZ4` compression uses an `8 byte header`, containing both the compressed and uncompressed lengths of the `payload`, the `self contained flag` and a `CRC24` for the `header`. As with uncompressed frames, the max payload size is `128KiB` and is followed by a `CRC32` trailer. This is the `CRC` of the *compressed* `payload`.
+
+....
+1. Compressed length (17 bits)
+2. Uncompressed length (17 bits)
+3. {@code isSelfContained} flag (1 bit)
+4. Header padding (5 bits)
+5. CRC24 of Header contents (24 bits)
+6. Compressed Payload (up to 2 ^ 17 - 1 bits)
+7. CRC32 of Compressed Payload (32 bits)
+
+ 0 1 2 3
+ 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
++-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+| Compressed Length | Uncompressed Length
++-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ |C| | CRC24 of Header |
++-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+| |
++ +
+| Compressed Payload |
++ +
+| |
++-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+| CRC32 of Compressed Payload |
++-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+....
+
+
+Protocol Changes
+----------------
+Other than the enclosure of `CQL envelopes` in `frames`, the changes required to the protocol itself are minimal. All message exchanges follow the same `REQUEST/RESPONSE` specifications as before and no changes are required to either the `CQL envelope (frame)` or `CQL message` formats. `CQL envelope` headers contain some information which is now redundant and currently ignored. The `CQL envelope` header contains the protocol version, which was always to some extent redundant as the version is set and enforced at the connection level. It was also previously possible to enable compression at the individual `CQL envelope` level. This is no longer an option, the outer framing format being responsible for compression, which is set for the lifetime of a connection and applies to all messages transmitted throughout it. To that end, the `compression` flag on the `CQL envelope` header is ignored in V5.
+
+** TODO - note on conditional compression
+
+Protocol Negotiation
+--------------------
+
+In order to support both V5 and earlier formats, the V5 `outer framing` format is not applied to message exchanges *before* a `STARTUP` exchange is completed. This means that the initial `STARTUP` message and any `OPTIONS` messages which precede it are expected *without* the `outer framing`. Likewise, the responses returned by the server (`SUPPORTED` for `OPTIONS` and either `READY` or `AUTHENTICATE` for `STARTUP`) are transmitted without the `outer framing`.
+
+After sending the response to a `STARTUP` (`READY` or `AUTHENTICATE`), the server will begin encoding and decoding further transmissions according to the protocol version of that `STARTUP` message. Compression of the `outer frames` is dictated by the `COMPRESSION` option sent in the `STARTUP` message. Only `LZ4` compression is currently supported for V5.
+
+Note: `OPTIONS` requests may be sent by the client at any time in the connection lifecycle, both before and after the `STARTUP` exchange. As mentioned, those transmitted _before_ `STARTUP`, as well as the `SUPPORTED` responses the server returns are *not* enclosed in `outer frames`. Any `OPTIONS/SUPPORTED` exchanges _after_ the `STARTUP` exchange *are* formatted according to the protocol version. So in V5, these *will* be enclosed in `outer frames`.
+
diff --git a/lib/cassandra-driver-core-3.9.0-shaded.jar b/lib/cassandra-driver-core-3.10.0-shaded.jar
similarity index 79%
rename from lib/cassandra-driver-core-3.9.0-shaded.jar
rename to lib/cassandra-driver-core-3.10.0-shaded.jar
index 1ecabbbe81..d53e577e87 100644
Binary files a/lib/cassandra-driver-core-3.9.0-shaded.jar and b/lib/cassandra-driver-core-3.10.0-shaded.jar differ
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index fbc7f98b2a..fd91dd744b 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -188,13 +188,13 @@ public class Config
public volatile long native_transport_max_concurrent_connections_per_ip = -1L;
public boolean native_transport_flush_in_batches_legacy = false;
public volatile boolean native_transport_allow_older_protocols = true;
- public int native_transport_frame_block_size_in_kb = 32;
public volatile long native_transport_max_concurrent_requests_in_bytes_per_ip = -1L;
public volatile long native_transport_max_concurrent_requests_in_bytes = -1L;
+ public int native_transport_receive_queue_capacity_in_bytes = 1 << 20; // 1MiB
+
@Deprecated
public Integer native_transport_max_negotiable_protocol_version = null;
-
/**
* Max size of values in SSTables, in MegaBytes.
* Default is the same as the native protocol frame limit: 256Mb.
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index d559d63899..9f659fd2e8 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -514,9 +514,6 @@ public class DatabaseDescriptor
checkValidForByteConversion(conf.batch_size_warn_threshold_in_kb,
"batch_size_warn_threshold_in_kb", ByteUnit.KIBI_BYTES);
- checkValidForByteConversion(conf.native_transport_frame_block_size_in_kb,
- "native_transport_frame_block_size_in_kb", ByteUnit.KIBI_BYTES);
-
if (conf.native_transport_max_negotiable_protocol_version != null)
logger.warn("The configuration option native_transport_max_negotiable_protocol_version has been deprecated " +
"and should be removed from cassandra.yaml as it has no longer has any effect.");
@@ -2192,11 +2189,6 @@ public class DatabaseDescriptor
conf.native_transport_allow_older_protocols = isEnabled;
}
- public static int getNativeTransportFrameBlockSize()
- {
- return (int) ByteUnit.KIBI_BYTES.toBytes(conf.native_transport_frame_block_size_in_kb);
- }
-
public static double getCommitLogSyncGroupWindow()
{
return conf.commitlog_sync_group_window_in_ms;
@@ -2207,6 +2199,16 @@ public class DatabaseDescriptor
conf.commitlog_sync_group_window_in_ms = windowMillis;
}
+ public static int getNativeTransportReceiveQueueCapacityInBytes()
+ {
+ return conf.native_transport_receive_queue_capacity_in_bytes;
+ }
+
+ public static void setNativeTransportReceiveQueueCapacityInBytes(int queueSize)
+ {
+ conf.native_transport_receive_queue_capacity_in_bytes = queueSize;
+ }
+
public static long getNativeTransportMaxConcurrentRequestsInBytesPerIp()
{
return conf.native_transport_max_concurrent_requests_in_bytes_per_ip;
diff --git a/src/java/org/apache/cassandra/metrics/ClientRequestSizeMetrics.java b/src/java/org/apache/cassandra/metrics/ClientMessageSizeMetrics.java
similarity index 61%
rename from src/java/org/apache/cassandra/metrics/ClientRequestSizeMetrics.java
rename to src/java/org/apache/cassandra/metrics/ClientMessageSizeMetrics.java
index d86515e39a..acdbf2851b 100644
--- a/src/java/org/apache/cassandra/metrics/ClientRequestSizeMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/ClientMessageSizeMetrics.java
@@ -26,11 +26,11 @@ import com.codahale.metrics.Histogram;
/**
* Metrics to track the size of incoming and outgoing bytes at Cassandra server.
*/
-public class ClientRequestSizeMetrics
+public class ClientMessageSizeMetrics
{
- private static final String TYPE = "ClientRequestSize";
- public static final Counter totalBytesRead = Metrics.counter(DefaultNameFactory.createMetricName(TYPE, "IncomingBytes", null));
- public static final Counter totalBytesWritten = Metrics.counter(DefaultNameFactory.createMetricName(TYPE, "OutgoingBytes", null));
- public static final Histogram bytesReceivedPerFrame = Metrics.histogram(DefaultNameFactory.createMetricName(TYPE, "BytesRecievedPerFrame", null), true);
- public static final Histogram bytesTransmittedPerFrame = Metrics.histogram(DefaultNameFactory.createMetricName(TYPE, "BytesTransmittedPerFrame", null), true);
+ private static final String TYPE = "ClientMessageSize";
+ public static final Counter bytesReceived = Metrics.counter(DefaultNameFactory.createMetricName(TYPE, "BytesReceived", null));
+ public static final Counter bytesSent = Metrics.counter(DefaultNameFactory.createMetricName(TYPE, "BytesSent", null));
+ public static final Histogram bytesReceivedPerRequest = Metrics.histogram(DefaultNameFactory.createMetricName(TYPE, "BytesReceivedPerRequest", null), true);
+ public static final Histogram bytesSentPerResponse = Metrics.histogram(DefaultNameFactory.createMetricName(TYPE, "BytesSentPerResponse", null), true);
}
diff --git a/src/java/org/apache/cassandra/metrics/ClientMetrics.java b/src/java/org/apache/cassandra/metrics/ClientMetrics.java
index c7d880edde..d416725028 100644
--- a/src/java/org/apache/cassandra/metrics/ClientMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/ClientMetrics.java
@@ -25,9 +25,7 @@ import com.codahale.metrics.Gauge;
import com.codahale.metrics.Histogram;
import com.codahale.metrics.Meter;
import com.codahale.metrics.Reservoir;
-import org.apache.cassandra.transport.ClientStat;
-import org.apache.cassandra.transport.ConnectedClient;
-import org.apache.cassandra.transport.Server;
+import org.apache.cassandra.transport.*;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
@@ -90,9 +88,9 @@ public final class ClientMetrics
registerGauge("ConnectedNativeClientsByUser", "connectedNativeClientsByUser", this::countConnectedClientsByUser);
registerGauge("Connections", "connections", this::connectedClients);
registerGauge("ClientsByProtocolVersion", "clientsByProtocolVersion", this::recentClientStats);
- registerGauge("RequestsSize", Server.EndpointPayloadTracker::getCurrentGlobalUsage);
+ registerGauge("RequestsSize", ClientResourceLimits::getCurrentGlobalUsage);
- Reservoir ipUsageReservoir = Server.EndpointPayloadTracker.ipUsageReservoir();
+ Reservoir ipUsageReservoir = ClientResourceLimits.ipUsageReservoir();
Metrics.register(factory.createMetricName("RequestsSizeByIpDistribution"),
new Histogram(ipUsageReservoir)
{
diff --git a/src/java/org/apache/cassandra/net/AbstractMessageHandler.java b/src/java/org/apache/cassandra/net/AbstractMessageHandler.java
new file mode 100644
index 0000000000..d7097299c8
--- /dev/null
+++ b/src/java/org/apache/cassandra/net/AbstractMessageHandler.java
@@ -0,0 +1,829 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.net;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.EventLoop;
+import org.apache.cassandra.net.FrameDecoder.CorruptFrame;
+import org.apache.cassandra.net.FrameDecoder.Frame;
+import org.apache.cassandra.net.FrameDecoder.FrameProcessor;
+import org.apache.cassandra.net.FrameDecoder.IntactFrame;
+import org.apache.cassandra.net.Message.Header;
+import org.apache.cassandra.net.ResourceLimits.Limit;
+import org.apache.cassandra.utils.NoSpamLogger;
+
+import static java.lang.Math.max;
+import static java.lang.Math.min;
+import static org.apache.cassandra.net.Crc.InvalidCrc;
+import static org.apache.cassandra.utils.MonotonicClock.approxTime;
+
+/**
+ * Core logic for handling inbound message deserialization and execution (in tandem with {@link FrameDecoder}).
+ *
+ * Handles small and large messages, corruption, flow control, dispatch of message processing to a suitable
+ * consumer.
+ *
+ * # Interaction with {@link FrameDecoder}
+ *
+ * An {@link AbstractMessageHandler} implementation sits on top of a {@link FrameDecoder} in the Netty pipeline,
+ * and is tightly coupled with it.
+ *
+ * {@link FrameDecoder} decodes inbound frames and relies on a supplied {@link FrameProcessor} to act on them.
+ * {@link AbstractMessageHandler} provides two implementations of that interface:
+ * - {@link #process(Frame)} is the default, primary processor, and is expected to be implemented by subclasses
+ * - {@link UpToOneMessageFrameProcessor}, supplied to the decoder when the handler is reactivated after being
+ * put in waiting mode due to lack of acquirable reserve memory capacity permits
+ *
+ * Return value of {@link FrameProcessor#process(Frame)} determines whether the decoder should keep processing
+ * frames (if {@code true} is returned) or stop until explicitly reactivated (if {@code false} is). To reactivate
+ * the decoder (once notified of available resource permits), {@link FrameDecoder#reactivate()} is invoked.
+ *
+ * # Frames
+ *
+ * {@link AbstractMessageHandler} operates on frames of messages, and there are several kinds of them:
+ * 1. {@link IntactFrame} that are contained. As names suggest, these contain one or multiple fully contained
+ * messages believed to be uncorrupted. Guaranteed to not contain an part of an incomplete message.
+ * See {@link #processFrameOfContainedMessages(ShareableBytes, Limit, Limit)}.
+ * 2. {@link IntactFrame} that are NOT contained. These are uncorrupted parts of a large message split over multiple
+ * parts due to their size. Can represent first or subsequent frame of a large message.
+ * See {@link #processFirstFrameOfLargeMessage(IntactFrame, Limit, Limit)} and
+ * {@link #processSubsequentFrameOfLargeMessage(Frame)}.
+ * 3. {@link CorruptFrame} with corrupt header. These are unrecoverable, and force a connection to be dropped.
+ * 4. {@link CorruptFrame} with a valid header, but corrupt payload. These can be either contained or uncontained.
+ * - contained frames with corrupt payload can be gracefully dropped without dropping the connection
+ * - uncontained frames with corrupt payload can be gracefully dropped unless they represent the first
+ * frame of a new large message, as in that case we don't know how many bytes to skip
+ * See {@link #processCorruptFrame(CorruptFrame)}.
+ *
+ * Fundamental frame invariants:
+ * 1. A contained frame can only have fully-encapsulated messages - 1 to n, that don't cross frame boundaries
+ * 2. An uncontained frame can hold a part of one message only. It can NOT, say, contain end of one large message
+ * and a beginning of another one. All the bytes in an uncontained frame always belong to a single message.
+ *
+ * # Small vs large messages
+ *
+ * A single handler is equipped to process both small and large messages, potentially interleaved, but the logic
+ * differs depending on size. Small messages are deserialized in place, and then handed off to an appropriate
+ * thread pool for processing. Large messages accumulate frames until completion of a message, then hand off
+ * the untouched frames to the correct thread pool for the verb to be deserialized there and immediately processed.
+ *
+ * See {@link LargeMessage} and subclasses for concrete {@link AbstractMessageHandler} implementations for details
+ * of the large-message accumulating state-machine, and {@link ProcessMessage} and its inheritors for the differences
+ *in execution.
+ *
+ * # Flow control (backpressure)
+ *
+ * To prevent message producers from overwhelming and bringing nodes down with more inbound messages that
+ * can be processed in a timely manner, {@link AbstractMessageHandler} provides support for implementations to
+ * provide their own flow control policy.
+ *
+ * Before we attempt to process a message fully, we first infer its size from the stream. This inference is
+ * delegated to implementations as the encoding of the message size is protocol specific. Having assertained
+ * the size of the incoming message, we then attempt to acquire the corresponding number of memory permits.
+ * If we succeed, then we move on actually process the message. If we fail, the frame decoder deactivates
+ * until sufficient permits are released for the message to be processed and the handler is activated again.
+ * Permits are released back once the message has been fully processed - the definition of which is again
+ * delegated to the concrete implementations.
+ *
+ * Every connection has an exclusive number of permits allocated to it. In addition to it, there is a per-endpoint
+ * reserve capacity and a global reserve capacity {@link Limit}, shared between all connections from the same host
+ * and all connections, respectively. So long as long as the handler stays within its exclusive limit, it doesn't
+ * need to tap into reserve capacity.
+ *
+ * If tapping into reserve capacity is necessary, but the handler fails to acquire capacity from either
+ * endpoint of global reserve (and it needs to acquire from both), the handler and its frame decoder become
+ * inactive and register with a {@link WaitQueue} of the appropriate type, depending on which of the reserves
+ * couldn't be tapped into. Once enough messages have finished processing and had their permits released back
+ * to the reserves, {@link WaitQueue} will reactivate the sleeping handlers and they'll resume processing frames.
+ *
+ * The reason we 'split' reserve capacity into two limits - endpoing and global - is to guarantee liveness, and
+ * prevent single endpoint's connections from taking over the whole reserve, starving other connections.
+ *
+ * One permit per byte of serialized message gets acquired. When inflated on-heap, each message will occupy more
+ * than that, necessarily, but despite wide variance, it's a good enough proxy that correlates with on-heap footprint.
+ */
+public abstract class AbstractMessageHandler extends ChannelInboundHandlerAdapter implements FrameProcessor
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractMessageHandler.class);
+ private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.SECONDS);
+
+ protected final FrameDecoder decoder;
+
+ protected final Channel channel;
+
+ protected final int largeThreshold;
+ protected LargeMessage> largeMessage;
+
+ protected final long queueCapacity;
+ volatile long queueSize = 0L;
+ private static final AtomicLongFieldUpdater queueSizeUpdater =
+ AtomicLongFieldUpdater.newUpdater(AbstractMessageHandler.class, "queueSize");
+
+ protected final Limit endpointReserveCapacity;
+ protected final WaitQueue endpointWaitQueue;
+
+ protected final Limit globalReserveCapacity;
+ protected final WaitQueue globalWaitQueue;
+
+ protected final OnHandlerClosed onClosed;
+
+ // wait queue handle, non-null if we overrun endpoint or global capacity and request to be resumed once it's released
+ private WaitQueue.Ticket ticket = null;
+
+ protected long corruptFramesRecovered, corruptFramesUnrecovered;
+ protected long receivedCount, receivedBytes;
+ protected long throttledCount, throttledNanos;
+
+ private boolean isClosed;
+
+ public AbstractMessageHandler(FrameDecoder decoder,
+
+ Channel channel,
+ int largeThreshold,
+
+ long queueCapacity,
+ Limit endpointReserveCapacity,
+ Limit globalReserveCapacity,
+ WaitQueue endpointWaitQueue,
+ WaitQueue globalWaitQueue,
+
+ OnHandlerClosed onClosed)
+ {
+ this.decoder = decoder;
+
+ this.channel = channel;
+ this.largeThreshold = largeThreshold;
+
+ this.queueCapacity = queueCapacity;
+ this.endpointReserveCapacity = endpointReserveCapacity;
+ this.endpointWaitQueue = endpointWaitQueue;
+ this.globalReserveCapacity = globalReserveCapacity;
+ this.globalWaitQueue = globalWaitQueue;
+
+ this.onClosed = onClosed;
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg)
+ {
+ /*
+ * InboundMessageHandler works in tandem with FrameDecoder to implement flow control
+ * and work stashing optimally. We rely on FrameDecoder to invoke the provided
+ * FrameProcessor rather than on the pipeline and invocations of channelRead().
+ * process(Frame) is the primary entry point for this class.
+ */
+ throw new IllegalStateException("InboundMessageHandler doesn't expect channelRead() to be invoked");
+ }
+
+ @Override
+ public void handlerAdded(ChannelHandlerContext ctx)
+ {
+ decoder.activate(this); // the frame decoder starts inactive until explicitly activated by the added inbound message handler
+ }
+
+ @Override
+ public boolean process(Frame frame) throws IOException
+ {
+ if (frame instanceof IntactFrame)
+ return processIntactFrame((IntactFrame) frame, endpointReserveCapacity, globalReserveCapacity);
+
+ processCorruptFrame((CorruptFrame) frame);
+ return true;
+ }
+
+ private boolean processIntactFrame(IntactFrame frame, Limit endpointReserve, Limit globalReserve) throws IOException
+ {
+ if (frame.isSelfContained)
+ return processFrameOfContainedMessages(frame.contents, endpointReserve, globalReserve);
+ else if (null == largeMessage)
+ return processFirstFrameOfLargeMessage(frame, endpointReserve, globalReserve);
+ else
+ return processSubsequentFrameOfLargeMessage(frame);
+ }
+
+ /*
+ * Handle contained messages (not crossing boundaries of the frame) - both small and large, for the inbound
+ * definition of large (breaching the size threshold for what we are willing to process on event-loop vs.
+ * off event-loop).
+ */
+ private boolean processFrameOfContainedMessages(ShareableBytes bytes, Limit endpointReserve, Limit globalReserve) throws IOException
+ {
+ while (bytes.hasRemaining())
+ if (!processOneContainedMessage(bytes, endpointReserve, globalReserve))
+ return false;
+ return true;
+ }
+
+ protected abstract boolean processOneContainedMessage(ShareableBytes bytes, Limit endpointReserve, Limit globalReserve) throws IOException;
+
+
+ /*
+ * Handling of multi-frame large messages
+ */
+
+ protected abstract boolean processFirstFrameOfLargeMessage(IntactFrame frame, Limit endpointReserve, Limit globalReserve) throws IOException;
+
+ protected boolean processSubsequentFrameOfLargeMessage(Frame frame)
+ {
+ receivedBytes += frame.frameSize;
+ if (largeMessage.supply(frame))
+ {
+ receivedCount++;
+ largeMessage = null;
+ }
+ return true;
+ }
+
+ /*
+ * We can handle some corrupt frames gracefully without dropping the connection and losing all the
+ * queued up messages, but not others.
+ *
+ * Corrupt frames that *ARE NOT* safe to skip gracefully and require the connection to be dropped:
+ * - any frame with corrupt header (!frame.isRecoverable())
+ * - first corrupt-payload frame of a large message (impossible to infer message size, and without it
+ * impossible to skip the message safely
+ *
+ * Corrupt frames that *ARE* safe to skip gracefully, without reconnecting:
+ * - any self-contained frame with a corrupt payload (but not header): we lose all the messages in the
+ * frame, but that has no effect on subsequent ones
+ * - any non-first payload-corrupt frame of a large message: we know the size of the large message in
+ * flight, so we just skip frames until we've seen all its bytes; we only lose the large message
+ */
+ protected abstract void processCorruptFrame(CorruptFrame frame) throws InvalidCrc;
+
+ private void onEndpointReserveCapacityRegained(Limit endpointReserve, long elapsedNanos)
+ {
+ onReserveCapacityRegained(endpointReserve, globalReserveCapacity, elapsedNanos);
+ }
+
+ private void onGlobalReserveCapacityRegained(Limit globalReserve, long elapsedNanos)
+ {
+ onReserveCapacityRegained(endpointReserveCapacity, globalReserve, elapsedNanos);
+ }
+
+ private void onReserveCapacityRegained(Limit endpointReserve, Limit globalReserve, long elapsedNanos)
+ {
+ if (isClosed)
+ return;
+
+ assert channel.eventLoop().inEventLoop();
+
+ ticket = null;
+ throttledNanos += elapsedNanos;
+
+ try
+ {
+ /*
+ * Process up to one message using supplied overriden reserves - one of them pre-allocated,
+ * and guaranteed to be enough for one message - then, if no obstacles enountered, reactivate
+ * the frame decoder using normal reserve capacities.
+ */
+ if (processUpToOneMessage(endpointReserve, globalReserve))
+ decoder.reactivate();
+ }
+ catch (Throwable t)
+ {
+ fatalExceptionCaught(t);
+ }
+ }
+
+ protected abstract void fatalExceptionCaught(Throwable t);
+
+ // return true if the handler should be reactivated - if no new hurdles were encountered,
+ // like running out of the other kind of reserve capacity
+ private boolean processUpToOneMessage(Limit endpointReserve, Limit globalReserve) throws IOException
+ {
+ UpToOneMessageFrameProcessor processor = new UpToOneMessageFrameProcessor(endpointReserve, globalReserve);
+ decoder.processBacklog(processor);
+ return processor.isActive;
+ }
+
+ /*
+ * Process at most one message. Won't always be an entire one (if the message in the head of line
+ * is a large one, and there aren't sufficient frames to decode it entirely), but will never be more than one.
+ */
+ private class UpToOneMessageFrameProcessor implements FrameProcessor
+ {
+ private final Limit endpointReserve;
+ private final Limit globalReserve;
+
+ boolean isActive = true;
+ boolean firstFrame = true;
+
+ private UpToOneMessageFrameProcessor(Limit endpointReserve, Limit globalReserve)
+ {
+ this.endpointReserve = endpointReserve;
+ this.globalReserve = globalReserve;
+ }
+
+ @Override
+ public boolean process(Frame frame) throws IOException
+ {
+ if (firstFrame)
+ {
+ if (!(frame instanceof IntactFrame))
+ throw new IllegalStateException("First backlog frame must be intact");
+ firstFrame = false;
+ return processFirstFrame((IntactFrame) frame);
+ }
+
+ return processSubsequentFrame(frame);
+ }
+
+ private boolean processFirstFrame(IntactFrame frame) throws IOException
+ {
+ if (frame.isSelfContained)
+ {
+ isActive = processOneContainedMessage(frame.contents, endpointReserve, globalReserve);
+ return false; // stop after one message
+ }
+ else
+ {
+ isActive = processFirstFrameOfLargeMessage(frame, endpointReserve, globalReserve);
+ return isActive; // continue unless fallen behind coprocessor or ran out of reserve capacity again
+ }
+ }
+
+ private boolean processSubsequentFrame(Frame frame) throws IOException
+ {
+ if (frame instanceof IntactFrame)
+ processSubsequentFrameOfLargeMessage(frame);
+ else
+ processCorruptFrame((CorruptFrame) frame);
+
+ return largeMessage != null; // continue until done with the large message
+ }
+ }
+
+ /**
+ * Try to acquire permits for the inbound message. In case of failure, register with the right wait queue to be
+ * reactivated once permit capacity is regained.
+ */
+ @SuppressWarnings("BooleanMethodIsAlwaysInverted")
+ protected boolean acquireCapacity(Limit endpointReserve, Limit globalReserve, int bytes, long currentTimeNanos, long expiresAtNanos)
+ {
+ ResourceLimits.Outcome outcome = acquireCapacity(endpointReserve, globalReserve, bytes);
+
+ if (outcome == ResourceLimits.Outcome.INSUFFICIENT_ENDPOINT)
+ ticket = endpointWaitQueue.register(this, bytes, currentTimeNanos, expiresAtNanos);
+ else if (outcome == ResourceLimits.Outcome.INSUFFICIENT_GLOBAL)
+ ticket = globalWaitQueue.register(this, bytes, currentTimeNanos, expiresAtNanos);
+
+ if (outcome != ResourceLimits.Outcome.SUCCESS)
+ throttledCount++;
+
+ return outcome == ResourceLimits.Outcome.SUCCESS;
+ }
+
+ protected ResourceLimits.Outcome acquireCapacity(Limit endpointReserve, Limit globalReserve, int bytes)
+ {
+ long currentQueueSize = queueSize;
+
+ /*
+ * acquireCapacity() is only ever called on the event loop, and as such queueSize is only ever increased
+ * on the event loop. If there is enough capacity, we can safely addAndGet() and immediately return.
+ */
+ if (currentQueueSize + bytes <= queueCapacity)
+ {
+ queueSizeUpdater.addAndGet(this, bytes);
+ return ResourceLimits.Outcome.SUCCESS;
+ }
+
+ // we know we don't have enough local queue capacity for the entire message, so we need to borrow some from reserve capacity
+ long allocatedExcess = min(currentQueueSize + bytes - queueCapacity, bytes);
+
+ if (!globalReserve.tryAllocate(allocatedExcess))
+ return ResourceLimits.Outcome.INSUFFICIENT_GLOBAL;
+
+ if (!endpointReserve.tryAllocate(allocatedExcess))
+ {
+ globalReserve.release(allocatedExcess);
+ globalWaitQueue.signal();
+ return ResourceLimits.Outcome.INSUFFICIENT_ENDPOINT;
+ }
+
+ long newQueueSize = queueSizeUpdater.addAndGet(this, bytes);
+ long actualExcess = max(0, min(newQueueSize - queueCapacity, bytes));
+
+ /*
+ * It's possible that some permits were released at some point after we loaded current queueSize,
+ * and we can satisfy more of the permits using our exclusive per-connection capacity, needing
+ * less than previously estimated from the reserves. If that's the case, release the now unneeded
+ * permit excess back to endpoint/global reserves.
+ */
+ if (actualExcess != allocatedExcess) // actualExcess < allocatedExcess
+ {
+ long excess = allocatedExcess - actualExcess;
+
+ endpointReserve.release(excess);
+ globalReserve.release(excess);
+
+ endpointWaitQueue.signal();
+ globalWaitQueue.signal();
+ }
+
+ return ResourceLimits.Outcome.SUCCESS;
+ }
+
+ public void releaseCapacity(int bytes)
+ {
+ long oldQueueSize = queueSizeUpdater.getAndAdd(this, -bytes);
+ if (oldQueueSize > queueCapacity)
+ {
+ long excess = min(oldQueueSize - queueCapacity, bytes);
+
+ endpointReserveCapacity.release(excess);
+ globalReserveCapacity.release(excess);
+
+ endpointWaitQueue.signal();
+ globalWaitQueue.signal();
+ }
+ }
+
+ /**
+ * Invoked to release capacity for a message that has been fully, successfully processed.
+ *
+ * Normally no different from invoking {@link #releaseCapacity(int)}, but is necessary for the verifier
+ * to be able to delay capacity release for backpressure testing.
+ */
+ @VisibleForTesting
+ protected void releaseProcessedCapacity(int size, Header header)
+ {
+ releaseCapacity(size);
+ }
+
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx)
+ {
+ isClosed = true;
+
+ if (null != largeMessage)
+ largeMessage.abort();
+
+ if (null != ticket)
+ ticket.invalidate();
+
+ onClosed.call(this);
+ }
+
+ private EventLoop eventLoop()
+ {
+ return channel.eventLoop();
+ }
+
+ protected abstract String id();
+
+ /*
+ * A large-message frame-accumulating state machine.
+ *
+ * Collects intact frames until it's has all the bytes necessary to deserialize the large message,
+ * at which point it schedules a task on the appropriate {@link Stage},
+ * a task that deserializes the message and immediately invokes the verb handler.
+ *
+ * Also handles corrupt frames and potential expiry of the large message during accumulation:
+ * if it's taking the frames too long to arrive, there is no point in holding on to the
+ * accumulated frames, or in gathering more - so we release the ones we already have, and
+ * skip any remaining ones, alongside with returning memory permits early.
+ */
+ protected abstract class LargeMessage
+ {
+ protected final int size;
+ protected final H header;
+
+ protected final List buffers = new ArrayList<>();
+ protected int received;
+
+ protected final long expiresAtNanos;
+
+ protected boolean isExpired;
+ protected boolean isCorrupt;
+
+ protected LargeMessage(int size, H header, long expiresAtNanos, boolean isExpired)
+ {
+ this.size = size;
+ this.header = header;
+ this.expiresAtNanos = expiresAtNanos;
+ this.isExpired = isExpired;
+ }
+
+ protected LargeMessage(int size, H header, long expiresAtNanos, ShareableBytes bytes)
+ {
+ this(size, header, expiresAtNanos, false);
+ buffers.add(bytes);
+ }
+
+ /**
+ * Return true if this was the last frame of the large message.
+ */
+ public boolean supply(Frame frame)
+ {
+ if (frame instanceof IntactFrame)
+ onIntactFrame((IntactFrame) frame);
+ else
+ onCorruptFrame();
+
+ received += frame.frameSize;
+ if (size == received)
+ onComplete();
+ return size == received;
+ }
+
+ private void onIntactFrame(IntactFrame frame)
+ {
+ boolean expires = approxTime.isAfter(expiresAtNanos);
+ if (!isExpired && !isCorrupt)
+ {
+ if (!expires)
+ {
+ buffers.add(frame.contents.sliceAndConsume(frame.frameSize).share());
+ return;
+ }
+ releaseBuffersAndCapacity(); // release resources once we transition from normal state to expired
+ }
+ frame.consume();
+ isExpired |= expires;
+ }
+
+ private void onCorruptFrame()
+ {
+ if (!isExpired && !isCorrupt)
+ releaseBuffersAndCapacity(); // release resources once we transition from normal state to corrupt
+ isCorrupt = true;
+ isExpired |= approxTime.isAfter(expiresAtNanos);
+ }
+
+ protected abstract void onComplete();
+
+ protected abstract void abort();
+
+ protected void releaseBuffers()
+ {
+ buffers.forEach(ShareableBytes::release); buffers.clear();
+ }
+
+ protected void releaseBuffersAndCapacity()
+ {
+ releaseBuffers(); releaseCapacity(size);
+ }
+ }
+
+ /**
+ * A special-purpose wait queue to park inbound message handlers that failed to allocate
+ * reserve capacity for a message in. Upon such failure a handler registers itself with
+ * a {@link WaitQueue} of the appropriate kind (either ENDPOINT or GLOBAL - if failed
+ * to allocate endpoint or global reserve capacity, respectively), stops processing any
+ * accumulated frames or receiving new ones, and waits - until reactivated.
+ *
+ * Every time permits are returned to an endpoint or global {@link Limit}, the respective
+ * queue gets signalled, and if there are any handlers registered in it, we will attempt
+ * to reactivate as many waiting handlers as current available reserve capacity allows
+ * us to - immediately, on the {@link #signal()}-calling thread. At most one such attempt
+ * will be in progress at any given time.
+ *
+ * Handlers that can be reactivated will be grouped by their {@link EventLoop} and a single
+ * {@link ReactivateHandlers} task will be scheduled per event loop, on the corresponding
+ * event loops.
+ *
+ * When run, the {@link ReactivateHandlers} task will ask each handler in its group to first
+ * process one message - using preallocated reserve capacity - and if no obstacles were met -
+ * reactivate the handlers, this time using their regular reserves.
+ *
+ * See {@link WaitQueue#schedule()}, {@link ReactivateHandlers#run()}, {@link Ticket#reactivateHandler(Limit)}.
+ */
+ public static final class WaitQueue
+ {
+ enum Kind { ENDPOINT, GLOBAL }
+
+ private static final int NOT_RUNNING = 0;
+ @SuppressWarnings("unused")
+ private static final int RUNNING = 1;
+ private static final int RUN_AGAIN = 2;
+
+ private volatile int scheduled;
+ private static final AtomicIntegerFieldUpdater scheduledUpdater =
+ AtomicIntegerFieldUpdater.newUpdater(WaitQueue.class, "scheduled");
+
+ private final Kind kind;
+ private final Limit reserveCapacity;
+
+ private final ManyToOneConcurrentLinkedQueue queue = new ManyToOneConcurrentLinkedQueue<>();
+
+ private WaitQueue(Kind kind, Limit reserveCapacity)
+ {
+ this.kind = kind;
+ this.reserveCapacity = reserveCapacity;
+ }
+
+ public static WaitQueue endpoint(Limit endpointReserveCapacity)
+ {
+ return new WaitQueue(Kind.ENDPOINT, endpointReserveCapacity);
+ }
+
+ public static WaitQueue global(Limit globalReserveCapacity)
+ {
+ return new WaitQueue(Kind.GLOBAL, globalReserveCapacity);
+ }
+
+ private Ticket register(AbstractMessageHandler handler, int bytesRequested, long registeredAtNanos, long expiresAtNanos)
+ {
+ Ticket ticket = new Ticket(this, handler, bytesRequested, registeredAtNanos, expiresAtNanos);
+ Ticket previous = queue.relaxedPeekLastAndOffer(ticket);
+ if (null == previous || !previous.isWaiting())
+ signal(); // only signal the queue if this handler is first to register
+ return ticket;
+ }
+
+ @VisibleForTesting
+ public void signal()
+ {
+ if (queue.relaxedIsEmpty())
+ return; // we can return early if no handlers have registered with the wait queue
+
+ if (NOT_RUNNING == scheduledUpdater.getAndUpdate(this, i -> min(RUN_AGAIN, i + 1)))
+ {
+ do
+ {
+ schedule();
+ }
+ while (RUN_AGAIN == scheduledUpdater.getAndDecrement(this));
+ }
+ }
+
+ private void schedule()
+ {
+ Map tasks = null;
+
+ long currentTimeNanos = approxTime.now();
+
+ Ticket t;
+ while ((t = queue.peek()) != null)
+ {
+ if (!t.call()) // invalidated
+ {
+ queue.remove();
+ continue;
+ }
+
+ boolean isLive = t.isLive(currentTimeNanos);
+ if (isLive && !reserveCapacity.tryAllocate(t.bytesRequested))
+ {
+ if (!t.reset()) // the ticket was invalidated after being called but before now
+ {
+ queue.remove();
+ continue;
+ }
+ break; // TODO: traverse the entire queue to unblock handlers that have expired or invalidated tickets
+ }
+
+ if (null == tasks)
+ tasks = new IdentityHashMap<>();
+
+ queue.remove();
+ tasks.computeIfAbsent(t.handler.eventLoop(), e -> new ReactivateHandlers()).add(t, isLive);
+ }
+
+ if (null != tasks)
+ tasks.forEach(EventLoop::execute);
+ }
+
+ private class ReactivateHandlers implements Runnable
+ {
+ List tickets = new ArrayList<>();
+ long capacity = 0L;
+
+ private void add(Ticket ticket, boolean isLive)
+ {
+ tickets.add(ticket);
+ if (isLive) capacity += ticket.bytesRequested;
+ }
+
+ public void run()
+ {
+ Limit limit = new ResourceLimits.Basic(capacity);
+ try
+ {
+ for (Ticket ticket : tickets)
+ ticket.reactivateHandler(limit);
+ }
+ finally
+ {
+ /*
+ * Free up any unused capacity, if any. Will be non-zero if one or more handlers were closed
+ * when we attempted to run their callback, or used more of their other reserve; or if the first
+ * message in the unprocessed stream has expired in the narrow time window.
+ */
+ long remaining = limit.remaining();
+ if (remaining > 0)
+ {
+ reserveCapacity.release(remaining);
+ signal();
+ }
+ }
+ }
+ }
+
+ private static final class Ticket
+ {
+ private static final int WAITING = 0;
+ private static final int CALLED = 1;
+ private static final int INVALIDATED = 2; // invalidated by a handler that got closed
+
+ private volatile int state;
+ private static final AtomicIntegerFieldUpdater stateUpdater =
+ AtomicIntegerFieldUpdater.newUpdater(Ticket.class, "state");
+
+ private final WaitQueue waitQueue;
+ private final AbstractMessageHandler handler;
+ private final int bytesRequested;
+ private final long reigsteredAtNanos;
+ private final long expiresAtNanos;
+
+ private Ticket(WaitQueue waitQueue, AbstractMessageHandler handler, int bytesRequested, long registeredAtNanos, long expiresAtNanos)
+ {
+ this.waitQueue = waitQueue;
+ this.handler = handler;
+ this.bytesRequested = bytesRequested;
+ this.reigsteredAtNanos = registeredAtNanos;
+ this.expiresAtNanos = expiresAtNanos;
+ }
+
+ private void reactivateHandler(Limit capacity)
+ {
+ long elapsedNanos = approxTime.now() - reigsteredAtNanos;
+ try
+ {
+ if (waitQueue.kind == Kind.ENDPOINT)
+ handler.onEndpointReserveCapacityRegained(capacity, elapsedNanos);
+ else
+ handler.onGlobalReserveCapacityRegained(capacity, elapsedNanos);
+ }
+ catch (Throwable t)
+ {
+ logger.error("{} exception caught while reactivating a handler", handler.id(), t);
+ }
+ }
+
+ private boolean isWaiting()
+ {
+ return state == WAITING;
+ }
+
+ private boolean isLive(long currentTimeNanos)
+ {
+ return !approxTime.isAfter(currentTimeNanos, expiresAtNanos);
+ }
+
+ private void invalidate()
+ {
+ state = INVALIDATED;
+ waitQueue.signal();
+ }
+
+ private boolean call()
+ {
+ return stateUpdater.compareAndSet(this, WAITING, CALLED);
+ }
+
+ private boolean reset()
+ {
+ return stateUpdater.compareAndSet(this, CALLED, WAITING);
+ }
+ }
+ }
+
+ public interface OnHandlerClosed
+ {
+ void call(AbstractMessageHandler handler);
+ }
+}
diff --git a/src/java/org/apache/cassandra/net/Crc.java b/src/java/org/apache/cassandra/net/Crc.java
index dbd26014d5..9cd6edd94f 100644
--- a/src/java/org/apache/cassandra/net/Crc.java
+++ b/src/java/org/apache/cassandra/net/Crc.java
@@ -24,7 +24,7 @@ import java.util.zip.CRC32;
import io.netty.buffer.ByteBuf;
import io.netty.util.concurrent.FastThreadLocal;
-class Crc
+public class Crc
{
private static final FastThreadLocal crc32 = new FastThreadLocal()
{
@@ -37,9 +37,9 @@ class Crc
private static final byte[] initialBytes = new byte[] { (byte) 0xFA, (byte) 0x2D, (byte) 0x55, (byte) 0xCA };
- static final class InvalidCrc extends IOException
+ public static final class InvalidCrc extends IOException
{
- InvalidCrc(int read, int computed)
+ public InvalidCrc(int read, int computed)
{
super(String.format("Read %d, Computed %d", read, computed));
}
diff --git a/src/java/org/apache/cassandra/net/FrameDecoder.java b/src/java/org/apache/cassandra/net/FrameDecoder.java
index ed96adda32..64e30ef521 100644
--- a/src/java/org/apache/cassandra/net/FrameDecoder.java
+++ b/src/java/org/apache/cassandra/net/FrameDecoder.java
@@ -56,7 +56,7 @@ import static org.apache.cassandra.utils.ByteBufferUtil.copyBytes;
* 5. {@link FrameDecoderLegacyLZ4}
* LZ4 compression using standard LZ4 frame format; groups legacy messages (< 4.0) into frames
*/
-abstract class FrameDecoder extends ChannelInboundHandlerAdapter
+public abstract class FrameDecoder extends ChannelInboundHandlerAdapter
{
private static final FrameProcessor NO_PROCESSOR =
frame -> { throw new IllegalStateException("Frame processor invoked on an unregistered FrameDecoder"); };
@@ -64,7 +64,7 @@ abstract class FrameDecoder extends ChannelInboundHandlerAdapter
private static final FrameProcessor CLOSED_PROCESSOR =
frame -> { throw new IllegalStateException("Frame processor invoked on a closed FrameDecoder"); };
- interface FrameProcessor
+ public interface FrameProcessor
{
/**
* Frame processor that the frames should be handed off to.
@@ -75,10 +75,10 @@ abstract class FrameDecoder extends ChannelInboundHandlerAdapter
boolean process(Frame frame) throws IOException;
}
- abstract static class Frame
+ public abstract static class Frame
{
- final boolean isSelfContained;
- final int frameSize;
+ public final boolean isSelfContained;
+ public final int frameSize;
Frame(boolean isSelfContained, int frameSize)
{
@@ -99,9 +99,9 @@ abstract class FrameDecoder extends ChannelInboundHandlerAdapter
* {@link Message} is contained in the payload; it can be relied upon that this partial {@link Message}
* will only be delivered in its own unique {@link Frame}.
*/
- final static class IntactFrame extends Frame
+ public final static class IntactFrame extends Frame
{
- final ShareableBytes contents;
+ public final ShareableBytes contents;
IntactFrame(boolean isSelfContained, ShareableBytes contents)
{
@@ -119,7 +119,7 @@ abstract class FrameDecoder extends ChannelInboundHandlerAdapter
return !contents.hasRemaining();
}
- void consume()
+ public void consume()
{
contents.consume();
}
@@ -136,9 +136,10 @@ abstract class FrameDecoder extends ChannelInboundHandlerAdapter
* A recoverable {@link CorruptFrame} can be considered unrecoverable by {@link InboundMessageHandler}
* if it's the first frame of a large message (isn't self contained).
*/
- final static class CorruptFrame extends Frame
+ public final static class CorruptFrame extends Frame
{
- final int readCRC, computedCRC;
+ public final int readCRC;
+ public final int computedCRC;
CorruptFrame(boolean isSelfContained, int frameSize, int readCRC, int computedCRC)
{
@@ -157,7 +158,7 @@ abstract class FrameDecoder extends ChannelInboundHandlerAdapter
return new CorruptFrame(false, Integer.MIN_VALUE, readCRC, computedCRC);
}
- boolean isRecoverable()
+ public boolean isRecoverable()
{
return frameSize != Integer.MIN_VALUE;
}
@@ -192,7 +193,7 @@ abstract class FrameDecoder extends ChannelInboundHandlerAdapter
/**
* For use by InboundMessageHandler (or other upstream handlers) that want to start receiving frames.
*/
- void activate(FrameProcessor processor)
+ public void activate(FrameProcessor processor)
{
if (this.processor != NO_PROCESSOR)
throw new IllegalStateException("Attempted to activate an already active FrameDecoder");
@@ -235,7 +236,7 @@ abstract class FrameDecoder extends ChannelInboundHandlerAdapter
* For use by InboundMessageHandler (or other upstream handlers) that want to permanently
* stop receiving frames, e.g. because of an exception caught.
*/
- void discard()
+ public void discard()
{
isActive = false;
processor = CLOSED_PROCESSOR;
diff --git a/src/java/org/apache/cassandra/net/FrameDecoderCrc.java b/src/java/org/apache/cassandra/net/FrameDecoderCrc.java
index 7cd52ac7df..238a890843 100644
--- a/src/java/org/apache/cassandra/net/FrameDecoderCrc.java
+++ b/src/java/org/apache/cassandra/net/FrameDecoderCrc.java
@@ -54,9 +54,9 @@ import static org.apache.cassandra.net.Crc.updateCrc32;
* | CRC32 of Payload |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
*/
-final class FrameDecoderCrc extends FrameDecoderWith8bHeader
+public final class FrameDecoderCrc extends FrameDecoderWith8bHeader
{
- private FrameDecoderCrc(BufferPoolAllocator allocator)
+ public FrameDecoderCrc(BufferPoolAllocator allocator)
{
super(allocator);
}
diff --git a/src/java/org/apache/cassandra/net/FrameDecoderLZ4.java b/src/java/org/apache/cassandra/net/FrameDecoderLZ4.java
index 2b32d18344..9cc100586e 100644
--- a/src/java/org/apache/cassandra/net/FrameDecoderLZ4.java
+++ b/src/java/org/apache/cassandra/net/FrameDecoderLZ4.java
@@ -57,7 +57,7 @@ import static org.apache.cassandra.net.Crc.*;
* | CRC32 of Compressed Payload |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
*/
-final class FrameDecoderLZ4 extends FrameDecoderWith8bHeader
+public final class FrameDecoderLZ4 extends FrameDecoderWith8bHeader
{
public static FrameDecoderLZ4 fast(BufferPoolAllocator allocator)
{
diff --git a/src/java/org/apache/cassandra/net/FrameEncoder.java b/src/java/org/apache/cassandra/net/FrameEncoder.java
index 5f2dc37e34..fc4f861909 100644
--- a/src/java/org/apache/cassandra/net/FrameEncoder.java
+++ b/src/java/org/apache/cassandra/net/FrameEncoder.java
@@ -27,7 +27,7 @@ import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.utils.memory.BufferPool;
import org.apache.cassandra.utils.memory.BufferPools;
-abstract class FrameEncoder extends ChannelOutboundHandlerAdapter
+public abstract class FrameEncoder extends ChannelOutboundHandlerAdapter
{
protected static final BufferPool bufferPool = BufferPools.forNetworking();
@@ -36,12 +36,13 @@ abstract class FrameEncoder extends ChannelOutboundHandlerAdapter
* of the {@code FrameEncoder} without knowledge of the encoder's frame layout, while ensuring
* enough space to write the remainder of the frame's contents is reserved.
*/
- static class Payload
+ public static class Payload
{
+ public static final int MAX_SIZE = 1 << 17;
// isSelfContained is a flag in the Frame API, indicating if the contents consists of only complete messages
private boolean isSelfContained;
// the buffer to write to
- final ByteBuffer buffer;
+ public final ByteBuffer buffer;
// the number of header bytes to reserve
final int headerLength;
// the number of trailer bytes to reserve
@@ -71,13 +72,6 @@ abstract class FrameEncoder extends ChannelOutboundHandlerAdapter
this.isSelfContained = isSelfContained;
}
- // do not invoke after finish()
- boolean isEmpty()
- {
- assert !isFinished;
- return buffer.position() == headerLength;
- }
-
// do not invoke after finish()
int length()
{
@@ -86,7 +80,7 @@ abstract class FrameEncoder extends ChannelOutboundHandlerAdapter
}
// do not invoke after finish()
- int remaining()
+ public int remaining()
{
assert !isFinished;
return buffer.remaining();
@@ -100,7 +94,7 @@ abstract class FrameEncoder extends ChannelOutboundHandlerAdapter
}
// may not be written to or queried, after this is invoked; must be passed straight to an encoder (or release called)
- void finish()
+ public void finish()
{
assert !isFinished;
isFinished = true;
@@ -109,19 +103,19 @@ abstract class FrameEncoder extends ChannelOutboundHandlerAdapter
bufferPool.putUnusedPortion(buffer);
}
- void release()
+ public void release()
{
bufferPool.put(buffer);
}
}
- interface PayloadAllocator
+ public interface PayloadAllocator
{
public static final PayloadAllocator simple = Payload::new;
Payload allocate(boolean isSelfContained, int capacity);
}
- PayloadAllocator allocator()
+ public PayloadAllocator allocator()
{
return PayloadAllocator.simple;
}
diff --git a/src/java/org/apache/cassandra/net/FrameEncoderCrc.java b/src/java/org/apache/cassandra/net/FrameEncoderCrc.java
index 5049f29267..1d16868a9c 100644
--- a/src/java/org/apache/cassandra/net/FrameEncoderCrc.java
+++ b/src/java/org/apache/cassandra/net/FrameEncoderCrc.java
@@ -31,17 +31,17 @@ import static org.apache.cassandra.net.Crc.*;
* Please see {@link FrameDecoderCrc} for description of the framing produced by this encoder.
*/
@ChannelHandler.Sharable
-class FrameEncoderCrc extends FrameEncoder
+public class FrameEncoderCrc extends FrameEncoder
{
static final int HEADER_LENGTH = 6;
private static final int TRAILER_LENGTH = 4;
- static final int HEADER_AND_TRAILER_LENGTH = 10;
+ public static final int HEADER_AND_TRAILER_LENGTH = 10;
- static final FrameEncoderCrc instance = new FrameEncoderCrc();
+ public static final FrameEncoderCrc instance = new FrameEncoderCrc();
static final PayloadAllocator allocator = (isSelfContained, capacity) ->
new Payload(isSelfContained, capacity, HEADER_LENGTH, TRAILER_LENGTH);
- PayloadAllocator allocator()
+ public PayloadAllocator allocator()
{
return allocator;
}
@@ -52,7 +52,6 @@ class FrameEncoderCrc extends FrameEncoder
if (isSelfContained)
header3b |= 1 << 17;
int crc = crc24(header3b, 3);
-
put3b(frame, 0, header3b);
put3b(frame, 3, crc);
}
@@ -87,6 +86,7 @@ class FrameEncoderCrc extends FrameEncoder
frame.limit(frameLength);
frame.putInt(frameLength - TRAILER_LENGTH, frameCrc);
frame.position(0);
+
return GlobalBufferPoolAllocator.wrap(frame);
}
catch (Throwable t)
diff --git a/src/java/org/apache/cassandra/net/FrameEncoderLZ4.java b/src/java/org/apache/cassandra/net/FrameEncoderLZ4.java
index 2d76170b02..75f15c726b 100644
--- a/src/java/org/apache/cassandra/net/FrameEncoderLZ4.java
+++ b/src/java/org/apache/cassandra/net/FrameEncoderLZ4.java
@@ -34,9 +34,10 @@ import static org.apache.cassandra.net.Crc.*;
* Please see {@link FrameDecoderLZ4} for description of the framing produced by this encoder.
*/
@ChannelHandler.Sharable
+public
class FrameEncoderLZ4 extends FrameEncoder
{
- static final FrameEncoderLZ4 fastInstance = new FrameEncoderLZ4(LZ4Factory.fastestInstance().fastCompressor());
+ public static final FrameEncoderLZ4 fastInstance = new FrameEncoderLZ4(LZ4Factory.fastestInstance().fastCompressor());
private final LZ4Compressor compressor;
@@ -46,7 +47,7 @@ class FrameEncoderLZ4 extends FrameEncoder
}
private static final int HEADER_LENGTH = 8;
- static final int HEADER_AND_TRAILER_LENGTH = 12;
+ public static final int HEADER_AND_TRAILER_LENGTH = 12;
private static void writeHeader(ByteBuffer frame, boolean isSelfContained, long compressedLength, long uncompressedLength)
{
diff --git a/src/java/org/apache/cassandra/net/FrameEncoderUnprotected.java b/src/java/org/apache/cassandra/net/FrameEncoderUnprotected.java
index 6158713eae..3d10acfafb 100644
--- a/src/java/org/apache/cassandra/net/FrameEncoderUnprotected.java
+++ b/src/java/org/apache/cassandra/net/FrameEncoderUnprotected.java
@@ -40,7 +40,7 @@ class FrameEncoderUnprotected extends FrameEncoder
static final PayloadAllocator allocator = (isSelfContained, capacity) ->
new Payload(isSelfContained, capacity, HEADER_LENGTH, 0);
- PayloadAllocator allocator()
+ public PayloadAllocator allocator()
{
return allocator;
}
diff --git a/src/java/org/apache/cassandra/net/GlobalBufferPoolAllocator.java b/src/java/org/apache/cassandra/net/GlobalBufferPoolAllocator.java
index 66cbc9e1eb..d3686425a9 100644
--- a/src/java/org/apache/cassandra/net/GlobalBufferPoolAllocator.java
+++ b/src/java/org/apache/cassandra/net/GlobalBufferPoolAllocator.java
@@ -25,9 +25,9 @@ import org.apache.cassandra.utils.memory.BufferPool;
/**
* Primary {@link ByteBuf} / {@link ByteBuffer} allocator - using the global {@link BufferPool}.
*/
-class GlobalBufferPoolAllocator extends BufferPoolAllocator
+public class GlobalBufferPoolAllocator extends BufferPoolAllocator
{
- static final GlobalBufferPoolAllocator instance = new GlobalBufferPoolAllocator();
+ public static final GlobalBufferPoolAllocator instance = new GlobalBufferPoolAllocator();
private GlobalBufferPoolAllocator()
{
diff --git a/src/java/org/apache/cassandra/net/InboundMessageHandler.java b/src/java/org/apache/cassandra/net/InboundMessageHandler.java
index 2cac3ebfd6..64d0a8c945 100644
--- a/src/java/org/apache/cassandra/net/InboundMessageHandler.java
+++ b/src/java/org/apache/cassandra/net/InboundMessageHandler.java
@@ -20,23 +20,14 @@ package org.apache.cassandra.net;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.IdentityHashMap;
-import java.util.List;
-import java.util.Map;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.function.Consumer;
-import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.channel.EventLoop;
import org.apache.cassandra.concurrent.ExecutorLocals;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.exceptions.IncompatibleSchemaException;
@@ -53,135 +44,44 @@ import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.NoSpamLogger;
-import static java.lang.Math.max;
-import static java.lang.Math.min;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
-import static org.apache.cassandra.net.Crc.*;
import static org.apache.cassandra.utils.MonotonicClock.approxTime;
/**
- * Core logic for handling inbound message deserialization and execution (in tandem with {@link FrameDecoder}).
- *
- * Handles small and large messages, corruption, flow control, dispatch of message processing onto an appropriate
- * thread pool.
- *
- * # Interaction with {@link FrameDecoder}
- *
- * {@link InboundMessageHandler} sits on top of a {@link FrameDecoder} in the Netty pipeline, and is tightly
- * coupled with it.
- *
- * {@link FrameDecoder} decodes inbound frames and relies on a supplied {@link FrameProcessor} to act on them.
- * {@link InboundMessageHandler} provides two implementations of that interface:
- * - {@link #process(Frame)} is the default, primary processor, and the primary entry point to this class
- * - {@link UpToOneMessageFrameProcessor}, supplied to the decoder when the handler is reactivated after being
- * put in waiting mode due to lack of acquirable reserve memory capacity permits
- *
- * Return value of {@link FrameProcessor#process(Frame)} determines whether the decoder should keep processing
- * frames (if {@code true} is returned) or stop until explicitly reactivated (if {@code false} is). To reactivate
- * the decoder (once notified of available resource permits), {@link FrameDecoder#reactivate()} is invoked.
- *
- * # Frames
- *
- * {@link InboundMessageHandler} operates on frames of messages, and there are several kinds of them:
- * 1. {@link IntactFrame} that are contained. As names suggest, these contain one or multiple fully contained
- * messages believed to be uncorrupted. Guaranteed to not contain an part of an incomplete message.
- * See {@link #processFrameOfContainedMessages(ShareableBytes, Limit, Limit)}.
- * 2. {@link IntactFrame} that are NOT contained. These are uncorrupted parts of a large message split over multiple
- * parts due to their size. Can represent first or subsequent frame of a large message.
- * See {@link #processFirstFrameOfLargeMessage(IntactFrame, Limit, Limit)} and
- * {@link #processSubsequentFrameOfLargeMessage(Frame)}.
- * 3. {@link CorruptFrame} with corrupt header. These are unrecoverable, and force a connection to be dropped.
- * 4. {@link CorruptFrame} with a valid header, but corrupt payload. These can be either contained or uncontained.
- * - contained frames with corrupt payload can be gracefully dropped without dropping the connection
- * - uncontained frames with corrupt payload can be gracefully dropped unless they represent the first
- * frame of a new large message, as in that case we don't know how many bytes to skip
- * See {@link #processCorruptFrame(CorruptFrame)}.
- *
- * Fundamental frame invariants:
- * 1. A contained frame can only have fully-encapsulated messages - 1 to n, that don't cross frame boundaries
- * 2. An uncontained frame can hold a part of one message only. It can NOT, say, contain end of one large message
- * and a beginning of another one. All the bytes in an uncontained frame always belong to a single message.
+ * Implementation of {@link AbstractMessageHandler} for processing internode messages from peers.
*
* # Small vs large messages
- *
- * A single handler is equipped to process both small and large messages, potentially interleaved, but the logic
- * differs depending on size. Small messages are deserialized in place, and then handed off to an appropriate
+ * Small messages are deserialized in place, and then handed off to an appropriate
* thread pool for processing. Large messages accumulate frames until completion of a message, then hand off
* the untouched frames to the correct thread pool for the verb to be deserialized there and immediately processed.
*
- * See {@link LargeMessage} for details of the large-message accumulating state-machine, and {@link ProcessMessage}
- * and its inheritors for the differences in execution.
- *
* # Flow control (backpressure)
*
* To prevent nodes from overwhelming and bringing each other to the knees with more inbound messages that
* can be processed in a timely manner, {@link InboundMessageHandler} implements a strict flow control policy.
+ * The size of the incoming message is dependent on the messaging version of the specific peer connection. See
+ * {@link Message.Serializer#inferMessageSize(ByteBuffer, int, int, int)}.
*
- * Before we attempt to process a message fully, we first infer its size from the stream. Then we attempt to
- * acquire memory permits for a message of that size. If we succeed, then we move on actually process the message.
- * If we fail, the frame decoder deactivates until sufficient permits are released for the message to be processed
- * and the handler is activated again. Permits are released back once the message has been fully processed -
- * after the verb handler has been invoked - on the {@link Stage} for the {@link Verb} of the message.
+ * By default, every connection has 4MiB of exlusive permits available before needing to access the per-endpoint
+ * and global reserves.
*
- * Every connection has an exclusive number of permits allocated to it (by default 4MiB). In addition to it,
- * there is a per-endpoint reserve capacity and a global reserve capacity {@link Limit}, shared between all
- * connections from the same host and all connections, respectively. So long as long as the handler stays within
- * its exclusive limit, it doesn't need to tap into reserve capacity.
- *
- * If tapping into reserve capacity is necessary, but the handler fails to acquire capacity from either
- * endpoint of global reserve (and it needs to acquire from both), the handler and its frame decoder become
- * inactive and register with a {@link WaitQueue} of the appropriate type, depending on which of the reserves
- * couldn't be tapped into. Once enough messages have finished processing and had their permits released back
- * to the reserves, {@link WaitQueue} will reactivate the sleeping handlers and they'll resume processing frames.
- *
- * The reason we 'split' reserve capacity into two limits - endpoing and global - is to guarantee liveness, and
- * prevent single endpoint's connections from taking over the whole reserve, starving other connections.
- *
- * One permit per byte of serialized message gets acquired. When inflated on-heap, each message will occupy more
- * than that, necessarily, but despite wide variance, it's a good enough proxy that correlates with on-heap footprint.
+ * Permits are released after the verb handler has been invoked on the {@link Stage} for the {@link Verb} of the message.
*/
-public class InboundMessageHandler extends ChannelInboundHandlerAdapter implements FrameProcessor
+public class InboundMessageHandler extends AbstractMessageHandler
{
private static final Logger logger = LoggerFactory.getLogger(InboundMessageHandler.class);
private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.SECONDS);
private static final Message.Serializer serializer = Message.serializer;
- private final FrameDecoder decoder;
-
private final ConnectionType type;
- private final Channel channel;
private final InetAddressAndPort self;
private final InetAddressAndPort peer;
private final int version;
- private final int largeThreshold;
- private LargeMessage largeMessage;
-
- private final long queueCapacity;
- volatile long queueSize = 0L;
- private static final AtomicLongFieldUpdater queueSizeUpdater =
- AtomicLongFieldUpdater.newUpdater(InboundMessageHandler.class, "queueSize");
-
- private final Limit endpointReserveCapacity;
- private final WaitQueue endpointWaitQueue;
-
- private final Limit globalReserveCapacity;
- private final WaitQueue globalWaitQueue;
-
- private final OnHandlerClosed onClosed;
private final InboundMessageCallbacks callbacks;
private final Consumer> consumer;
- // wait queue handle, non-null if we overrun endpoint or global capacity and request to be resumed once it's released
- private WaitQueue.Ticket ticket = null;
-
- long corruptFramesRecovered, corruptFramesUnrecovered;
- long receivedCount, receivedBytes;
- long throttledCount, throttledNanos;
-
- private boolean isClosed;
-
InboundMessageHandler(FrameDecoder decoder,
ConnectionType type,
@@ -201,78 +101,26 @@ public class InboundMessageHandler extends ChannelInboundHandlerAdapter implemen
InboundMessageCallbacks callbacks,
Consumer> consumer)
{
- this.decoder = decoder;
+ super(decoder,
+ channel,
+ largeThreshold,
+ queueCapacity,
+ endpointReserveCapacity,
+ globalReserveCapacity,
+ endpointWaitQueue,
+ globalWaitQueue,
+ onClosed);
+
this.type = type;
- this.channel = channel;
this.self = self;
this.peer = peer;
this.version = version;
- this.largeThreshold = largeThreshold;
-
- this.queueCapacity = queueCapacity;
- this.endpointReserveCapacity = endpointReserveCapacity;
- this.endpointWaitQueue = endpointWaitQueue;
- this.globalReserveCapacity = globalReserveCapacity;
- this.globalWaitQueue = globalWaitQueue;
-
- this.onClosed = onClosed;
this.callbacks = callbacks;
this.consumer = consumer;
}
- @Override
- public void channelRead(ChannelHandlerContext ctx, Object msg)
- {
- /*
- * InboundMessageHandler works in tandem with FrameDecoder to implement flow control
- * and work stashing optimally. We rely on FrameDecoder to invoke the provided
- * FrameProcessor rather than on the pipeline and invocations of channelRead().
- * process(Frame) is the primary entry point for this class.
- */
- throw new IllegalStateException("InboundMessageHandler doesn't expect channelRead() to be invoked");
- }
-
- @Override
- public void handlerAdded(ChannelHandlerContext ctx)
- {
- decoder.activate(this); // the frame decoder starts inactive until explicitly activated by the added inbound message handler
- }
-
- @Override
- public boolean process(Frame frame) throws IOException
- {
- if (frame instanceof IntactFrame)
- return processIntactFrame((IntactFrame) frame, endpointReserveCapacity, globalReserveCapacity);
-
- processCorruptFrame((CorruptFrame) frame);
- return true;
- }
-
- private boolean processIntactFrame(IntactFrame frame, Limit endpointReserve, Limit globalReserve) throws IOException
- {
- if (frame.isSelfContained)
- return processFrameOfContainedMessages(frame.contents, endpointReserve, globalReserve);
- else if (null == largeMessage)
- return processFirstFrameOfLargeMessage(frame, endpointReserve, globalReserve);
- else
- return processSubsequentFrameOfLargeMessage(frame);
- }
-
- /*
- * Handle contained messages (not crossing boundaries of the frame) - both small and large, for the inbound
- * definition of large (breaching the size threshold for what we are willing to process on event-loop vs.
- * off event-loop).
- */
- private boolean processFrameOfContainedMessages(ShareableBytes bytes, Limit endpointReserve, Limit globalReserve) throws IOException
- {
- while (bytes.hasRemaining())
- if (!processOneContainedMessage(bytes, endpointReserve, globalReserve))
- return false;
- return true;
- }
-
- private boolean processOneContainedMessage(ShareableBytes bytes, Limit endpointReserve, Limit globalReserve) throws IOException
+ protected boolean processOneContainedMessage(ShareableBytes bytes, Limit endpointReserve, Limit globalReserve) throws IOException
{
ByteBuffer buf = bytes.get();
@@ -358,7 +206,7 @@ public class InboundMessageHandler extends ChannelInboundHandlerAdapter implemen
* Handling of multi-frame large messages
*/
- private boolean processFirstFrameOfLargeMessage(IntactFrame frame, Limit endpointReserve, Limit globalReserve) throws IOException
+ protected boolean processFirstFrameOfLargeMessage(IntactFrame frame, Limit endpointReserve, Limit globalReserve) throws IOException
{
ShareableBytes bytes = frame.contents;
ByteBuffer buf = bytes.get();
@@ -378,305 +226,34 @@ public class InboundMessageHandler extends ChannelInboundHandlerAdapter implemen
return true;
}
- private boolean processSubsequentFrameOfLargeMessage(Frame frame)
- {
- receivedBytes += frame.frameSize;
- if (largeMessage.supply(frame))
- {
- receivedCount++;
- largeMessage = null;
- }
- return true;
- }
-
- /*
- * We can handle some corrupt frames gracefully without dropping the connection and losing all the
- * queued up messages, but not others.
- *
- * Corrupt frames that *ARE NOT* safe to skip gracefully and require the connection to be dropped:
- * - any frame with corrupt header (!frame.isRecoverable())
- * - first corrupt-payload frame of a large message (impossible to infer message size, and without it
- * impossible to skip the message safely
- *
- * Corrupt frames that *ARE* safe to skip gracefully, without reconnecting:
- * - any self-contained frame with a corrupt payload (but not header): we lose all the messages in the
- * frame, but that has no effect on subsequent ones
- * - any non-first payload-corrupt frame of a large message: we know the size of the large message in
- * flight, so we just skip frames until we've seen all its bytes; we only lose the large message
- */
- private void processCorruptFrame(CorruptFrame frame) throws InvalidCrc
+ protected void processCorruptFrame(CorruptFrame frame) throws Crc.InvalidCrc
{
if (!frame.isRecoverable())
{
corruptFramesUnrecovered++;
- throw new InvalidCrc(frame.readCRC, frame.computedCRC);
+ throw new Crc.InvalidCrc(frame.readCRC, frame.computedCRC);
}
else if (frame.isSelfContained)
{
receivedBytes += frame.frameSize;
corruptFramesRecovered++;
- noSpamLogger.warn("{} invalid, recoverable CRC mismatch detected while reading messages (corrupted self-contained frame)", this);
+ noSpamLogger.warn("{} invalid, recoverable CRC mismatch detected while reading messages (corrupted self-contained frame)", id());
}
else if (null == largeMessage) // first frame of a large message
{
receivedBytes += frame.frameSize;
corruptFramesUnrecovered++;
- noSpamLogger.error("{} invalid, unrecoverable CRC mismatch detected while reading messages (corrupted first frame of a large message)", this);
- throw new InvalidCrc(frame.readCRC, frame.computedCRC);
+ noSpamLogger.error("{} invalid, unrecoverable CRC mismatch detected while reading messages (corrupted first frame of a large message)", id());
+ throw new Crc.InvalidCrc(frame.readCRC, frame.computedCRC);
}
else // subsequent frame of a large message
{
processSubsequentFrameOfLargeMessage(frame);
corruptFramesRecovered++;
- noSpamLogger.warn("{} invalid, recoverable CRC mismatch detected while reading a large message", this);
+ noSpamLogger.warn("{} invalid, recoverable CRC mismatch detected while reading a large message", id());
}
}
- private void onEndpointReserveCapacityRegained(Limit endpointReserve, long elapsedNanos)
- {
- onReserveCapacityRegained(endpointReserve, globalReserveCapacity, elapsedNanos);
- }
-
- private void onGlobalReserveCapacityRegained(Limit globalReserve, long elapsedNanos)
- {
- onReserveCapacityRegained(endpointReserveCapacity, globalReserve, elapsedNanos);
- }
-
- private void onReserveCapacityRegained(Limit endpointReserve, Limit globalReserve, long elapsedNanos)
- {
- if (isClosed)
- return;
-
- assert channel.eventLoop().inEventLoop();
-
- ticket = null;
- throttledNanos += elapsedNanos;
-
- try
- {
- /*
- * Process up to one message using supplied overriden reserves - one of them pre-allocated,
- * and guaranteed to be enough for one message - then, if no obstacles enountered, reactivate
- * the frame decoder using normal reserve capacities.
- */
- if (processUpToOneMessage(endpointReserve, globalReserve))
- decoder.reactivate();
- }
- catch (Throwable t)
- {
- exceptionCaught(t);
- }
- }
-
- // return true if the handler should be reactivated - if no new hurdles were encountered,
- // like running out of the other kind of reserve capacity
- private boolean processUpToOneMessage(Limit endpointReserve, Limit globalReserve) throws IOException
- {
- UpToOneMessageFrameProcessor processor = new UpToOneMessageFrameProcessor(endpointReserve, globalReserve);
- decoder.processBacklog(processor);
- return processor.isActive;
- }
-
- /*
- * Process at most one message. Won't always be an entire one (if the message in the head of line
- * is a large one, and there aren't sufficient frames to decode it entirely), but will never be more than one.
- */
- private class UpToOneMessageFrameProcessor implements FrameProcessor
- {
- private final Limit endpointReserve;
- private final Limit globalReserve;
-
- boolean isActive = true;
- boolean firstFrame = true;
-
- private UpToOneMessageFrameProcessor(Limit endpointReserve, Limit globalReserve)
- {
- this.endpointReserve = endpointReserve;
- this.globalReserve = globalReserve;
- }
-
- @Override
- public boolean process(Frame frame) throws IOException
- {
- if (firstFrame)
- {
- if (!(frame instanceof IntactFrame))
- throw new IllegalStateException("First backlog frame must be intact");
- firstFrame = false;
- return processFirstFrame((IntactFrame) frame);
- }
-
- return processSubsequentFrame(frame);
- }
-
- private boolean processFirstFrame(IntactFrame frame) throws IOException
- {
- if (frame.isSelfContained)
- {
- isActive = processOneContainedMessage(frame.contents, endpointReserve, globalReserve);
- return false; // stop after one message
- }
- else
- {
- isActive = processFirstFrameOfLargeMessage(frame, endpointReserve, globalReserve);
- return isActive; // continue unless fallen behind coprocessor or ran out of reserve capacity again
- }
- }
-
- private boolean processSubsequentFrame(Frame frame) throws IOException
- {
- if (frame instanceof IntactFrame)
- processSubsequentFrameOfLargeMessage(frame);
- else
- processCorruptFrame((CorruptFrame) frame);
-
- return largeMessage != null; // continue until done with the large message
- }
- }
-
- /**
- * Try to acquire permits for the inbound message. In case of failure, register with the right wait queue to be
- * reactivated once permit capacity is regained.
- */
- @SuppressWarnings("BooleanMethodIsAlwaysInverted")
- private boolean acquireCapacity(Limit endpointReserve, Limit globalReserve, int bytes, long currentTimeNanos, long expiresAtNanos)
- {
- ResourceLimits.Outcome outcome = acquireCapacity(endpointReserve, globalReserve, bytes);
-
- if (outcome == ResourceLimits.Outcome.INSUFFICIENT_ENDPOINT)
- ticket = endpointWaitQueue.register(this, bytes, currentTimeNanos, expiresAtNanos);
- else if (outcome == ResourceLimits.Outcome.INSUFFICIENT_GLOBAL)
- ticket = globalWaitQueue.register(this, bytes, currentTimeNanos, expiresAtNanos);
-
- if (outcome != ResourceLimits.Outcome.SUCCESS)
- throttledCount++;
-
- return outcome == ResourceLimits.Outcome.SUCCESS;
- }
-
- private ResourceLimits.Outcome acquireCapacity(Limit endpointReserve, Limit globalReserve, int bytes)
- {
- long currentQueueSize = queueSize;
-
- /*
- * acquireCapacity() is only ever called on the event loop, and as such queueSize is only ever increased
- * on the event loop. If there is enough capacity, we can safely addAndGet() and immediately return.
- */
- if (currentQueueSize + bytes <= queueCapacity)
- {
- queueSizeUpdater.addAndGet(this, bytes);
- return ResourceLimits.Outcome.SUCCESS;
- }
-
- // we know we don't have enough local queue capacity for the entire message, so we need to borrow some from reserve capacity
- long allocatedExcess = min(currentQueueSize + bytes - queueCapacity, bytes);
-
- if (!globalReserve.tryAllocate(allocatedExcess))
- return ResourceLimits.Outcome.INSUFFICIENT_GLOBAL;
-
- if (!endpointReserve.tryAllocate(allocatedExcess))
- {
- globalReserve.release(allocatedExcess);
- globalWaitQueue.signal();
- return ResourceLimits.Outcome.INSUFFICIENT_ENDPOINT;
- }
-
- long newQueueSize = queueSizeUpdater.addAndGet(this, bytes);
- long actualExcess = max(0, min(newQueueSize - queueCapacity, bytes));
-
- /*
- * It's possible that some permits were released at some point after we loaded current queueSize,
- * and we can satisfy more of the permits using our exclusive per-connection capacity, needing
- * less than previously estimated from the reserves. If that's the case, release the now unneeded
- * permit excess back to endpoint/global reserves.
- */
- if (actualExcess != allocatedExcess) // actualExcess < allocatedExcess
- {
- long excess = allocatedExcess - actualExcess;
-
- endpointReserve.release(excess);
- globalReserve.release(excess);
-
- endpointWaitQueue.signal();
- globalWaitQueue.signal();
- }
-
- return ResourceLimits.Outcome.SUCCESS;
- }
-
- private void releaseCapacity(int bytes)
- {
- long oldQueueSize = queueSizeUpdater.getAndAdd(this, -bytes);
- if (oldQueueSize > queueCapacity)
- {
- long excess = min(oldQueueSize - queueCapacity, bytes);
-
- endpointReserveCapacity.release(excess);
- globalReserveCapacity.release(excess);
-
- endpointWaitQueue.signal();
- globalWaitQueue.signal();
- }
- }
-
- /**
- * Invoked to release capacity for a message that has been fully, successfully processed.
- *
- * Normally no different from invoking {@link #releaseCapacity(int)}, but is necessary for the verifier
- * to be able to delay capacity release for backpressure testing.
- */
- @VisibleForTesting
- protected void releaseProcessedCapacity(int size, Header header)
- {
- releaseCapacity(size);
- }
-
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
- {
- try
- {
- exceptionCaught(cause);
- }
- catch (Throwable t)
- {
- logger.error("Unexpected exception in {}.exceptionCaught", this.getClass().getSimpleName(), t);
- }
- }
-
- private void exceptionCaught(Throwable cause)
- {
- decoder.discard();
-
- JVMStabilityInspector.inspectThrowable(cause);
-
- if (cause instanceof Message.InvalidLegacyProtocolMagic)
- logger.error("{} invalid, unrecoverable CRC mismatch detected while reading messages - closing the connection", id());
- else
- logger.error("{} unexpected exception caught while processing inbound messages; terminating connection", id(), cause);
-
- channel.close();
- }
-
- @Override
- public void channelInactive(ChannelHandlerContext ctx)
- {
- isClosed = true;
-
- if (null != largeMessage)
- largeMessage.abort();
-
- if (null != ticket)
- ticket.invalidate();
-
- onClosed.call(this);
- }
-
- private EventLoop eventLoop()
- {
- return channel.eventLoop();
- }
-
String id(boolean includeReal)
{
if (!includeReal)
@@ -687,7 +264,7 @@ public class InboundMessageHandler extends ChannelInboundHandlerAdapter implemen
type, channel.id().asShortText());
}
- String id()
+ protected String id()
{
return SocketFactory.channelId(peer, self, type, channel.id().asShortText());
}
@@ -698,6 +275,33 @@ public class InboundMessageHandler extends ChannelInboundHandlerAdapter implemen
return id();
}
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
+ {
+ try
+ {
+ fatalExceptionCaught(cause);
+ }
+ catch (Throwable t)
+ {
+ logger.error("Unexpected exception in {}.exceptionCaught", this.getClass().getSimpleName(), t);
+ }
+ }
+
+ protected void fatalExceptionCaught(Throwable cause)
+ {
+ decoder.discard();
+
+ JVMStabilityInspector.inspectThrowable(cause);
+
+ if (cause instanceof Message.InvalidLegacyProtocolMagic)
+ logger.error("{} invalid, unrecoverable CRC mismatch detected while reading messages - closing the connection", id());
+ else
+ logger.error("{} unexpected exception caught while processing inbound messages; terminating connection", id(), cause);
+
+ channel.close();
+ }
+
/*
* A large-message frame-accumulating state machine.
*
@@ -710,28 +314,16 @@ public class InboundMessageHandler extends ChannelInboundHandlerAdapter implemen
* accumulated frames, or in gathering more - so we release the ones we already have, and
* skip any remaining ones, alongside with returning memory permits early.
*/
- private class LargeMessage
+ private class LargeMessage extends AbstractMessageHandler.LargeMessage
{
- private final int size;
- private final Header header;
-
- private final List buffers = new ArrayList<>();
- private int received;
-
- private boolean isExpired;
- private boolean isCorrupt;
-
private LargeMessage(int size, Header header, boolean isExpired)
{
- this.size = size;
- this.header = header;
- this.isExpired = isExpired;
+ super(size, header, header.expiresAtNanos, isExpired);
}
private LargeMessage(int size, Header header, ShareableBytes bytes)
{
- this(size, header, false);
- buffers.add(bytes);
+ super(size, header, header.expiresAtNanos, bytes);
}
private void schedule()
@@ -739,47 +331,7 @@ public class InboundMessageHandler extends ChannelInboundHandlerAdapter implemen
dispatch(new ProcessLargeMessage(this));
}
- /**
- * Return true if this was the last frame of the large message.
- */
- private boolean supply(Frame frame)
- {
- if (frame instanceof IntactFrame)
- onIntactFrame((IntactFrame) frame);
- else
- onCorruptFrame();
-
- received += frame.frameSize;
- if (size == received)
- onComplete();
- return size == received;
- }
-
- private void onIntactFrame(IntactFrame frame)
- {
- boolean expires = approxTime.isAfter(header.expiresAtNanos);
- if (!isExpired && !isCorrupt)
- {
- if (!expires)
- {
- buffers.add(frame.contents.sliceAndConsume(frame.frameSize).share());
- return;
- }
- releaseBuffersAndCapacity(); // release resources once we transition from normal state to expired
- }
- frame.consume();
- isExpired |= expires;
- }
-
- private void onCorruptFrame()
- {
- if (!isExpired && !isCorrupt)
- releaseBuffersAndCapacity(); // release resources once we transition from normal state to corrupt
- isCorrupt = true;
- isExpired |= approxTime.isAfter(header.expiresAtNanos);
- }
-
- private void onComplete()
+ protected void onComplete()
{
long timeElapsed = approxTime.now() - header.createdAtNanos;
@@ -798,23 +350,13 @@ public class InboundMessageHandler extends ChannelInboundHandlerAdapter implemen
}
}
- private void abort()
+ protected void abort()
{
if (!isExpired && !isCorrupt)
releaseBuffersAndCapacity(); // release resources if in normal state when abort() is invoked
callbacks.onClosedBeforeArrival(size, header, received, isCorrupt, isExpired);
}
- private void releaseBuffers()
- {
- buffers.forEach(ShareableBytes::release); buffers.clear();
- }
-
- private void releaseBuffersAndCapacity()
- {
- releaseBuffers(); releaseCapacity(size);
- }
-
private Message deserialize()
{
try (ChunkedInputPlus input = ChunkedInputPlus.of(buffers))
@@ -968,232 +510,4 @@ public class InboundMessageHandler extends ChannelInboundHandlerAdapter implemen
message.releaseBuffers(); // releases buffers if they haven't been yet (by deserialize() call)
}
}
-
- /**
- * A special-purpose wait queue to park inbound message handlers that failed to allocate
- * reserve capacity for a message in. Upon such failure a handler registers itself with
- * a {@link WaitQueue} of the appropriate kind (either ENDPOINT or GLOBAL - if failed
- * to allocate endpoint or global reserve capacity, respectively), stops processing any
- * accumulated frames or receiving new ones, and waits - until reactivated.
- *
- * Every time permits are returned to an endpoint or global {@link Limit}, the respective
- * queue gets signalled, and if there are any handlers registered in it, we will attempt
- * to reactivate as many waiting handlers as current available reserve capacity allows
- * us to - immediately, on the {@link #signal()}-calling thread. At most one such attempt
- * will be in progress at any given time.
- *
- * Handlers that can be reactivated will be grouped by their {@link EventLoop} and a single
- * {@link ReactivateHandlers} task will be scheduled per event loop, on the corresponding
- * event loops.
- *
- * When run, the {@link ReactivateHandlers} task will ask each handler in its group to first
- * process one message - using preallocated reserve capacity - and if no obstacles were met -
- * reactivate the handlers, this time using their regular reserves.
- *
- * See {@link WaitQueue#schedule()}, {@link ReactivateHandlers#run()}, {@link Ticket#reactivateHandler(Limit)}.
- */
- public static final class WaitQueue
- {
- enum Kind { ENDPOINT, GLOBAL }
-
- private static final int NOT_RUNNING = 0;
- @SuppressWarnings("unused")
- private static final int RUNNING = 1;
- private static final int RUN_AGAIN = 2;
-
- private volatile int scheduled;
- private static final AtomicIntegerFieldUpdater scheduledUpdater =
- AtomicIntegerFieldUpdater.newUpdater(WaitQueue.class, "scheduled");
-
- private final Kind kind;
- private final Limit reserveCapacity;
-
- private final ManyToOneConcurrentLinkedQueue queue = new ManyToOneConcurrentLinkedQueue<>();
-
- private WaitQueue(Kind kind, Limit reserveCapacity)
- {
- this.kind = kind;
- this.reserveCapacity = reserveCapacity;
- }
-
- public static WaitQueue endpoint(Limit endpointReserveCapacity)
- {
- return new WaitQueue(Kind.ENDPOINT, endpointReserveCapacity);
- }
-
- public static WaitQueue global(Limit globalReserveCapacity)
- {
- return new WaitQueue(Kind.GLOBAL, globalReserveCapacity);
- }
-
- private Ticket register(InboundMessageHandler handler, int bytesRequested, long registeredAtNanos, long expiresAtNanos)
- {
- Ticket ticket = new Ticket(this, handler, bytesRequested, registeredAtNanos, expiresAtNanos);
- Ticket previous = queue.relaxedPeekLastAndOffer(ticket);
- if (null == previous || !previous.isWaiting())
- signal(); // only signal the queue if this handler is first to register
- return ticket;
- }
-
- private void signal()
- {
- if (queue.relaxedIsEmpty())
- return; // we can return early if no handlers have registered with the wait queue
-
- if (NOT_RUNNING == scheduledUpdater.getAndUpdate(this, i -> min(RUN_AGAIN, i + 1)))
- {
- do
- {
- schedule();
- }
- while (RUN_AGAIN == scheduledUpdater.getAndDecrement(this));
- }
- }
-
- private void schedule()
- {
- Map tasks = null;
-
- long currentTimeNanos = approxTime.now();
-
- Ticket t;
- while ((t = queue.peek()) != null)
- {
- if (!t.call()) // invalidated
- {
- queue.remove();
- continue;
- }
-
- boolean isLive = t.isLive(currentTimeNanos);
- if (isLive && !reserveCapacity.tryAllocate(t.bytesRequested))
- {
- if (!t.reset()) // the ticket was invalidated after being called but before now
- {
- queue.remove();
- continue;
- }
- break; // TODO: traverse the entire queue to unblock handlers that have expired or invalidated tickets
- }
-
- if (null == tasks)
- tasks = new IdentityHashMap<>();
-
- queue.remove();
- tasks.computeIfAbsent(t.handler.eventLoop(), e -> new ReactivateHandlers()).add(t, isLive);
- }
-
- if (null != tasks)
- tasks.forEach(EventLoop::execute);
- }
-
- private class ReactivateHandlers implements Runnable
- {
- List tickets = new ArrayList<>();
- long capacity = 0L;
-
- private void add(Ticket ticket, boolean isLive)
- {
- tickets.add(ticket);
- if (isLive) capacity += ticket.bytesRequested;
- }
-
- public void run()
- {
- Limit limit = new ResourceLimits.Basic(capacity);
- try
- {
- for (Ticket ticket : tickets)
- ticket.reactivateHandler(limit);
- }
- finally
- {
- /*
- * Free up any unused capacity, if any. Will be non-zero if one or more handlers were closed
- * when we attempted to run their callback, or used more of their other reserve; or if the first
- * message in the unprocessed stream has expired in the narrow time window.
- */
- long remaining = limit.remaining();
- if (remaining > 0)
- {
- reserveCapacity.release(remaining);
- signal();
- }
- }
- }
- }
-
- private static final class Ticket
- {
- private static final int WAITING = 0;
- private static final int CALLED = 1;
- private static final int INVALIDATED = 2; // invalidated by a handler that got closed
-
- private volatile int state;
- private static final AtomicIntegerFieldUpdater stateUpdater =
- AtomicIntegerFieldUpdater.newUpdater(Ticket.class, "state");
-
- private final WaitQueue waitQueue;
- private final InboundMessageHandler handler;
- private final int bytesRequested;
- private final long reigsteredAtNanos;
- private final long expiresAtNanos;
-
- private Ticket(WaitQueue waitQueue, InboundMessageHandler handler, int bytesRequested, long registeredAtNanos, long expiresAtNanos)
- {
- this.waitQueue = waitQueue;
- this.handler = handler;
- this.bytesRequested = bytesRequested;
- this.reigsteredAtNanos = registeredAtNanos;
- this.expiresAtNanos = expiresAtNanos;
- }
-
- private void reactivateHandler(Limit capacity)
- {
- long elapsedNanos = approxTime.now() - reigsteredAtNanos;
- try
- {
- if (waitQueue.kind == Kind.ENDPOINT)
- handler.onEndpointReserveCapacityRegained(capacity, elapsedNanos);
- else
- handler.onGlobalReserveCapacityRegained(capacity, elapsedNanos);
- }
- catch (Throwable t)
- {
- logger.error("{} exception caught while reactivating a handler", handler.id(), t);
- }
- }
-
- private boolean isWaiting()
- {
- return state == WAITING;
- }
-
- private boolean isLive(long currentTimeNanos)
- {
- return !approxTime.isAfter(currentTimeNanos, expiresAtNanos);
- }
-
- private void invalidate()
- {
- state = INVALIDATED;
- waitQueue.signal();
- }
-
- private boolean call()
- {
- return stateUpdater.compareAndSet(this, WAITING, CALLED);
- }
-
- private boolean reset()
- {
- return stateUpdater.compareAndSet(this, CALLED, WAITING);
- }
- }
- }
-
- public interface OnHandlerClosed
- {
- void call(InboundMessageHandler handler);
- }
}
diff --git a/src/java/org/apache/cassandra/net/InboundMessageHandlers.java b/src/java/org/apache/cassandra/net/InboundMessageHandlers.java
index 4ebd5ad76a..4466466acb 100644
--- a/src/java/org/apache/cassandra/net/InboundMessageHandlers.java
+++ b/src/java/org/apache/cassandra/net/InboundMessageHandlers.java
@@ -28,7 +28,6 @@ import io.netty.channel.Channel;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.InternodeInboundMetrics;
import org.apache.cassandra.net.Message.Header;
-import org.apache.cassandra.utils.ApproximateTime;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.utils.MonotonicClock.approxTime;
@@ -161,10 +160,11 @@ public final class InboundMessageHandlers
metrics.release();
}
- private void onHandlerClosed(InboundMessageHandler handler)
+ private void onHandlerClosed(AbstractMessageHandler handler)
{
+ assert handler instanceof InboundMessageHandler;
handlers.remove(handler);
- absorbCounters(handler);
+ absorbCounters((InboundMessageHandler)handler);
}
/*
diff --git a/src/java/org/apache/cassandra/net/ResourceLimits.java b/src/java/org/apache/cassandra/net/ResourceLimits.java
index 4c97c2a483..7658d5f3fa 100644
--- a/src/java/org/apache/cassandra/net/ResourceLimits.java
+++ b/src/java/org/apache/cassandra/net/ResourceLimits.java
@@ -163,12 +163,12 @@ public abstract class ResourceLimits
/**
* A cheaper, thread-unsafe permit container to be used for unshared limits.
*/
- static class Basic implements Limit
+ public static class Basic implements Limit
{
private long limit;
private long using;
- Basic(long limit)
+ public Basic(long limit)
{
this.limit = limit;
}
diff --git a/src/java/org/apache/cassandra/net/ShareableBytes.java b/src/java/org/apache/cassandra/net/ShareableBytes.java
index 71c272a17d..1cc8fa6099 100644
--- a/src/java/org/apache/cassandra/net/ShareableBytes.java
+++ b/src/java/org/apache/cassandra/net/ShareableBytes.java
@@ -29,7 +29,7 @@ import org.apache.cassandra.utils.memory.BufferPools;
* When sharing is necessary, {@link #share()} method must be invoked by the owning thread
* before a {@link ShareableBytes} instance can be shared with another thread.
*/
-class ShareableBytes
+public class ShareableBytes
{
private final ByteBuffer bytes;
private final ShareableBytes owner;
@@ -53,18 +53,18 @@ class ShareableBytes
this.bytes = bytes;
}
- ByteBuffer get()
+ public ByteBuffer get()
{
assert owner.count != 0;
return bytes;
}
- boolean hasRemaining()
+ public boolean hasRemaining()
{
return bytes.hasRemaining();
}
- int remaining()
+ public int remaining()
{
return bytes.remaining();
}
@@ -84,7 +84,7 @@ class ShareableBytes
* The first invocation must occur while the calling thread has exclusive access (though there may be more
* than one 'owner', these must all either be owned by the calling thread or otherwise not being used)
*/
- ShareableBytes share()
+ public ShareableBytes share()
{
int count = owner.count;
if (count < 0)
@@ -119,7 +119,7 @@ class ShareableBytes
}
}
- void release()
+ public void release()
{
owner.doRelease();
}
@@ -147,7 +147,7 @@ class ShareableBytes
/**
* Create a slice over the next {@code length} bytes, consuming them from our buffer, and incrementing the owner count
*/
- ShareableBytes sliceAndConsume(int length)
+ public ShareableBytes sliceAndConsume(int length)
{
int begin = bytes.position();
int end = begin + length;
@@ -166,7 +166,7 @@ class ShareableBytes
return new ShareableBytes(owner.retain(), slice);
}
- static ShareableBytes wrap(ByteBuffer buffer)
+ public static ShareableBytes wrap(ByteBuffer buffer)
{
return new ShareableBytes(buffer);
}
diff --git a/src/java/org/apache/cassandra/service/NativeTransportService.java b/src/java/org/apache/cassandra/service/NativeTransportService.java
index f4f35856e3..7556f81982 100644
--- a/src/java/org/apache/cassandra/service/NativeTransportService.java
+++ b/src/java/org/apache/cassandra/service/NativeTransportService.java
@@ -36,7 +36,7 @@ import io.netty.util.Version;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.metrics.ClientMetrics;
-import org.apache.cassandra.transport.Message;
+import org.apache.cassandra.transport.Dispatcher;
import org.apache.cassandra.transport.Server;
import org.apache.cassandra.utils.NativeLibrary;
@@ -151,7 +151,7 @@ public class NativeTransportService
// shutdown executors used by netty for native transport server
workerGroup.shutdownGracefully(3, 5, TimeUnit.SECONDS).awaitUninterruptibly();
- Message.Dispatcher.shutdown();
+ Dispatcher.shutdown();
}
/**
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 8baf7fe22d..abcd25d67e 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -102,6 +102,7 @@ import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.schema.ViewMetadata;
import org.apache.cassandra.streaming.*;
import org.apache.cassandra.tracing.TraceKeyspace;
+import org.apache.cassandra.transport.ClientResourceLimits;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.transport.Server;
import org.apache.cassandra.utils.*;
@@ -5682,25 +5683,25 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
@Override
public long getNativeTransportMaxConcurrentRequestsInBytes()
{
- return Server.EndpointPayloadTracker.getGlobalLimit();
+ return ClientResourceLimits.getGlobalLimit();
}
@Override
public void setNativeTransportMaxConcurrentRequestsInBytes(long newLimit)
{
- Server.EndpointPayloadTracker.setGlobalLimit(newLimit);
+ ClientResourceLimits.setGlobalLimit(newLimit);
}
@Override
public long getNativeTransportMaxConcurrentRequestsInBytesPerIp()
{
- return Server.EndpointPayloadTracker.getEndpointLimit();
+ return ClientResourceLimits.getEndpointLimit();
}
@Override
public void setNativeTransportMaxConcurrentRequestsInBytesPerIp(long newLimit)
{
- Server.EndpointPayloadTracker.setEndpointLimit(newLimit);
+ ClientResourceLimits.setEndpointLimit(newLimit);
}
@VisibleForTesting
diff --git a/src/java/org/apache/cassandra/transport/CBUtil.java b/src/java/org/apache/cassandra/transport/CBUtil.java
index 6e0c8ff754..fd6a0ff7c3 100644
--- a/src/java/org/apache/cassandra/transport/CBUtil.java
+++ b/src/java/org/apache/cassandra/transport/CBUtil.java
@@ -600,10 +600,4 @@ public abstract class CBUtil
return bytes;
}
- public static int readUnsignedShort(ByteBuf buf)
- {
- int ch1 = buf.readByte() & 0xFF;
- int ch2 = buf.readByte() & 0xFF;
- return (ch1 << 8) + (ch2);
- }
}
diff --git a/src/java/org/apache/cassandra/transport/CQLMessageHandler.java b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java
new file mode 100644
index 0000000000..48cdf64bde
--- /dev/null
+++ b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java
@@ -0,0 +1,484 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.transport;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.primitives.Ints;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import org.apache.cassandra.exceptions.OverloadedException;
+import org.apache.cassandra.metrics.ClientMetrics;
+import org.apache.cassandra.metrics.ClientMessageSizeMetrics;
+import org.apache.cassandra.net.AbstractMessageHandler;
+import org.apache.cassandra.net.FrameDecoder;
+import org.apache.cassandra.net.FrameDecoder.IntactFrame;
+import org.apache.cassandra.net.FrameEncoder;
+import org.apache.cassandra.net.ResourceLimits;
+import org.apache.cassandra.net.ResourceLimits.Limit;
+import org.apache.cassandra.net.ShareableBytes;
+import org.apache.cassandra.transport.Flusher.FlushItem.Framed;
+import org.apache.cassandra.transport.messages.ErrorMessage;
+import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.NoSpamLogger;
+
+import static org.apache.cassandra.utils.MonotonicClock.approxTime;
+
+/**
+ * Implementation of {@link AbstractMessageHandler} for processing CQL messages which comprise a {@link Message} wrapped
+ * in an {@link Envelope}. This class is parameterized by a {@link Message} subtype, expected to be either
+ * {@link Message.Request} or {@link Message.Response}. Most commonly, an instance for handling {@link Message.Request}
+ * is created for each inbound CQL client connection.
+ *
+ * # Small vs large messages
+ * Small messages are deserialized in place, and then handed off to a consumer for processing.
+ * Large messages accumulate frames until all bytes for the envelope are received, then concatenate and deserialize the
+ * frames on the event loop thread and pass them on to the same consumer.
+ *
+ * # Flow control (backpressure)
+ *
+ * The size of an incoming message is explicit in the {@link Envelope.Header}.
+ * {@link org.apache.cassandra.net.Message.Serializer#inferMessageSize(ByteBuffer, int, int, int)}.
+ *
+ * By default, every connection has 1MiB of exlusive permits available before needing to access the per-endpoint
+ * and global reserves. By default, those reserves are sized proportionally to the heap - 2.5% of heap per-endpoint
+ * and a 10% for the global reserve.
+ *
+ * Permits are held while CQL messages are processed and released after the response has been encoded into the
+ * buffers of the response frame.
+ */
+public class CQLMessageHandler extends AbstractMessageHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(CQLMessageHandler.class);
+ private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.SECONDS);
+
+ public static final int LARGE_MESSAGE_THRESHOLD = FrameEncoder.Payload.MAX_SIZE - 1;
+
+ private final Envelope.Decoder envelopeDecoder;
+ private final Message.Decoder messageDecoder;
+ private final FrameEncoder.PayloadAllocator payloadAllocator;
+ private final MessageConsumer dispatcher;
+ private final ErrorHandler errorHandler;
+ private final boolean throwOnOverload;
+
+ long channelPayloadBytesInFlight;
+
+ interface MessageConsumer
+ {
+ void accept(Channel channel, M message, Dispatcher.FlushItemConverter toFlushItem);
+ }
+
+ interface ErrorHandler
+ {
+ void accept(Throwable error);
+ }
+
+ CQLMessageHandler(Channel channel,
+ FrameDecoder decoder,
+ Envelope.Decoder envelopeDecoder,
+ Message.Decoder messageDecoder,
+ MessageConsumer dispatcher,
+ FrameEncoder.PayloadAllocator payloadAllocator,
+ int queueCapacity,
+ ClientResourceLimits.ResourceProvider resources,
+ OnHandlerClosed onClosed,
+ ErrorHandler errorHandler,
+ boolean throwOnOverload)
+ {
+ super(decoder,
+ channel,
+ LARGE_MESSAGE_THRESHOLD,
+ queueCapacity,
+ resources.endpointLimit(),
+ resources.globalLimit(),
+ resources.endpointWaitQueue(),
+ resources.globalWaitQueue(),
+ onClosed);
+ this.envelopeDecoder = envelopeDecoder;
+ this.messageDecoder = messageDecoder;
+ this.payloadAllocator = payloadAllocator;
+ this.dispatcher = dispatcher;
+ this.errorHandler = errorHandler;
+ this.throwOnOverload = throwOnOverload;
+ }
+
+ protected boolean processOneContainedMessage(ShareableBytes bytes, Limit endpointReserve, Limit globalReserve)
+ {
+ Envelope.Header header = extractHeader(bytes);
+ // null indicates a failure to extract the CQL message header.
+ // This will trigger a protocol exception and closing of the connection.
+ if (null == header)
+ return false;
+
+ // max CQL message size defaults to 256mb, so should be safe to downcast
+ int messageSize = Ints.checkedCast(header.bodySizeInBytes);
+
+ if (throwOnOverload)
+ {
+ if (!acquireCapacity(header, endpointReserve, globalReserve))
+ {
+ // discard the request and throw an exception
+ ClientMetrics.instance.markRequestDiscarded();
+ logger.trace("Discarded request of size: {}. InflightChannelRequestPayload: {}, " +
+ "InflightEndpointRequestPayload: {}, InflightOverallRequestPayload: {}, Header: {}",
+ messageSize,
+ channelPayloadBytesInFlight,
+ endpointReserve.using(),
+ globalReserve.using(),
+ header);
+
+ handleError(new OverloadedException("Server is in overloaded state. " +
+ "Cannot accept more requests at this point"), header);
+
+ // Don't stop processing incoming messages, rely on the client to apply
+ // backpressure when it receives OverloadedException
+ // but discard this message as we're responding with the overloaded error
+ incrementReceivedMessageMetrics(messageSize);
+ ByteBuffer buf = bytes.get();
+ buf.position(buf.position() + Envelope.Header.LENGTH + messageSize);
+ return true;
+ }
+ }
+ else if (!acquireCapacityAndQueueOnFailure(header, endpointReserve, globalReserve))
+ {
+ // set backpressure on the channel, queuing the request until we have capacity
+ ClientMetrics.instance.pauseConnection();
+ return false;
+ }
+
+ channelPayloadBytesInFlight += messageSize;
+ incrementReceivedMessageMetrics(messageSize);
+ processRequest(composeRequest(header, bytes));
+ return true;
+ }
+
+ private void incrementReceivedMessageMetrics(int messageSize)
+ {
+ receivedCount++;
+ receivedBytes += messageSize + Envelope.Header.LENGTH;
+ ClientMessageSizeMetrics.bytesReceived.inc(messageSize + Envelope.Header.LENGTH);
+ ClientMessageSizeMetrics.bytesReceivedPerRequest.update(messageSize + Envelope.Header.LENGTH);
+ }
+
+ private Envelope.Header extractHeader(ShareableBytes bytes)
+ {
+ try
+ {
+ return envelopeDecoder.extractHeader(bytes.get());
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.error("{} unexpected exception caught while deserializing a message header", id(), t);
+ // Ideally we would recover from this as we could simply drop the entire frame.
+ // However, we have no good way to communicate this back to the client as we don't
+ // yet know the stream id. So this will trigger an Error response and the connection
+ // to be closed.
+ handleError(new ProtocolException("Deserialization error processing CQL payload"));
+ return null;
+ }
+ }
+
+ private Envelope composeRequest(Envelope.Header header, ShareableBytes bytes)
+ {
+ // extract body
+ ByteBuffer buf = bytes.get();
+ int idx = buf.position() + Envelope.Header.LENGTH;
+ final int end = idx + Ints.checkedCast(header.bodySizeInBytes);
+ ByteBuf body = Unpooled.wrappedBuffer(buf.slice());
+ body.readerIndex(Envelope.Header.LENGTH);
+ body.retain();
+ buf.position(end);
+ return new Envelope(header, body);
+ }
+
+ protected void processRequest(Envelope request)
+ {
+ M message = null;
+ try
+ {
+ message = messageDecoder.decode(channel, request);
+ dispatcher.accept(channel, message, this::toFlushItem);
+ }
+ catch (Exception e)
+ {
+ if (message != null)
+ request.release();
+ handleErrorAndRelease(e, request.header);
+ }
+ }
+
+ /**
+ * For "expected" errors this ensures we pass a WrappedException,
+ * which contains a streamId, to the error handler. This makes
+ * sure that whereever possible, the streamId is propagated back
+ * to the client.
+ * This also releases the capacity acquired for processing as
+ * indicated by supplied header.
+ * @param t
+ * @param header
+ */
+ private void handleErrorAndRelease(Throwable t, Envelope.Header header)
+ {
+ release(header);
+ handleError(t, header);
+ }
+
+ /**
+ * For "expected" errors this ensures we pass a WrappedException,
+ * which contains a streamId, to the error handler. This makes
+ * sure that whereever possible, the streamId is propagated back
+ * to the client.
+ * This variant doesn't call release as it is intended for use
+ * when an error occurs without any capacity being acquired.
+ * Typically, this would be the result of an acquisition failure
+ * if the THROW_ON_OVERLOAD option has been specified by the client.
+ * @param t
+ * @param header
+ */
+ private void handleError(Throwable t, Envelope.Header header)
+ {
+ errorHandler.accept(ErrorMessage.wrap(t, header.streamId));
+ }
+
+ /**
+ * For use in the case where the error can't be mapped to a specific stream id,
+ * such as a corrupted frame, or when extracting a CQL message from the frame's
+ * payload fails. This does not attempt to release any resources, as these errors
+ * should only occur before any capacity acquisition is attempted (e.g. on receipt
+ * of a corrupt frame, or failure to extract a CQL message from the envelope).
+ *
+ * @param t
+ */
+ private void handleError(Throwable t)
+ {
+ errorHandler.accept(t);
+ }
+
+ // Acts as a Dispatcher.FlushItemConverter
+ private Framed toFlushItem(Channel channel, Message.Request request, Message.Response response)
+ {
+ // Returns a FlushItem.Framed instance which wraps a Consumer that performs
+ // the work of returning the capacity allocated for processing the request.
+ // The Dispatcher will call this to obtain the FlushItem to enqueue with its Flusher once
+ // a dispatched request has been processed.
+
+ Envelope responseFrame = response.encode(request.getSource().header.version);
+ int responseSize = envelopeSize(responseFrame.header);
+ ClientMessageSizeMetrics.bytesSent.inc(responseSize);
+ ClientMessageSizeMetrics.bytesSentPerResponse.update(responseSize);
+
+ return new Framed(channel,
+ responseFrame,
+ request.getSource(),
+ payloadAllocator,
+ this::release);
+ }
+
+ private void release(Flusher.FlushItem flushItem)
+ {
+ release(flushItem.request.header);
+ flushItem.request.release();
+ flushItem.response.release();
+ }
+
+ private void release(Envelope.Header header)
+ {
+ releaseCapacity(Ints.checkedCast(header.bodySizeInBytes));
+ channelPayloadBytesInFlight -= header.bodySizeInBytes;
+ }
+
+ /*
+ * Handling of multi-frame large messages
+ */
+ protected boolean processFirstFrameOfLargeMessage(IntactFrame frame, Limit endpointReserve, Limit globalReserve) throws IOException
+ {
+ ShareableBytes bytes = frame.contents;
+ ByteBuffer buf = bytes.get();
+ try
+ {
+ Envelope.Header header = envelopeDecoder.extractHeader(buf);
+ // max CQL message size defaults to 256mb, so should be safe to downcast
+ int messageSize = Ints.checkedCast(header.bodySizeInBytes);
+ receivedBytes += buf.remaining();
+
+ if (throwOnOverload)
+ {
+ LargeMessage largeMessage = new LargeMessage(header);
+ if (!acquireCapacity(header, endpointReserve, globalReserve))
+ {
+ // discard the request and throw an exception
+ ClientMetrics.instance.markRequestDiscarded();
+ logger.trace("Discarded request of size: {}. InflightChannelRequestPayload: {}, " +
+ "InflightEndpointRequestPayload: {}, InflightOverallRequestPayload: {}, Header: {}",
+ messageSize,
+ channelPayloadBytesInFlight,
+ endpointReserve.using(),
+ globalReserve.using(),
+ header);
+
+ // mark as overloaded so that we consume
+ // subsequent frames and then discard the message
+ largeMessage.markOverloaded();
+ }
+ this.largeMessage = largeMessage;
+ largeMessage.supply(frame);
+ // Don't stop processing incoming frames, rely on the client to apply
+ // backpressure when it receives OverloadedException
+ return true;
+ }
+ else
+ {
+ if (!acquireCapacityAndQueueOnFailure(header, endpointReserve, globalReserve))
+ {
+ receivedBytes += frame.frameSize;
+ return false;
+ }
+ }
+
+ largeMessage = new LargeMessage(header);
+ largeMessage.supply(frame);
+ return true;
+ }
+ catch (Exception e)
+ {
+ throw new IOException("Error decoding CQL Message", e);
+ }
+ }
+
+ protected String id()
+ {
+ return channel.id().asShortText();
+ }
+
+ @SuppressWarnings("BooleanMethodIsAlwaysInverted")
+ private boolean acquireCapacityAndQueueOnFailure(Envelope.Header header, Limit endpointReserve, Limit globalReserve)
+ {
+ int bytesRequired = Ints.checkedCast(header.bodySizeInBytes);
+ long currentTimeNanos = approxTime.now();
+ return acquireCapacity(endpointReserve, globalReserve, bytesRequired, currentTimeNanos, Long.MAX_VALUE);
+ }
+
+ @SuppressWarnings("BooleanMethodIsAlwaysInverted")
+ private boolean acquireCapacity(Envelope.Header header, Limit endpointReserve, Limit globalReserve)
+ {
+ int bytesRequired = Ints.checkedCast(header.bodySizeInBytes);
+ return acquireCapacity(endpointReserve, globalReserve, bytesRequired) == ResourceLimits.Outcome.SUCCESS;
+ }
+
+ /*
+ * Although it would be possible to recover when certain types of corrupt frame are encountered,
+ * this could cause problems for clients as the payload may contain CQL messages from multiple
+ * streams. Simply dropping the corrupt frame or returning an error response would not give the
+ * client enough information to map back to inflight requests, leading to timeouts.
+ * Instead, we need to fail fast, possibly dropping the connection whenever a corrupt frame is
+ * encountered. Consequently, we terminate the connection (via a ProtocolException) whenever a
+ * corrupt frame is encountered, regardless of its type.
+ */
+ protected void processCorruptFrame(FrameDecoder.CorruptFrame frame)
+ {
+ corruptFramesUnrecovered++;
+ String error = String.format("%s invalid, unrecoverable CRC mismatch detected in frame %s. Read %d, Computed %d",
+ id(), frame.isRecoverable() ? "body" : "header", frame.readCRC, frame.computedCRC);
+
+ noSpamLogger.error(error);
+
+ // If this is part of a multi-frame message, process it before passing control to the error handler.
+ // This is so we can take care of any housekeeping associated with large messages.
+ if (!frame.isSelfContained)
+ {
+ if (null == largeMessage) // first frame of a large message
+ receivedBytes += frame.frameSize;
+ else // subsequent frame of a large message
+ processSubsequentFrameOfLargeMessage(frame);
+ }
+
+ handleError(new ProtocolException(error));
+ }
+
+ protected void fatalExceptionCaught(Throwable cause)
+ {
+ decoder.discard();
+ logger.warn("Unrecoverable exception caught in CQL message processing pipeline, closing the connection", cause);
+ channel.close();
+ }
+
+ static int envelopeSize(Envelope.Header header)
+ {
+ return Envelope.Header.LENGTH + Ints.checkedCast(header.bodySizeInBytes);
+ }
+
+ private class LargeMessage extends AbstractMessageHandler.LargeMessage
+ {
+ private static final long EXPIRES_AT = Long.MAX_VALUE;
+
+ private boolean overloaded = false;
+
+ private LargeMessage(Envelope.Header header)
+ {
+ super(envelopeSize(header), header, EXPIRES_AT, false);
+ }
+
+ private Envelope assembleFrame()
+ {
+ ByteBuf body = Unpooled.wrappedBuffer(buffers.stream()
+ .map(ShareableBytes::get)
+ .toArray(ByteBuffer[]::new));
+
+ body.readerIndex(Envelope.Header.LENGTH);
+ body.retain();
+ return new Envelope(header, body);
+ }
+
+ /**
+ * Used to indicate that a message should be dropped and not processed.
+ * We do this on receipt of the first frame of a large message if sufficient capacity
+ * cannot be acquired to process it and throwOnOverload is set for the connection.
+ * In this case, the client has elected to shed load rather than apply backpressure
+ * so we must ensure that subsequent frames are consumed from the channel. At that
+ * point an error response is returned to the client, rather than processing the message.
+ */
+ private void markOverloaded()
+ {
+ overloaded = true;
+ }
+
+ protected void onComplete()
+ {
+ if (overloaded)
+ handleErrorAndRelease(new OverloadedException("Server is in overloaded state. " +
+ "Cannot accept more requests at this point"), header);
+ else if (!isCorrupt)
+ processRequest(assembleFrame());
+
+ }
+
+ protected void abort()
+ {
+ if (!isCorrupt)
+ releaseBuffersAndCapacity(); // release resources if in normal state when abort() is invoked
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/transport/Client.java b/src/java/org/apache/cassandra/transport/Client.java
index 4f87cf41b6..76f710e5ee 100644
--- a/src/java/org/apache/cassandra/transport/Client.java
+++ b/src/java/org/apache/cassandra/transport/Client.java
@@ -33,13 +33,7 @@ import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.UTF8Type;
-import org.apache.cassandra.transport.frame.checksum.ChecksummingTransformer;
-import org.apache.cassandra.transport.frame.compress.CompressingTransformer;
-import org.apache.cassandra.transport.frame.compress.Compressor;
-import org.apache.cassandra.transport.frame.compress.LZ4Compressor;
-import org.apache.cassandra.transport.frame.compress.SnappyCompressor;
import org.apache.cassandra.transport.messages.*;
-import org.apache.cassandra.utils.ChecksumType;
import org.apache.cassandra.utils.Hex;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.MD5Digest;
@@ -111,56 +105,25 @@ public class Client extends SimpleClient
{
Map options = new HashMap();
options.put(StartupMessage.CQL_VERSION, "3.0.0");
- Compressor compressor = null;
- ChecksumType checksumType = null;
while (iter.hasNext())
{
- String next = iter.next().toLowerCase();
- switch (next)
+ String next = iter.next();
+ if (next.toLowerCase().equals("snappy"))
{
- case "snappy": {
- if (options.containsKey(StartupMessage.COMPRESSION))
- throw new RuntimeException("Multiple compression types supplied");
- options.put(StartupMessage.COMPRESSION, "snappy");
- compressor = SnappyCompressor.INSTANCE;
- break;
- }
- case "lz4": {
- if (options.containsKey(StartupMessage.COMPRESSION))
- throw new RuntimeException("Multiple compression types supplied");
- options.put(StartupMessage.COMPRESSION, "lz4");
- compressor = LZ4Compressor.INSTANCE;
- break;
- }
- case "crc32": {
- if (options.containsKey(StartupMessage.CHECKSUM))
- throw new RuntimeException("Multiple checksum types supplied");
- options.put(StartupMessage.CHECKSUM, ChecksumType.CRC32.name());
- checksumType = ChecksumType.CRC32;
- break;
- }
- case "adler32": {
- if (options.containsKey(StartupMessage.CHECKSUM))
- throw new RuntimeException("Multiple checksum types supplied");
- options.put(StartupMessage.CHECKSUM, ChecksumType.ADLER32.name());
- checksumType = ChecksumType.ADLER32;
- break;
- }
+ options.put(StartupMessage.COMPRESSION, "snappy");
+ connection.setCompressor(Compressor.SnappyCompressor.instance);
+ }
+ if (next.toLowerCase().equals("lz4"))
+ {
+ options.put(StartupMessage.COMPRESSION, "lz4");
+ connection.setCompressor(Compressor.LZ4Compressor.instance);
+ }
+ if (next.toLowerCase().equals("throw_on_overload"))
+ {
+ options.put(StartupMessage.THROW_ON_OVERLOAD, "1");
+ connection.setThrowOnOverload(true);
}
}
-
- if (checksumType == null)
- {
- if (compressor != null)
- {
- connection.setTransformer(CompressingTransformer.getTransformer(compressor));
- }
- }
- else
- {
- connection.setTransformer(ChecksummingTransformer.getTransformer(checksumType, compressor));
- }
-
return new StartupMessage(options);
}
else if (msgType.equals("QUERY"))
diff --git a/src/java/org/apache/cassandra/transport/ClientResourceLimits.java b/src/java/org/apache/cassandra/transport/ClientResourceLimits.java
new file mode 100644
index 0000000000..f9e7dab4ab
--- /dev/null
+++ b/src/java/org/apache/cassandra/transport/ClientResourceLimits.java
@@ -0,0 +1,271 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.transport;
+
+import java.net.InetAddress;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.codahale.metrics.Reservoir;
+import com.codahale.metrics.Snapshot;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.metrics.DecayingEstimatedHistogramReservoir;
+import org.apache.cassandra.net.AbstractMessageHandler;
+import org.apache.cassandra.net.ResourceLimits;
+
+public class ClientResourceLimits
+{
+ private static final Logger logger = LoggerFactory.getLogger(ClientResourceLimits.class);
+
+ private static final ResourceLimits.Concurrent GLOBAL_LIMIT = new ResourceLimits.Concurrent(getGlobalLimit());
+ private static final AbstractMessageHandler.WaitQueue GLOBAL_QUEUE = AbstractMessageHandler.WaitQueue.global(GLOBAL_LIMIT);
+ private static final ConcurrentMap PER_ENDPOINT_ALLOCATORS = new ConcurrentHashMap<>();
+
+ public static Allocator getAllocatorForEndpoint(InetAddress endpoint)
+ {
+ while (true)
+ {
+ Allocator result = PER_ENDPOINT_ALLOCATORS.computeIfAbsent(endpoint, Allocator::new);
+ if (result.acquire())
+ return result;
+
+ PER_ENDPOINT_ALLOCATORS.remove(endpoint, result);
+ }
+ }
+
+ public static long getGlobalLimit()
+ {
+ return DatabaseDescriptor.getNativeTransportMaxConcurrentRequestsInBytes();
+ }
+
+ public static void setGlobalLimit(long newLimit)
+ {
+ DatabaseDescriptor.setNativeTransportMaxConcurrentRequestsInBytes(newLimit);
+ long existingLimit = GLOBAL_LIMIT.setLimit(getGlobalLimit());
+ logger.info("Changed native_max_transport_requests_in_bytes from {} to {}", existingLimit, newLimit);
+ }
+
+ public static long getCurrentGlobalUsage()
+ {
+ return GLOBAL_LIMIT.using();
+ }
+
+ public static long getEndpointLimit()
+ {
+ return DatabaseDescriptor.getNativeTransportMaxConcurrentRequestsInBytesPerIp();
+ }
+
+ public static void setEndpointLimit(long newLimit)
+ {
+ long existingLimit = DatabaseDescriptor.getNativeTransportMaxConcurrentRequestsInBytesPerIp();
+ DatabaseDescriptor.setNativeTransportMaxConcurrentRequestsInBytesPerIp(newLimit); // ensure new instances get the new limit
+ for (Allocator allocator : PER_ENDPOINT_ALLOCATORS.values())
+ existingLimit = allocator.endpointAndGlobal.endpoint().setLimit(newLimit);
+ logger.info("Changed native_max_transport_requests_in_bytes_per_ip from {} to {}", existingLimit, newLimit);
+ }
+
+ public static Snapshot getCurrentIpUsage()
+ {
+ DecayingEstimatedHistogramReservoir histogram = new DecayingEstimatedHistogramReservoir();
+ for (Allocator allocator : PER_ENDPOINT_ALLOCATORS.values())
+ {
+ histogram.update(allocator.endpointAndGlobal.endpoint().using());
+ }
+ return histogram.getSnapshot();
+ }
+
+ /**
+ * This will recompute the ip usage histo on each query of the snapshot when requested instead of trying to keep
+ * a histogram up to date with each request
+ */
+ public static Reservoir ipUsageReservoir()
+ {
+ return new Reservoir()
+ {
+ public int size()
+ {
+ return PER_ENDPOINT_ALLOCATORS.size();
+ }
+
+ public void update(long l)
+ {
+ throw new IllegalStateException();
+ }
+
+ public Snapshot getSnapshot()
+ {
+ return getCurrentIpUsage();
+ }
+ };
+ }
+
+ /**
+ * Used during protocol negotiation in {@link InitialConnectionHandler} and then if protocol V4 or earlier is
+ * selected. V5 connections convert this to a {@link ResourceProvider} and use the global/endpoint
+ * limits and queues directly.
+ */
+ static class Allocator
+ {
+ private final AtomicInteger refCount = new AtomicInteger(0);
+ private final InetAddress endpoint;
+
+ private final ResourceLimits.EndpointAndGlobal endpointAndGlobal;
+ // This is not used or exposed by instances of this class, but it is used for V5+
+ // client connections so it's initialized here as the same queue must be shared
+ // across all clients from the same remote address.
+ private final AbstractMessageHandler.WaitQueue waitQueue;
+
+ private Allocator(InetAddress endpoint)
+ {
+ this.endpoint = endpoint;
+ ResourceLimits.Concurrent limit = new ResourceLimits.Concurrent(getEndpointLimit());
+ endpointAndGlobal = new ResourceLimits.EndpointAndGlobal(limit, GLOBAL_LIMIT);
+ waitQueue = AbstractMessageHandler.WaitQueue.endpoint(limit);
+ }
+
+ private boolean acquire()
+ {
+ return 0 < refCount.updateAndGet(i -> i < 0 ? i : i + 1);
+ }
+
+ /**
+ * Decrement the reference count, possibly removing the instance from the cache
+ * if this is its final reference
+ */
+ void release()
+ {
+ if (-1 == refCount.updateAndGet(i -> i == 1 ? -1 : i - 1))
+ PER_ENDPOINT_ALLOCATORS.remove(endpoint, this);
+ }
+
+ /**
+ * Attempt to allocate a number of permits representing bytes towards the inflight
+ * limits. To succeed, it must be possible to allocate from both the per-endpoint
+ * and global reserves.
+ *
+ * @param amount number permits to allocate
+ * @return outcome SUCCESS if the allocation was successful. In the case of failure,
+ * either INSUFFICIENT_GLOBAL or INSUFFICIENT_ENPOINT to indicate which reserve rejected
+ * the allocation request.
+ */
+ ResourceLimits.Outcome tryAllocate(long amount)
+ {
+ return endpointAndGlobal.tryAllocate(amount);
+ }
+
+ /**
+ * Force an allocation of a number of permits representing bytes from the inflight
+ * limits. Permits will be acquired from both the per-endpoint and global reserves
+ * which may lead to either or both reserves going over their limits.
+ * @param amount number permits to allocate
+ */
+ void allocate(long amount)
+ {
+ endpointAndGlobal.allocate(amount);
+ }
+
+ /**
+ * Release a number of permits representing bytes back to the both the per-endpoint and
+ * global limits for inflight requests.
+ *
+ * @param amount number of permits to release
+ * @return outcome, ABOVE_LIMIT if either reserve is above its configured limit after
+ * the operation completes or, BELOW_LIMIT if neither is.
+ * rejected the allocation request.
+ */
+ ResourceLimits.Outcome release(long amount)
+ {
+ return endpointAndGlobal.release(amount);
+ }
+
+ @VisibleForTesting
+ long endpointUsing()
+ {
+ return endpointAndGlobal.endpoint().using();
+ }
+
+ @VisibleForTesting
+ long globallyUsing()
+ {
+ return endpointAndGlobal.global().using();
+ }
+
+ public String toString()
+ {
+ return String.format("InflightEndpointRequestPayload: %d, InflightOverallRequestPayload: %d",
+ endpointAndGlobal.endpoint().using(), endpointAndGlobal.global().using());
+ }
+ }
+
+ /**
+ * Used in protocol V5 and later by the AbstractMessageHandler/CQLMessageHandler hierarchy.
+ * This hides the allocate/tryAllocate/release methods from EndpointResourceLimits and exposes
+ * the endpoint and global limits, along with their corresponding
+ * {@link org.apache.cassandra.net.AbstractMessageHandler.WaitQueue} directly.
+ * Provided as an interface and single implementation for testing (see CQLConnectionTest)
+ */
+ interface ResourceProvider
+ {
+ ResourceLimits.Limit globalLimit();
+ AbstractMessageHandler.WaitQueue globalWaitQueue();
+ ResourceLimits.Limit endpointLimit();
+ AbstractMessageHandler.WaitQueue endpointWaitQueue();
+ void release();
+
+ static class Default implements ResourceProvider
+ {
+ private final Allocator limits;
+
+ Default(Allocator limits)
+ {
+ this.limits = limits;
+ }
+
+ public ResourceLimits.Limit globalLimit()
+ {
+ return limits.endpointAndGlobal.global();
+ }
+
+ public AbstractMessageHandler.WaitQueue globalWaitQueue()
+ {
+ return GLOBAL_QUEUE;
+ }
+
+ public ResourceLimits.Limit endpointLimit()
+ {
+ return limits.endpointAndGlobal.endpoint();
+ }
+
+ public AbstractMessageHandler.WaitQueue endpointWaitQueue()
+ {
+ return limits.waitQueue;
+ }
+
+ public void release()
+ {
+ limits.release();
+ }
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/transport/Compressor.java b/src/java/org/apache/cassandra/transport/Compressor.java
new file mode 100644
index 0000000000..ae28e7ac16
--- /dev/null
+++ b/src/java/org/apache/cassandra/transport/Compressor.java
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.transport;
+
+import java.io.IOException;
+
+import io.netty.buffer.ByteBuf;
+import org.xerial.snappy.Snappy;
+import org.xerial.snappy.SnappyError;
+
+import net.jpountz.lz4.LZ4Factory;
+
+import org.apache.cassandra.utils.JVMStabilityInspector;
+
+public interface Compressor
+{
+ public Envelope compress(Envelope uncompressed) throws IOException;
+ public Envelope decompress(Envelope compressed) throws IOException;
+
+ /*
+ * TODO: We can probably do more efficient, like by avoiding copy.
+ * Also, we don't reuse ICompressor because the API doesn't expose enough.
+ */
+ public static class SnappyCompressor implements Compressor
+ {
+ public static final SnappyCompressor instance;
+ static
+ {
+ SnappyCompressor i;
+ try
+ {
+ i = new SnappyCompressor();
+ }
+ catch (Exception e)
+ {
+ JVMStabilityInspector.inspectThrowable(e);
+ i = null;
+ }
+ catch (NoClassDefFoundError | SnappyError | UnsatisfiedLinkError e)
+ {
+ i = null;
+ }
+ instance = i;
+ }
+
+ private SnappyCompressor()
+ {
+ // this would throw java.lang.NoClassDefFoundError if Snappy class
+ // wasn't found at runtime which should be processed by the calling method
+ Snappy.getNativeLibraryVersion();
+ }
+
+ public Envelope compress(Envelope uncompressed) throws IOException
+ {
+ byte[] input = CBUtil.readRawBytes(uncompressed.body);
+ ByteBuf output = CBUtil.allocator.heapBuffer(Snappy.maxCompressedLength(input.length));
+
+ try
+ {
+ int written = Snappy.compress(input, 0, input.length, output.array(), output.arrayOffset());
+ output.writerIndex(written);
+ }
+ catch (final Throwable e)
+ {
+ output.release();
+ throw e;
+ }
+ finally
+ {
+ uncompressed.release();
+ }
+
+ return uncompressed.with(output);
+ }
+
+ public Envelope decompress(Envelope compressed) throws IOException
+ {
+ byte[] input = CBUtil.readRawBytes(compressed.body);
+
+ if (!Snappy.isValidCompressedBuffer(input, 0, input.length))
+ throw new ProtocolException("Provided frame does not appear to be Snappy compressed");
+
+ ByteBuf output = CBUtil.allocator.heapBuffer(Snappy.uncompressedLength(input));
+
+ try
+ {
+ int size = Snappy.uncompress(input, 0, input.length, output.array(), output.arrayOffset());
+ output.writerIndex(size);
+ }
+ catch (final Throwable e)
+ {
+ output.release();
+ throw e;
+ }
+ finally
+ {
+ compressed.release();
+ }
+
+ return compressed.with(output);
+ }
+ }
+
+ /*
+ * This is very close to the ICompressor implementation, and in particular
+ * it also layout the uncompressed size at the beginning of the message to
+ * make uncompression faster, but contrarly to the ICompressor, that length
+ * is written in big-endian. The native protocol is entirely big-endian, so
+ * it feels like putting little-endian here would be a annoying trap for
+ * client writer.
+ */
+ public static class LZ4Compressor implements Compressor
+ {
+ public static final LZ4Compressor instance = new LZ4Compressor();
+
+ private static final int INTEGER_BYTES = 4;
+ private final net.jpountz.lz4.LZ4Compressor compressor;
+ private final net.jpountz.lz4.LZ4Decompressor decompressor;
+
+ private LZ4Compressor()
+ {
+ final LZ4Factory lz4Factory = LZ4Factory.fastestInstance();
+ compressor = lz4Factory.fastCompressor();
+ decompressor = lz4Factory.decompressor();
+ }
+
+ public Envelope compress(Envelope uncompressed)
+ {
+ byte[] input = CBUtil.readRawBytes(uncompressed.body);
+
+ int maxCompressedLength = compressor.maxCompressedLength(input.length);
+ ByteBuf outputBuf = CBUtil.allocator.heapBuffer(INTEGER_BYTES + maxCompressedLength);
+
+ byte[] output = outputBuf.array();
+ int outputOffset = outputBuf.arrayOffset();
+
+ output[outputOffset ] = (byte) (input.length >>> 24);
+ output[outputOffset + 1] = (byte) (input.length >>> 16);
+ output[outputOffset + 2] = (byte) (input.length >>> 8);
+ output[outputOffset + 3] = (byte) (input.length);
+
+ try
+ {
+ int written = compressor.compress(input, 0, input.length, output, outputOffset + INTEGER_BYTES, maxCompressedLength);
+ outputBuf.writerIndex(INTEGER_BYTES + written);
+
+ return uncompressed.with(outputBuf);
+ }
+ catch (final Throwable e)
+ {
+ outputBuf.release();
+ throw e;
+ }
+ finally
+ {
+ uncompressed.release();
+ }
+ }
+
+ public Envelope decompress(Envelope compressed) throws IOException
+ {
+ byte[] input = CBUtil.readRawBytes(compressed.body);
+
+ int uncompressedLength = ((input[0] & 0xFF) << 24)
+ | ((input[1] & 0xFF) << 16)
+ | ((input[2] & 0xFF) << 8)
+ | ((input[3] & 0xFF));
+
+ ByteBuf output = CBUtil.allocator.heapBuffer(uncompressedLength);
+
+ try
+ {
+ int read = decompressor.decompress(input, INTEGER_BYTES, output.array(), output.arrayOffset(), uncompressedLength);
+ if (read != input.length - INTEGER_BYTES)
+ throw new IOException("Compressed lengths mismatch");
+
+ output.writerIndex(uncompressedLength);
+
+ return compressed.with(output);
+ }
+ catch (final Throwable e)
+ {
+ output.release();
+ throw e;
+ }
+ finally
+ {
+ //release the old message
+ compressed.release();
+ }
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/transport/Connection.java b/src/java/org/apache/cassandra/transport/Connection.java
index b7f5b170ff..cae10c2825 100644
--- a/src/java/org/apache/cassandra/transport/Connection.java
+++ b/src/java/org/apache/cassandra/transport/Connection.java
@@ -19,7 +19,6 @@ package org.apache.cassandra.transport;
import io.netty.channel.Channel;
import io.netty.util.AttributeKey;
-import org.apache.cassandra.transport.frame.FrameBodyTransformer;
public class Connection
{
@@ -29,8 +28,8 @@ public class Connection
private final ProtocolVersion version;
private final Tracker tracker;
- private volatile FrameBodyTransformer transformer;
private boolean throwOnOverload;
+ private volatile Compressor preV5MessageCompressor;
public Connection(Channel channel, ProtocolVersion version, Tracker tracker)
{
@@ -41,14 +40,14 @@ public class Connection
tracker.addConnection(channel, this);
}
- public void setTransformer(FrameBodyTransformer transformer)
+ public void setCompressor(Compressor compressor)
{
- this.transformer = transformer;
+ this.preV5MessageCompressor = compressor;
}
- public FrameBodyTransformer getTransformer()
+ public Compressor getCompressor()
{
- return transformer;
+ return preV5MessageCompressor;
}
public void setThrowOnOverload(boolean throwOnOverload)
diff --git a/src/java/org/apache/cassandra/transport/Dispatcher.java b/src/java/org/apache/cassandra/transport/Dispatcher.java
new file mode 100644
index 0000000000..65093f339a
--- /dev/null
+++ b/src/java/org/apache/cassandra/transport/Dispatcher.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.transport;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import io.netty.channel.Channel;
+import io.netty.channel.EventLoop;
+import org.apache.cassandra.concurrent.LocalAwareExecutorService;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.service.ClientWarn;
+import org.apache.cassandra.service.QueryState;
+import org.apache.cassandra.transport.Flusher.FlushItem;
+import org.apache.cassandra.transport.messages.ErrorMessage;
+import org.apache.cassandra.utils.JVMStabilityInspector;
+
+import static org.apache.cassandra.concurrent.SharedExecutorPool.SHARED;
+
+public class Dispatcher
+{
+ private static final LocalAwareExecutorService requestExecutor = SHARED.newExecutor(DatabaseDescriptor.getNativeTransportMaxThreads(),
+ DatabaseDescriptor::setNativeTransportMaxThreads,
+ "transport",
+ "Native-Transport-Requests");
+
+ private static final ConcurrentMap flusherLookup = new ConcurrentHashMap<>();
+ private final boolean useLegacyFlusher;
+
+ /**
+ * Takes a Channel, Request and the Response produced by processRequest and outputs a FlushItem
+ * appropriate for the pipeline, which is specific to the protocol version. V5 and above will
+ * produce FlushItem.Framed instances whereas earlier versions require FlushItem.Unframed.
+ * The instances of these FlushItem subclasses are specialized to release resources in the
+ * right way for the specific pipeline that produced them.
+ */
+ // TODO parameterize with FlushItem subclass
+ interface FlushItemConverter
+ {
+ FlushItem> toFlushItem(Channel channel, Message.Request request, Message.Response response);
+ }
+
+ public Dispatcher(boolean useLegacyFlusher)
+ {
+ this.useLegacyFlusher = useLegacyFlusher;
+ }
+
+ public void dispatch(Channel channel, Message.Request request, FlushItemConverter forFlusher) {
+ requestExecutor.submit(() -> processRequest(channel, request, forFlusher));
+ }
+
+ /**
+ * Note: this method may be executed on the netty event loop, during initial protocol negotiation
+ */
+ static Message.Response processRequest(ServerConnection connection, Message.Request request)
+ {
+ long queryStartNanoTime = System.nanoTime();
+ if (connection.getVersion().isGreaterOrEqualTo(ProtocolVersion.V4))
+ ClientWarn.instance.captureWarnings();
+
+ QueryState qstate = connection.validateNewMessage(request.type, connection.getVersion());
+
+ Message.logger.trace("Received: {}, v={}", request, connection.getVersion());
+ connection.requests.inc();
+ Message.Response response = request.execute(qstate, queryStartNanoTime);
+ response.setStreamId(request.getStreamId());
+ response.setWarnings(ClientWarn.instance.getWarnings());
+ response.attach(connection);
+ connection.applyStateTransition(request.type, response.type);
+ return response;
+ }
+
+ /**
+ * Note: this method is not expected to execute on the netty event loop.
+ */
+ void processRequest(Channel channel, Message.Request request, FlushItemConverter forFlusher)
+ {
+ final Message.Response response;
+ final ServerConnection connection;
+ FlushItem> toFlush;
+ try
+ {
+ assert request.connection() instanceof ServerConnection;
+ connection = (ServerConnection) request.connection();
+ response = processRequest(connection, request);
+ toFlush = forFlusher.toFlushItem(channel, request, response);
+ Message.logger.trace("Responding: {}, v={}", response, connection.getVersion());
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ ExceptionHandlers.UnexpectedChannelExceptionHandler handler = new ExceptionHandlers.UnexpectedChannelExceptionHandler(channel, true);
+ ErrorMessage error = ErrorMessage.fromException(t, handler);
+ error.setStreamId(request.getStreamId());
+ toFlush = forFlusher.toFlushItem(channel, request, error);
+ }
+ finally
+ {
+ ClientWarn.instance.resetWarnings();
+ }
+ flush(toFlush);
+ }
+
+ private void flush(FlushItem> item)
+ {
+ EventLoop loop = item.channel.eventLoop();
+ Flusher flusher = flusherLookup.get(loop);
+ if (flusher == null)
+ {
+ Flusher created = useLegacyFlusher ? Flusher.legacy(loop) : Flusher.immediate(loop);
+ Flusher alt = flusherLookup.putIfAbsent(loop, flusher = created);
+ if (alt != null)
+ flusher = alt;
+ }
+
+ flusher.enqueue(item);
+ flusher.start();
+ }
+
+ public static void shutdown()
+ {
+ if (requestExecutor != null)
+ {
+ requestExecutor.shutdown();
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/transport/Frame.java b/src/java/org/apache/cassandra/transport/Envelope.java
similarity index 52%
rename from src/java/org/apache/cassandra/transport/Frame.java
rename to src/java/org/apache/cassandra/transport/Envelope.java
index ec4317ab06..e78862406a 100644
--- a/src/java/org/apache/cassandra/transport/Frame.java
+++ b/src/java/org/apache/cassandra/transport/Envelope.java
@@ -19,24 +19,26 @@
package org.apache.cassandra.transport;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.EnumSet;
import java.util.List;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.codec.MessageToMessageEncoder;
-import io.netty.util.Attribute;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.InvalidRequestException;
-import org.apache.cassandra.metrics.ClientRequestSizeMetrics;
-import org.apache.cassandra.transport.frame.FrameBodyTransformer;
+import org.apache.cassandra.metrics.ClientMessageSizeMetrics;
import org.apache.cassandra.transport.messages.ErrorMessage;
+import org.apache.cassandra.utils.ByteBufferUtil;
-public class Frame
+public class Envelope
{
public static final byte PROTOCOL_VERSION_MASK = 0x7f;
@@ -44,7 +46,7 @@ public class Frame
public final ByteBuf body;
/**
- * An on-wire frame consists of a header and a body.
+ * An on-wire message envelope consists of a header and a body.
*
* The header is defined the following way in native protocol version 3 and later:
*
@@ -55,7 +57,7 @@ public class Frame
* | length |
* +---------+---------+---------+---------+
*/
- private Frame(Header header, ByteBuf body)
+ public Envelope(Header header, ByteBuf body)
{
this.header = header;
this.body = body;
@@ -71,10 +73,36 @@ public class Frame
return body.release();
}
- public static Frame create(Message.Type type, int streamId, ProtocolVersion version, EnumSet flags, ByteBuf body)
+ @VisibleForTesting
+ public Envelope clone()
+ {
+ return new Envelope(header, Unpooled.wrappedBuffer(ByteBufferUtil.clone(body.nioBuffer())));
+ }
+
+ public static Envelope create(Message.Type type, int streamId, ProtocolVersion version, EnumSet flags, ByteBuf body)
{
Header header = new Header(version, flags, streamId, type, body.readableBytes());
- return new Frame(header, body);
+ return new Envelope(header, body);
+ }
+
+ public void encodeHeaderInto(ByteBuffer buf)
+ {
+ buf.put((byte) header.type.direction.addToVersion(header.version.asInt()));
+ buf.put((byte) Envelope.Header.Flag.serialize(header.flags));
+
+ if (header.version.isGreaterOrEqualTo(ProtocolVersion.V3))
+ buf.putShort((short) header.streamId);
+ else
+ buf.put((byte) header.streamId);
+
+ buf.put((byte) header.type.opcode);
+ buf.putInt(body.readableBytes());
+ }
+
+ public void encodeInto(ByteBuffer buf)
+ {
+ encodeHeaderInto(buf);
+ buf.put(body.nioBuffer());
}
public static class Header
@@ -106,8 +134,7 @@ public class Frame
TRACING,
CUSTOM_PAYLOAD,
WARNING,
- USE_BETA,
- CHECKSUMMED;
+ USE_BETA;
private static final Flag[] ALL_VALUES = values();
@@ -132,32 +159,70 @@ public class Frame
}
}
- public Frame with(ByteBuf newBody)
+ public Envelope with(ByteBuf newBody)
{
- return new Frame(header, newBody);
+ return new Envelope(header, newBody);
}
public static class Decoder extends ByteToMessageDecoder
{
- private static final int MAX_FRAME_LENGTH = DatabaseDescriptor.getNativeTransportMaxFrameSize();
+ private static final int MAX_TOTAL_LENGTH = DatabaseDescriptor.getNativeTransportMaxFrameSize();
- private boolean discardingTooLongFrame;
- private long tooLongFrameLength;
+ private boolean discardingTooLongMessage;
+ private long tooLongTotalLength;
private long bytesToDiscard;
private int tooLongStreamId;
- private final Connection.Factory factory;
-
- public Decoder(Connection.Factory factory)
+ /**
+ * Used by protocol V5 and later to extract a CQL message header from the buffer containing
+ * it, without modifying the position of the underlying buffer. This essentially mirrors the
+ * pre-V5 code in {@link org.apache.cassandra.transport.Envelope.Decoder#decode(ByteBuf)},
+ * with two differences:
+ * The input is a ByteBuffer rather than a ByteBuf
+ * This cannot return null, as V5 always deals with entire CQL messages. Coalescing of bytes
+ * off the wire happens at the layer below, in {@link org.apache.cassandra.net.FrameDecoder}
+ *
+ * @param buffer ByteBuffer containing the message envelope
+ * @return CQL Message header
+ */
+ Envelope.Header extractHeader(ByteBuffer buffer)
{
- this.factory = factory;
+ Preconditions.checkArgument(buffer.remaining() >= Header.LENGTH,
+ "Undersized buffer supplied. Expected %s, actual %s",
+ Header.LENGTH,
+ buffer.remaining());
+
+ int idx = buffer.position();
+ int firstByte = buffer.get(idx++);
+ Message.Direction direction = Message.Direction.extractFromVersion(firstByte);
+ int versionNum = firstByte & PROTOCOL_VERSION_MASK;
+ ProtocolVersion version = ProtocolVersion.decode(versionNum, DatabaseDescriptor.getNativeTransportAllowOlderProtocols());
+
+ int flags = buffer.get(idx++);
+ EnumSet decodedFlags = decodeFlags(version, flags);
+
+ int streamId = buffer.getShort(idx);
+ idx += 2;
+
+ // This throws a protocol exceptions if the opcode is unknown
+ Message.Type type;
+ try
+ {
+ type = Message.Type.fromOpcode(buffer.get(idx++), direction);
+ }
+ catch (ProtocolException e)
+ {
+ throw ErrorMessage.wrap(e, streamId);
+ }
+
+ long bodyLength = buffer.getInt(idx);
+ return new Header(version, decodedFlags, streamId, type, bodyLength);
}
@VisibleForTesting
- Frame decodeFrame(ByteBuf buffer)
- throws Exception
+ Envelope decode(ByteBuf buffer)
{
- if (discardingTooLongFrame)
+ if (discardingTooLongMessage)
{
bytesToDiscard = discard(buffer, bytesToDiscard);
// If we have discarded everything, throw the exception
@@ -184,11 +249,7 @@ public class Frame
return null;
int flags = buffer.getByte(idx++);
- EnumSet decodedFlags = Header.Flag.deserialize(flags);
-
- if (version.isBeta() && !decodedFlags.contains(Header.Flag.USE_BETA))
- throw new ProtocolException(String.format("Beta version of the protocol used (%s), but USE_BETA flag is unset", version),
- version);
+ EnumSet decodedFlags = decodeFlags(version, flags);
int streamId = buffer.getShort(idx);
idx += 2;
@@ -207,24 +268,24 @@ public class Frame
long bodyLength = buffer.getUnsignedInt(idx);
idx += Header.BODY_LENGTH_SIZE;
- long frameLength = bodyLength + Header.LENGTH;
- if (frameLength > MAX_FRAME_LENGTH)
+ long totalLength = bodyLength + Header.LENGTH;
+ if (totalLength > MAX_TOTAL_LENGTH)
{
// Enter the discard mode and discard everything received so far.
- discardingTooLongFrame = true;
+ discardingTooLongMessage = true;
tooLongStreamId = streamId;
- tooLongFrameLength = frameLength;
- bytesToDiscard = discard(buffer, frameLength);
+ tooLongTotalLength = totalLength;
+ bytesToDiscard = discard(buffer, totalLength);
if (bytesToDiscard <= 0)
fail();
return null;
}
- if (buffer.readableBytes() < frameLength)
+ if (buffer.readableBytes() < totalLength)
return null;
- ClientRequestSizeMetrics.totalBytesRead.inc(frameLength);
- ClientRequestSizeMetrics.bytesReceivedPerFrame.update(frameLength);
+ ClientMessageSizeMetrics.bytesReceived.inc(totalLength);
+ ClientMessageSizeMetrics.bytesReceivedPerRequest.update(totalLength);
// extract body
ByteBuf body = buffer.slice(idx, (int) bodyLength);
@@ -233,42 +294,36 @@ public class Frame
idx += bodyLength;
buffer.readerIndex(idx);
- return new Frame(new Header(version, decodedFlags, streamId, type, bodyLength), body);
+ return new Envelope(new Header(version, decodedFlags, streamId, type, bodyLength), body);
+ }
+
+ private EnumSet decodeFlags(ProtocolVersion version, int flags)
+ {
+ EnumSet decodedFlags = Header.Flag.deserialize(flags);
+
+ if (version.isBeta() && !decodedFlags.contains(Header.Flag.USE_BETA))
+ throw new ProtocolException(String.format("Beta version of the protocol used (%s), but USE_BETA flag is unset", version),
+ version);
+ return decodedFlags;
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List