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

Request-Based Native Transport Rate-Limiting 

patch by Caleb Rackliffe; reviewed by Benedict Elliott Smith and Josh McKenzie for CASSANDRA-16663
This commit is contained in:
Caleb Rackliffe 2021-08-19 10:55:58 -05:00
parent b286639eea
commit d220d24994
25 changed files with 1329 additions and 264 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.1
* Request-Based Native Transport Rate-Limiting (CASSANDRA-16663)
* Implement nodetool getauditlog command (CASSANDRA-16725)
* Clean up repair code (CASSANDRA-13720)
* Background schedule to clean up orphaned hints files (CASSANDRA-16815)

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

@ -34,8 +34,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.audit.AuditLogOptions;
import org.apache.cassandra.fql.FullQueryLoggerOptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.fql.FullQueryLoggerOptions;
/**
* A class that contains configuration properties for the cassandra node it runs within.
@ -193,6 +193,8 @@ public class Config
public volatile boolean native_transport_allow_older_protocols = true;
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 volatile boolean native_transport_rate_limiting_enabled = false;
public volatile int native_transport_max_requests_per_second = 1000000;
public int native_transport_receive_queue_capacity_in_bytes = 1 << 20; // 1MiB
@Deprecated

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

@ -540,6 +540,11 @@ public class DatabaseDescriptor
{
conf.native_transport_max_concurrent_requests_in_bytes_per_ip = Runtime.getRuntime().maxMemory() / 40;
}
if (conf.native_transport_rate_limiting_enabled)
logger.info("Native transport rate-limiting enabled at {} requests/second.", conf.native_transport_max_requests_per_second);
else
logger.info("Native transport rate-limiting disabled.");
if (conf.commitlog_total_space_in_mb == null)
{
@ -2324,6 +2329,28 @@ public class DatabaseDescriptor
conf.native_transport_max_concurrent_requests_in_bytes = maxConcurrentRequestsInBytes;
}
public static int getNativeTransportMaxRequestsPerSecond()
{
return conf.native_transport_max_requests_per_second;
}
public static void setNativeTransportMaxRequestsPerSecond(int perSecond)
{
Preconditions.checkArgument(perSecond > 0, "native_transport_max_requests_per_second must be greater than zero");
conf.native_transport_max_requests_per_second = perSecond;
}
public static void setNativeTransportRateLimitingEnabled(boolean enabled)
{
logger.info("native_transport_rate_limiting_enabled set to {}", enabled);
conf.native_transport_rate_limiting_enabled = enabled;
}
public static boolean getNativeTransportRateLimitingEnabled()
{
return conf.native_transport_rate_limiting_enabled;
}
public static int getCommitLogSyncPeriod()
{
return conf.commitlog_sync_period_in_ms;

21
src/java/org/apache/cassandra/net/AbstractMessageHandler.java
View File

@ -22,7 +22,6 @@ 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;
@ -34,13 +33,13 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.EventLoop;
import org.apache.cassandra.metrics.ClientMetrics;
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;
@ -98,8 +97,8 @@ import static org.apache.cassandra.utils.MonotonicClock.approxTime;
* 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.
* of the large-message accumulating state-machine, and {@link InboundMessageHandler.ProcessMessage} and its inheritors
* for the differences in execution.
*
* # Flow control (backpressure)
*
@ -135,8 +134,7 @@ import static org.apache.cassandra.utils.MonotonicClock.approxTime;
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;
@ -304,12 +302,17 @@ public abstract class AbstractMessageHandler extends ChannelInboundHandlerAdapte
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
* Process up to one message using supplied overridden reserves - one of them pre-allocated,
* and guaranteed to be enough for one message - then, if no obstacles encountered, reactivate
* the frame decoder using normal reserve capacities.
*/
if (processUpToOneMessage(endpointReserve, globalReserve))
{
decoder.reactivate();
if (decoder.isActive())
ClientMetrics.instance.unpauseConnection();
}
}
catch (Throwable t)
{
@ -321,7 +324,7 @@ public abstract class AbstractMessageHandler extends ChannelInboundHandlerAdapte
// 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
protected boolean processUpToOneMessage(Limit endpointReserve, Limit globalReserve) throws IOException
{
UpToOneMessageFrameProcessor processor = new UpToOneMessageFrameProcessor(endpointReserve, globalReserve);
decoder.processBacklog(processor);

13
src/java/org/apache/cassandra/net/FrameDecoder.java
View File

@ -190,6 +190,14 @@ public abstract class FrameDecoder extends ChannelInboundHandlerAdapter
abstract void decode(Collection<Frame> into, ShareableBytes bytes);
abstract void addLastTo(ChannelPipeline pipeline);
/**
* @return true if we are actively decoding and processing frames
*/
public boolean isActive()
{
return isActive;
}
/**
* For use by InboundMessageHandler (or other upstream handlers) that want to start receiving frames.
*/
@ -208,7 +216,7 @@ public abstract class FrameDecoder extends ChannelInboundHandlerAdapter
* For use by InboundMessageHandler (or other upstream handlers) that want to resume
* receiving frames after previously indicating that processing should be paused.
*/
void reactivate() throws IOException
public void reactivate() throws IOException
{
if (isActive)
throw new IllegalStateException("Tried to reactivate an already active FrameDecoder");
@ -282,7 +290,8 @@ public abstract class FrameDecoder extends ChannelInboundHandlerAdapter
{
decode(frames, bytes);
if (isActive) isActive = deliver(processor);
if (isActive)
isActive = deliver(processor);
}
@Override

2
src/java/org/apache/cassandra/net/InboundMessageHandler.java
View File

@ -34,8 +34,6 @@ import org.apache.cassandra.exceptions.IncompatibleSchemaException;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.Message.Header;
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.FrameDecoder.CorruptFrame;
import org.apache.cassandra.net.ResourceLimits.Limit;

4
src/java/org/apache/cassandra/net/ResourceLimits.java
View File

@ -41,7 +41,7 @@ public abstract class ResourceLimits
* Sets the total amount of permits represented by this {@link Limit} - the capacity
*
* If the old limit has been reached and the new limit is large enough to allow for more
* permits to be aqcuired, subsequent calls to {@link #allocate(long)} or {@link #tryAllocate(long)}
* permits to be acquired, subsequent calls to {@link #allocate(long)} or {@link #tryAllocate(long)}
* will succeed.
*
* If the new limit is lower than the current amount of allocated permits then subsequent calls
@ -163,7 +163,7 @@ public abstract class ResourceLimits
// possible it would require synchronizing the closing of all outbound connections
// and reinitializing the Concurrent limit before reopening. For such an unlikely path
// (previously this was an assert), it is safer to terminate the JVM and have something external
// restart and get back to a known good state rather than intermittendly crashing on any of
// restart and get back to a known good state rather than intermittently crashing on any of
// the connections sharing this limit.
throw new UnrecoverableIllegalStateException(
"Internode messaging byte limits that are shared between connections is invalid (using="+using+")");

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

@ -5875,6 +5875,30 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
ClientResourceLimits.setEndpointLimit(newLimit);
}
@Override
public int getNativeTransportMaxRequestsPerSecond()
{
return ClientResourceLimits.getNativeTransportMaxRequestsPerSecond();
}
@Override
public void setNativeTransportMaxRequestsPerSecond(int newPerSecond)
{
ClientResourceLimits.setNativeTransportMaxRequestsPerSecond(newPerSecond);
}
@Override
public void setNativeTransportRateLimitingEnabled(boolean enabled)
{
DatabaseDescriptor.setNativeTransportRateLimitingEnabled(enabled);
}
@Override
public boolean getNativeTransportRateLimitingEnabled()
{
return DatabaseDescriptor.getNativeTransportRateLimitingEnabled();
}
@VisibleForTesting
public void shutdownServer()
{

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

@ -35,7 +35,6 @@ import javax.management.openmbean.TabularData;
import org.apache.cassandra.db.ColumnFamilyStoreMBean;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.locator.InetAddressAndPort;
public interface StorageServiceMBean extends NotificationEmitter
{
@ -556,7 +555,10 @@ public interface StorageServiceMBean extends NotificationEmitter
public void setNativeTransportMaxConcurrentRequestsInBytes(long newLimit);
public long getNativeTransportMaxConcurrentRequestsInBytesPerIp();
public void setNativeTransportMaxConcurrentRequestsInBytesPerIp(long newLimit);
public int getNativeTransportMaxRequestsPerSecond();
public void setNativeTransportMaxRequestsPerSecond(int newPerSecond);
public void setNativeTransportRateLimitingEnabled(boolean enabled);
public boolean getNativeTransportRateLimitingEnabled();
// allows a node that have been started without joining the ring to join it
public void joinRing() throws IOException;

252
src/java/org/apache/cassandra/transport/CQLMessageHandler.java
View File

@ -23,6 +23,8 @@ import java.nio.ByteBuffer;
import java.util.concurrent.TimeUnit;
import com.google.common.primitives.Ints;
import org.apache.cassandra.transport.ClientResourceLimits.Overload;
import org.apache.cassandra.utils.concurrent.NonBlockingRateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -42,7 +44,6 @@ 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;
@ -78,6 +79,7 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.SECONDS);
public static final int LARGE_MESSAGE_THRESHOLD = FrameEncoder.Payload.MAX_SIZE - 1;
public static final TimeUnit RATE_LIMITER_DELAY_UNIT = TimeUnit.NANOSECONDS;
private final Envelope.Decoder envelopeDecoder;
private final Message.Decoder<M> messageDecoder;
@ -86,13 +88,14 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
private final ErrorHandler errorHandler;
private final boolean throwOnOverload;
private final ProtocolVersion version;
private final NonBlockingRateLimiter requestRateLimiter;
long channelPayloadBytesInFlight;
private int consecutiveMessageErrors = 0;
interface MessageConsumer<M extends Message>
{
void accept(Channel channel, M message, Dispatcher.FlushItemConverter toFlushItem);
void accept(Channel channel, M message, Dispatcher.FlushItemConverter toFlushItem, Overload backpressure);
}
interface ErrorHandler
@ -129,6 +132,7 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
this.errorHandler = errorHandler;
this.throwOnOverload = throwOnOverload;
this.version = version;
this.requestRateLimiter = resources.requestRateLimiter();
}
@Override
@ -139,6 +143,23 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
return super.process(frame);
}
/**
* Checks limits on bytes in flight and the request rate limiter (if enabled), then takes one of three actions:
*
* 1.) If no limits are breached, process the request.
* 2.) If a limit is breached, and the connection is configured to throw on overload, throw {@link OverloadedException}.
* 3.) If a limit is breached, and the connection is not configurd to throw, process the request, and return false
* to let the {@link FrameDecoder} know it should stop processing frames.
*
* If the connection is configured to throw {@link OverloadedException}, requests that breach the rate limit are
* not counted against that limit.
*
* @return true if the {@link FrameDecoder} should continue to process incoming frames, and false if it should stop
* processing them, effectively applying backpressure to clients
*
* @throws ErrorMessage.WrappedException with an {@link OverloadedException} if overload occurs and the
* connection is configured to throw on overload
*/
protected boolean processOneContainedMessage(ShareableBytes bytes, Limit endpointReserve, Limit globalReserve)
{
ByteBuffer buf = bytes.get();
@ -157,41 +178,108 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
// 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);
discardAndThrow(endpointReserve, globalReserve, buf, header, messageSize, Overload.BYTES_IN_FLIGHT);
return true;
}
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);
buf.position(buf.position() + Envelope.Header.LENGTH + messageSize);
if (DatabaseDescriptor.getNativeTransportRateLimitingEnabled() && !requestRateLimiter.tryReserve())
{
// We've already allocated against the bytes-in-flight limits, so release those resources.
release(header);
discardAndThrow(endpointReserve, globalReserve, buf, header, messageSize, Overload.REQUESTS);
return true;
}
}
else if (!acquireCapacityAndQueueOnFailure(header, endpointReserve, globalReserve))
else
{
// set backpressure on the channel, queuing the request until we have capacity
ClientMetrics.instance.pauseConnection();
return false;
Overload backpressure = Overload.NONE;
if (!acquireCapacityAndQueueOnFailure(header, endpointReserve, globalReserve))
{
if (processRequestAndUpdateMetrics(bytes, header, messageSize, Overload.BYTES_IN_FLIGHT))
{
if (decoder.isActive())
ClientMetrics.instance.pauseConnection();
}
backpressure = Overload.BYTES_IN_FLIGHT;
}
if (DatabaseDescriptor.getNativeTransportRateLimitingEnabled())
{
// Reserve a permit even if we've already triggered backpressure on bytes in flight.
long delay = requestRateLimiter.reserveAndGetDelay(RATE_LIMITER_DELAY_UNIT);
if (backpressure == Overload.NONE && delay > 0)
{
if (processRequestAndUpdateMetrics(bytes, header, messageSize, Overload.REQUESTS))
{
if (decoder.isActive())
ClientMetrics.instance.pauseConnection();
// Schedule a wakup here if we process successfully. The connection should be closing otherwise.
scheduleConnectionWakeupTask(delay, RATE_LIMITER_DELAY_UNIT);
}
backpressure = Overload.REQUESTS;
}
}
// If we triggered backpressure, make sure the caller stops processing frames after the request completes.
if (backpressure != Overload.NONE)
return false;
}
return processRequestAndUpdateMetrics(bytes, header, messageSize, Overload.NONE);
}
private boolean processRequestAndUpdateMetrics(ShareableBytes bytes, Envelope.Header header, int messageSize, Overload backpressure)
{
channelPayloadBytesInFlight += messageSize;
incrementReceivedMessageMetrics(messageSize);
return processRequest(composeRequest(header, bytes));
return processRequest(composeRequest(header, bytes), backpressure);
}
private void discardAndThrow(Limit endpointReserve, Limit globalReserve,
ByteBuffer buf, Envelope.Header header, int messageSize,
Overload overload)
{
ClientMetrics.instance.markRequestDiscarded();
logOverload(endpointReserve, globalReserve, header, messageSize);
OverloadedException exception = buildOverloadedException(endpointReserve, globalReserve, overload);
handleError(exception, header);
// Don't stop processing incoming messages, as we 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);
buf.position(buf.position() + Envelope.Header.LENGTH + messageSize);
}
private OverloadedException buildOverloadedException(Limit endpointReserve, Limit globalReserve, Overload overload) {
return overload == Overload.REQUESTS
? new OverloadedException(String.format("Request breached global limit of %d requests/second. Server is " +
"currently in an overloaded state and cannot accept more requests.",
requestRateLimiter.getRate()))
: new OverloadedException(String.format("Request breached limit on bytes in flight. (Endpoint: %d/%d bytes, Global: %d/%d bytes.) " +
"Server is currently in an overloaded state and cannot accept more requests.",
endpointReserve.using(), endpointReserve.limit(), globalReserve.using(), globalReserve.limit()));
}
private void logOverload(Limit endpointReserve, Limit globalReserve, Envelope.Header header, int messageSize)
{
logger.trace("Discarded request of size {} with {} bytes in flight on channel. " +
"Using {}/{} bytes of endpoint limit and {}/{} bytes of global limit. " +
"Global rate limiter: {} Header: {}",
messageSize, channelPayloadBytesInFlight,
endpointReserve.using(), endpointReserve.limit(), globalReserve.using(), globalReserve.limit(),
requestRateLimiter, header);
}
private boolean handleProtocolException(ProtocolException exception,
@ -250,12 +338,18 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
}
protected boolean processRequest(Envelope request)
{
return processRequest(request, Overload.NONE);
}
protected boolean processRequest(Envelope request, Overload backpressure)
{
M message = null;
try
{
message = messageDecoder.decode(channel, request);
dispatcher.accept(channel, message, this::toFlushItem);
dispatcher.accept(channel, message, this::toFlushItem, backpressure);
// sucessfully delivered a CQL message to the execution
// stage, so reset the counter of consecutive errors
consecutiveMessageErrors = 0;
@ -293,8 +387,6 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
* 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)
{
@ -311,8 +403,6 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
* 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)
{
@ -328,8 +418,6 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
* 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 streamId
*/
private void handleError(Throwable t, int streamId)
{
@ -342,8 +430,6 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
* 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)
{
@ -405,8 +491,9 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
// max CQL message size defaults to 256mb, so should be safe to downcast
int messageSize = Ints.checkedCast(header.bodySizeInBytes);
receivedBytes += buf.remaining();
LargeMessage largeMessage = new LargeMessage(header);
if (!acquireCapacity(header, endpointReserve, globalReserve))
{
// In the case of large messages, never stop processing incoming frames
@ -424,21 +511,49 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
// concurrently.
if (throwOnOverload)
{
// discard the request and throw an exception
// Mark as overloaded so that discard the message after consuming any subsequent frames.
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 discard the message
// after consuming any subsequent frames
largeMessage.markOverloaded();
logOverload(endpointReserve, globalReserve, header, messageSize);
largeMessage.markOverloaded(Overload.BYTES_IN_FLIGHT);
}
}
else if (DatabaseDescriptor.getNativeTransportRateLimitingEnabled())
{
if (throwOnOverload)
{
if (!requestRateLimiter.tryReserve())
{
ClientMetrics.instance.markRequestDiscarded();
logOverload(endpointReserve, globalReserve, header, messageSize);
// Mark as overloaded so that we discard the message after consuming any subsequent frames.
// (i.e. Request resources we may already have acquired above will be released.)
largeMessage.markOverloaded(Overload.REQUESTS);
this.largeMessage = largeMessage;
largeMessage.supply(frame);
return true;
}
}
else
{
long delay = requestRateLimiter.reserveAndGetDelay(RATE_LIMITER_DELAY_UNIT);
if (delay > 0)
{
this.largeMessage = largeMessage;
largeMessage.markBackpressure(Overload.REQUESTS);
largeMessage.supply(frame);
if (decoder.isActive())
ClientMetrics.instance.pauseConnection();
scheduleConnectionWakeupTask(delay, RATE_LIMITER_DELAY_UNIT);
return false;
}
}
}
this.largeMessage = largeMessage;
largeMessage.supply(frame);
return true;
@ -454,6 +569,31 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
return channel.id().asShortText();
}
private void scheduleConnectionWakeupTask(long waitLength, TimeUnit unit)
{
channel.eventLoop().schedule(() ->
{
try
{
// We might have already reactivated via another wake task.
if (!decoder.isActive())
{
decoder.reactivate();
// Only update the relevant metric if we've actually activated.
if (decoder.isActive())
ClientMetrics.instance.unpauseConnection();
}
}
catch (Throwable t)
{
fatalExceptionCaught(t);
}
},
waitLength,
unit);
}
@SuppressWarnings("BooleanMethodIsAlwaysInverted")
private boolean acquireCapacityAndQueueOnFailure(Envelope.Header header, Limit endpointReserve, Limit globalReserve)
{
@ -515,7 +655,8 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
{
private static final long EXPIRES_AT = Long.MAX_VALUE;
private boolean overloaded = false;
private Overload overload = Overload.NONE;
private Overload backpressure = Overload.NONE;
private LargeMessage(Envelope.Header header)
{
@ -541,19 +682,22 @@ public class CQLMessageHandler<M extends Message> extends AbstractMessageHandler
* 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()
private void markOverloaded(Overload overload)
{
overloaded = true;
this.overload = overload;
}
private void markBackpressure(Overload backpressure)
{
this.backpressure = backpressure;
}
protected void onComplete()
{
if (overloaded)
handleErrorAndRelease(new OverloadedException("Server is in overloaded state. " +
"Cannot accept more requests at this point"), header);
if (overload != Overload.NONE)
handleErrorAndRelease(buildOverloadedException(endpointReserveCapacity, globalReserveCapacity, overload), header);
else if (!isCorrupt)
processRequest(assembleFrame());
processRequest(assembleFrame(), backpressure);
}
protected void abort()

41
src/java/org/apache/cassandra/transport/ClientResourceLimits.java
View File

@ -24,6 +24,7 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.utils.concurrent.NonBlockingRateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -42,6 +43,10 @@ public class ClientResourceLimits
private static final AbstractMessageHandler.WaitQueue GLOBAL_QUEUE = AbstractMessageHandler.WaitQueue.global(GLOBAL_LIMIT);
private static final ConcurrentMap<InetAddress, Allocator> PER_ENDPOINT_ALLOCATORS = new ConcurrentHashMap<>();
public static final NonBlockingRateLimiter GLOBAL_REQUEST_LIMITER = new NonBlockingRateLimiter(getNativeTransportMaxRequestsPerSecond());
public enum Overload { NONE, REQUESTS, BYTES_IN_FLIGHT }
public static Allocator getAllocatorForEndpoint(InetAddress endpoint)
{
while (true)
@ -95,6 +100,19 @@ public class ClientResourceLimits
return histogram.getSnapshot();
}
public static int getNativeTransportMaxRequestsPerSecond()
{
return DatabaseDescriptor.getNativeTransportMaxRequestsPerSecond();
}
public static void setNativeTransportMaxRequestsPerSecond(int newPerSecond)
{
int existingPerSecond = getNativeTransportMaxRequestsPerSecond();
DatabaseDescriptor.setNativeTransportMaxRequestsPerSecond(newPerSecond);
GLOBAL_REQUEST_LIMITER.setRate(newPerSecond);
logger.info("Changed native_transport_max_requests_per_second from {} to {}", existingPerSecond, newPerSecond);
}
/**
* 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
@ -166,8 +184,8 @@ public class ClientResourceLimits
*
* @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.
* either INSUFFICIENT_GLOBAL or INSUFFICIENT_ENDPOINT to indicate which
* reserve rejected the allocation request.
*/
ResourceLimits.Outcome tryAllocate(long amount)
{
@ -211,19 +229,18 @@ public class ClientResourceLimits
return endpointAndGlobal.global().using();
}
@Override
public String toString()
{
return String.format("InflightEndpointRequestPayload: %d/%d, InflightOverallRequestPayload: %d/%d",
endpointAndGlobal.endpoint().using(),
endpointAndGlobal.endpoint().limit(),
endpointAndGlobal.global().using(),
endpointAndGlobal.global().limit());
return String.format("Using %d/%d bytes of endpoint limit and %d/%d bytes of global limit.",
endpointAndGlobal.endpoint().using(), endpointAndGlobal.endpoint().limit(),
endpointAndGlobal.global().using(), endpointAndGlobal.global().limit());
}
}
/**
* Used in protocol V5 and later by the AbstractMessageHandler/CQLMessageHandler hierarchy.
* This hides the allocate/tryAllocate/release methods from EndpointResourceLimits and exposes
* This hides the allocate/tryAllocate/release methods from {@link ClientResourceLimits} 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)
@ -234,9 +251,10 @@ public class ClientResourceLimits
AbstractMessageHandler.WaitQueue globalWaitQueue();
ResourceLimits.Limit endpointLimit();
AbstractMessageHandler.WaitQueue endpointWaitQueue();
NonBlockingRateLimiter requestRateLimiter();
void release();
static class Default implements ResourceProvider
class Default implements ResourceProvider
{
private final Allocator limits;
@ -265,6 +283,11 @@ public class ClientResourceLimits
return limits.waitQueue;
}
public NonBlockingRateLimiter requestRateLimiter()
{
return GLOBAL_REQUEST_LIMITER;
}
public void release()
{
limits.release();

37
src/java/org/apache/cassandra/transport/Dispatcher.java
View File

@ -20,8 +20,12 @@ package org.apache.cassandra.transport;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.Channel;
import io.netty.channel.EventLoop;
import io.netty.util.AttributeKey;
@ -30,15 +34,19 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.net.FrameEncoder;
import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.transport.ClientResourceLimits.Overload;
import org.apache.cassandra.transport.Flusher.FlushItem;
import org.apache.cassandra.transport.messages.ErrorMessage;
import org.apache.cassandra.transport.messages.EventMessage;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.NoSpamLogger;
import static org.apache.cassandra.concurrent.SharedExecutorPool.SHARED;
public class Dispatcher
{
private static final Logger logger = LoggerFactory.getLogger(Dispatcher.class);
private static final LocalAwareExecutorService requestExecutor = SHARED.newExecutor(DatabaseDescriptor.getNativeTransportMaxThreads(),
DatabaseDescriptor::setNativeTransportMaxThreads,
"transport",
@ -65,20 +73,37 @@ public class Dispatcher
this.useLegacyFlusher = useLegacyFlusher;
}
public void dispatch(Channel channel, Message.Request request, FlushItemConverter forFlusher)
public void dispatch(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure)
{
requestExecutor.submit(() -> processRequest(channel, request, forFlusher));
requestExecutor.submit(() -> processRequest(channel, request, forFlusher, backpressure));
}
/**
* Note: this method may be executed on the netty event loop, during initial protocol negotiation
*/
static Message.Response processRequest(ServerConnection connection, Message.Request request)
static Message.Response processRequest(ServerConnection connection, Message.Request request, Overload backpressure)
{
long queryStartNanoTime = System.nanoTime();
if (connection.getVersion().isGreaterOrEqualTo(ProtocolVersion.V4))
ClientWarn.instance.captureWarnings();
if (backpressure == Overload.REQUESTS)
{
String message = String.format("Request breached global limit of %d requests/second and triggered backpressure.",
ClientResourceLimits.getNativeTransportMaxRequestsPerSecond());
NoSpamLogger.log(logger, NoSpamLogger.Level.INFO, 1, TimeUnit.MINUTES, message);
ClientWarn.instance.warn(message);
}
else if (backpressure == Overload.BYTES_IN_FLIGHT)
{
String message = String.format("Request breached limit(s) on bytes in flight (Endpoint: %d, Global: %d) and triggered backpressure.",
ClientResourceLimits.getEndpointLimit(), ClientResourceLimits.getGlobalLimit());
NoSpamLogger.log(logger, NoSpamLogger.Level.INFO, 1, TimeUnit.MINUTES, message);
ClientWarn.instance.warn(message);
}
QueryState qstate = connection.validateNewMessage(request.type, connection.getVersion());
Message.logger.trace("Received: {}, v={}", request, connection.getVersion());
@ -94,7 +119,7 @@ public class Dispatcher
/**
* Note: this method is not expected to execute on the netty event loop.
*/
void processRequest(Channel channel, Message.Request request, FlushItemConverter forFlusher)
void processRequest(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure)
{
final Message.Response response;
final ServerConnection connection;
@ -103,7 +128,7 @@ public class Dispatcher
{
assert request.connection() instanceof ServerConnection;
connection = (ServerConnection) request.connection();
response = processRequest(connection, request);
response = processRequest(connection, request, backpressure);
toFlush = forFlusher.toFlushItem(channel, request, response);
Message.logger.trace("Responding: {}, v={}", response, connection.getVersion());
}
@ -152,7 +177,7 @@ public class Dispatcher
* for delivering events to registered clients is dependent on protocol version and the configuration
* of the pipeline. For v5 and newer connections, the event message is encoded into an Envelope,
* wrapped in a FlushItem and then delivered via the pipeline's flusher, in a similar way to
* a Response returned from {@link #processRequest(Channel, Message.Request, FlushItemConverter)}.
* a Response returned from {@link #processRequest(Channel, Message.Request, FlushItemConverter, Overload)}.
* It's worth noting that events are not generally fired as a direct response to a client request,
* so this flush item has a null request attribute. The dispatcher itself is created when the
* pipeline is first configured during protocol negotiation and is attached to the channel for

44
src/java/org/apache/cassandra/transport/ExceptionHandlers.java
View File

@ -32,6 +32,7 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.metrics.ClientMetrics;
import org.apache.cassandra.net.FrameEncoder;
import org.apache.cassandra.transport.messages.ErrorMessage;
@ -87,22 +88,8 @@ public class ExceptionHandlers
JVMStabilityInspector.inspectThrowable(cause);
}
}
if (Throwables.anyCauseMatches(cause, t -> t instanceof ProtocolException))
{
// if any ProtocolExceptions is not silent, then handle
if (Throwables.anyCauseMatches(cause, t -> t instanceof ProtocolException && !((ProtocolException) t).isSilent()))
{
ClientMetrics.instance.markProtocolException();
// since protocol exceptions are expected to be client issues, not logging stack trace
// to avoid spamming the logs once a bad client shows up
NoSpamLogger.log(logger, NoSpamLogger.Level.WARN, 1, TimeUnit.MINUTES, "Protocol exception with client networking: " + cause.getMessage());
}
}
else
{
ClientMetrics.instance.markUnknownException();
logger.warn("Unknown exception in client networking", cause);
}
logClientNetworkingExceptions(cause);
}
private static boolean isFatal(Throwable cause)
@ -112,6 +99,31 @@ public class ExceptionHandlers
}
}
static void logClientNetworkingExceptions(Throwable cause)
{
if (Throwables.anyCauseMatches(cause, t -> t instanceof ProtocolException))
{
// if any ProtocolExceptions is not silent, then handle
if (Throwables.anyCauseMatches(cause, t -> t instanceof ProtocolException && !((ProtocolException) t).isSilent()))
{
ClientMetrics.instance.markProtocolException();
// since protocol exceptions are expected to be client issues, not logging stack trace
// to avoid spamming the logs once a bad client shows up
NoSpamLogger.log(logger, NoSpamLogger.Level.WARN, 1, TimeUnit.MINUTES, "Protocol exception with client networking: " + cause.getMessage());
}
}
else if (Throwables.anyCauseMatches(cause, t -> t instanceof OverloadedException))
{
// Once the threshold for overload is breached, it will very likely spam the logs...
NoSpamLogger.log(logger, NoSpamLogger.Level.INFO, 1, TimeUnit.MINUTES, cause.getMessage());
}
else
{
ClientMetrics.instance.markUnknownException();
logger.warn("Unknown exception in client networking", cause);
}
}
/**
* Include the channel info in the logged information for unexpected errors, and (if {@link #alwaysLogAtError} is
* false then choose the log level based on the type of exception (some are clearly client issues and shouldn't be

3
src/java/org/apache/cassandra/transport/InitialConnectionHandler.java
View File

@ -25,6 +25,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.cassandra.transport.ClientResourceLimits.Overload;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -147,7 +148,7 @@ public class InitialConnectionHandler extends ByteToMessageDecoder
promise = new VoidChannelPromise(ctx.channel(), false);
}
final Message.Response response = Dispatcher.processRequest((ServerConnection) connection, startup);
final Message.Response response = Dispatcher.processRequest((ServerConnection) connection, startup, Overload.NONE);
outbound = response.encode(inbound.header.version);
ctx.writeAndFlush(outbound, promise);
logger.debug("Configured pipeline: {}", ctx.pipeline());

208
src/java/org/apache/cassandra/transport/PreV5Handlers.java
View File

@ -19,8 +19,9 @@
package org.apache.cassandra.transport;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.transport.ClientResourceLimits.Overload;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -39,10 +40,9 @@ import org.apache.cassandra.metrics.ClientMetrics;
import org.apache.cassandra.net.ResourceLimits;
import org.apache.cassandra.transport.messages.ErrorMessage;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.Throwables;
import static org.apache.cassandra.transport.Message.logger;
import static org.apache.cassandra.transport.CQLMessageHandler.RATE_LIMITER_DELAY_UNIT;
import static org.apache.cassandra.transport.ClientResourceLimits.GLOBAL_REQUEST_LIMITER;
public class PreV5Handlers
{
@ -63,7 +63,9 @@ public class PreV5Handlers
* Note: should only be accessed while on the netty event loop.
*/
private long channelPayloadBytesInFlight;
private boolean paused;
/** The cause of the current connection pause, or {@link Overload#NONE} if it is unpaused. */
private Overload backpressure = Overload.NONE;
LegacyDispatchHandler(Dispatcher dispatcher, ClientResourceLimits.Allocator endpointPayloadTracker)
{
@ -71,11 +73,12 @@ public class PreV5Handlers
this.endpointPayloadTracker = endpointPayloadTracker;
}
protected void channelRead0(ChannelHandlerContext ctx, Message.Request request) throws Exception
protected void channelRead0(ChannelHandlerContext ctx, Message.Request request)
{
// if we decide to handle this message, process it outside of the netty event loop
if (shouldHandleRequest(ctx, request))
dispatcher.dispatch(ctx.channel(), request, this::toFlushItem);
// The only reason we won't process this message is if checkLimits() throws an OverloadedException.
// (i.e. Even if backpressure is applied, the current request is allowed to finish.)
checkLimits(ctx, request);
dispatcher.dispatch(ctx.channel(), request, this::toFlushItem, backpressure);
}
// Acts as a Dispatcher.FlushItemConverter
@ -95,76 +98,143 @@ public class PreV5Handlers
// since the request has been processed, decrement inflight payload at channel, endpoint and global levels
channelPayloadBytesInFlight -= itemSize;
ResourceLimits.Outcome endpointGlobalReleaseOutcome = endpointPayloadTracker.release(itemSize);
boolean globalInFlightBytesBelowLimit = endpointPayloadTracker.release(itemSize) == ResourceLimits.Outcome.BELOW_LIMIT;
// now check to see if we need to reenable the channel's autoRead.
// If the current payload side is zero, we must reenable autoread as
// 1) we allow no other thread/channel to do it, and
// 2) there's no other events following this one (becuase we're at zero bytes in flight),
// so no successive to trigger the other clause in this if-block
// Now check to see if we need to reenable the channel's autoRead.
//
// note: this path is only relevant when part of a pre-V5 pipeline, as only in this case is
// If the current payload bytes in flight is zero, we must reenable autoread as
// 1) we allow no other thread/channel to do it, and
// 2) there are no other events following this one (becuase we're at zero bytes in flight),
// so no successive to trigger the other clause in this if-block.
//
// The only exception to this is if the global request rate limit has been breached, which means
// we'll have to wait until a scheduled wakeup task unpauses the connection.
//
// Note: This path is only relevant when part of a pre-V5 pipeline, as only in this case is
// paused ever set to true. In pipelines configured for V5 or later, backpressure and control
// over the inbound pipeline's autoread status are handled by the FrameDecoder/FrameProcessor.
ChannelConfig config = item.channel.config();
if (paused && (channelPayloadBytesInFlight == 0 || endpointGlobalReleaseOutcome == ResourceLimits.Outcome.BELOW_LIMIT))
if (backpressure == Overload.BYTES_IN_FLIGHT && (channelPayloadBytesInFlight == 0 || globalInFlightBytesBelowLimit))
{
unpauseConnection(config);
}
}
/**
* Checks limits on bytes in flight and the request rate limiter (if enabled) to determine whether to drop a
* request or trigger backpressure and pause the connection.
* <p>
* The check for inflight payload to potentially discard the request should have been ideally in one of the
* first handlers in the pipeline (Envelope.Decoder::decode()). However, in case of any exception thrown between
* that handler (where inflight payload is incremented) and this handler (Dispatcher::channelRead0) (where
* inflight payload in decremented), inflight payload becomes erroneous. ExceptionHandler is not sufficient for
* this purpose since it does not have the message envelope associated with the exception.
* <p>
* If the connection is configured to throw {@link OverloadedException}, requests that breach the rate limit are
* not counted against that limit.
* <p>
* Note: this method should execute on the netty event loop.
*
* @throws ErrorMessage.WrappedException with an {@link OverloadedException} if overload occurs and the
* connection is configured to throw on overload
*/
private void checkLimits(ChannelHandlerContext ctx, Message.Request request)
{
long requestSize = request.getSource().header.bodySizeInBytes;
if (request.connection.isThrowOnOverload())
{
if (endpointPayloadTracker.tryAllocate(requestSize) != ResourceLimits.Outcome.SUCCESS)
{
discardAndThrow(request, requestSize, Overload.BYTES_IN_FLIGHT);
}
if (DatabaseDescriptor.getNativeTransportRateLimitingEnabled() && !GLOBAL_REQUEST_LIMITER.tryReserve())
{
// We've already allocated against the payload tracker here, so release those resources.
endpointPayloadTracker.release(requestSize);
discardAndThrow(request, requestSize, Overload.REQUESTS);
}
}
else
{
// Any request that gets here will be processed, so increment the channel bytes in flight.
channelPayloadBytesInFlight += requestSize;
// Check for overloaded state by trying to allocate the message size from inflight payload trackers
if (endpointPayloadTracker.tryAllocate(requestSize) != ResourceLimits.Outcome.SUCCESS)
{
endpointPayloadTracker.allocate(requestSize);
pauseConnection(ctx);
backpressure = Overload.BYTES_IN_FLIGHT;
}
if (DatabaseDescriptor.getNativeTransportRateLimitingEnabled())
{
// Reserve a permit even if we've already triggered backpressure on bytes in flight.
long delay = GLOBAL_REQUEST_LIMITER.reserveAndGetDelay(RATE_LIMITER_DELAY_UNIT);
// If we've already triggered backpressure on bytes in flight, no further action is necessary.
if (backpressure == Overload.NONE && delay > 0)
{
pauseConnection(ctx);
// A permit isn't immediately available, so schedule an unpause for when it is.
ctx.channel().eventLoop().schedule(() -> unpauseConnection(ctx.channel().config()), delay, RATE_LIMITER_DELAY_UNIT);
backpressure = Overload.REQUESTS;
}
}
}
}
private void pauseConnection(ChannelHandlerContext ctx)
{
if (ctx.channel().config().isAutoRead())
{
ctx.channel().config().setAutoRead(false);
ClientMetrics.instance.pauseConnection();
}
}
private void unpauseConnection(ChannelConfig config)
{
backpressure = Overload.NONE;
if (!config.isAutoRead())
{
paused = false;
ClientMetrics.instance.unpauseConnection();
config.setAutoRead(true);
}
}
/**
* This check for inflight payload to potentially discard the request should have been ideally in one of the
* first handlers in the pipeline (Envelope.Decoder::decode()). However, incase of any exception thrown between that
* handler (where inflight payload is incremented) and this handler (Dispatcher::channelRead0) (where inflight
* payload in decremented), inflight payload becomes erroneous. ExceptionHandler is not sufficient for this
* purpose since it does not have the message envelope associated with the exception.
* <p>
* Note: this method should execute on the netty event loop.
*/
private boolean shouldHandleRequest(ChannelHandlerContext ctx, Message.Request request)
private void discardAndThrow(Message.Request request, long requestSize, Overload overload)
{
long requestSize = request.getSource().header.bodySizeInBytes;
ClientMetrics.instance.markRequestDiscarded();
// check for overloaded state by trying to allocate the message size from inflight payload trackers
if (endpointPayloadTracker.tryAllocate(requestSize) != ResourceLimits.Outcome.SUCCESS)
{
if (request.connection.isThrowOnOverload())
{
// discard the request and throw an exception
ClientMetrics.instance.markRequestDiscarded();
logger.trace("Discarded request of size: {}. InflightChannelRequestPayload: {}, {}, Request: {}",
requestSize,
channelPayloadBytesInFlight,
endpointPayloadTracker.toString(),
request);
throw ErrorMessage.wrap(new OverloadedException("Server is in overloaded state. Cannot accept more requests at this point"),
request.getSource().header.streamId);
}
else
{
// set backpressure on the channel, and handle the request
endpointPayloadTracker.allocate(requestSize);
ctx.channel().config().setAutoRead(false);
ClientMetrics.instance.pauseConnection();
paused = true;
}
}
logger.trace("Discarded request of size {} with {} bytes in flight on channel. {} " +
"Global rate limiter: {} Request: {}",
requestSize, channelPayloadBytesInFlight, endpointPayloadTracker,
GLOBAL_REQUEST_LIMITER, request);
channelPayloadBytesInFlight += requestSize;
return true;
OverloadedException exception = overload == Overload.REQUESTS
? new OverloadedException(String.format("Request breached global limit of %d requests/second. Server is " +
"currently in an overloaded state and cannot accept more requests.",
GLOBAL_REQUEST_LIMITER.getRate()))
: new OverloadedException(String.format("Request breached limit on bytes in flight. (%s)) " +
"Server is currently in an overloaded state and cannot accept more requests.",
endpointPayloadTracker));
throw ErrorMessage.wrap(exception, request.getSource().header.streamId);
}
@Override
public void channelInactive(ChannelHandlerContext ctx)
{
endpointPayloadTracker.release();
if (paused)
if (!ctx.channel().config().isAutoRead())
{
paused = false;
ClientMetrics.instance.unpauseConnection();
}
ctx.fireChannelInactive();
@ -181,7 +251,7 @@ public class PreV5Handlers
public static final ProtocolDecoder instance = new ProtocolDecoder();
private ProtocolDecoder(){}
public void decode(ChannelHandlerContext ctx, Envelope source, List results)
public void decode(ChannelHandlerContext ctx, Envelope source, List<Object> results)
{
try
{
@ -212,7 +282,7 @@ public class PreV5Handlers
{
public static final ProtocolEncoder instance = new ProtocolEncoder();
private ProtocolEncoder(){}
public void encode(ChannelHandlerContext ctx, Message source, List results)
public void encode(ChannelHandlerContext ctx, Message source, List<Object> results)
{
ProtocolVersion version = getConnectionVersion(ctx);
results.add(source.encode(version));
@ -244,22 +314,8 @@ public class PreV5Handlers
if (isFatal(cause))
future.addListener((ChannelFutureListener) f -> ctx.close());
}
if (Throwables.anyCauseMatches(cause, t -> t instanceof ProtocolException))
{
// if any ProtocolExceptions is not silent, then handle
if (Throwables.anyCauseMatches(cause, t -> t instanceof ProtocolException && !((ProtocolException) t).isSilent()))
{
ClientMetrics.instance.markProtocolException();
// since protocol exceptions are expected to be client issues, not logging stack trace
// to avoid spamming the logs once a bad client shows up
NoSpamLogger.log(logger, NoSpamLogger.Level.WARN, 1, TimeUnit.MINUTES, "Protocol exception with client networking: " + cause.getMessage());
}
}
else
{
ClientMetrics.instance.markUnknownException();
logger.warn("Unknown exception in client networking", cause);
}
ExceptionHandlers.logClientNetworkingExceptions(cause);
JVMStabilityInspector.inspectThrowable(cause);
}

51
src/java/org/apache/cassandra/transport/SimpleClient.java
View File

@ -27,6 +27,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
import org.apache.cassandra.transport.ClientResourceLimits.Overload;
import org.apache.cassandra.utils.concurrent.NonBlockingRateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -50,9 +52,13 @@ import org.apache.cassandra.transport.messages.*;
import static org.apache.cassandra.transport.CQLMessageHandler.envelopeSize;
import static org.apache.cassandra.transport.Flusher.MAX_FRAMED_PAYLOAD_SIZE;
import static org.apache.cassandra.utils.concurrent.NonBlockingRateLimiter.NO_OP_LIMITER;
public class SimpleClient implements Closeable
{
public static final int TIMEOUT_SECONDS = 10;
static
{
InternalLoggerFactory.setDefaultFactory(new Slf4JLoggerFactory());
@ -281,7 +287,7 @@ public class SimpleClient implements Closeable
{
request.attach(connection);
lastWriteFuture = channel.writeAndFlush(Collections.singletonList(request));
Message.Response msg = responseHandler.responses.poll(10, TimeUnit.SECONDS);
Message.Response msg = responseHandler.responses.poll(TIMEOUT_SECONDS, TimeUnit.SECONDS);
if (msg == null)
throw new RuntimeException("timeout");
if (throwOnErrorResponse && msg instanceof ErrorMessage)
@ -310,7 +316,7 @@ public class SimpleClient implements Closeable
}
lastWriteFuture = channel.writeAndFlush(requests);
long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(10);
long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS);
for (int i = 0; i < requests.size(); i++)
{
Message.Response msg = responseHandler.responses.poll(deadline - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
@ -393,7 +399,7 @@ public class SimpleClient implements Closeable
case AUTHENTICATE:
if (response.header.version.isGreaterOrEqualTo(ProtocolVersion.V5))
{
configureModernPipeline(ctx, response);
configureModernPipeline(ctx, response, largeMessageThreshold);
// consuming the message is done when setting up the pipeline
}
else
@ -414,7 +420,7 @@ public class SimpleClient implements Closeable
}
}
private void configureModernPipeline(ChannelHandlerContext ctx, Envelope response)
private void configureModernPipeline(ChannelHandlerContext ctx, Envelope response, int largeMessageThreshold)
{
logger.info("Configuring modern pipeline");
ChannelPipeline pipeline = ctx.pipeline();
@ -434,7 +440,7 @@ public class SimpleClient implements Closeable
FrameEncoder frameEncoder = frameEncoder(ctx);
FrameEncoder.PayloadAllocator payloadAllocator = frameEncoder.allocator();
CQLMessageHandler.MessageConsumer<Message.Response> responseConsumer = (c, message, converter) -> {
CQLMessageHandler.MessageConsumer<Message.Response> responseConsumer = (c, message, converter, backpressured) -> {
responseHandler.handleResponse(c, message);
};
@ -470,6 +476,12 @@ public class SimpleClient implements Closeable
return endpointQueue;
}
@Override
public NonBlockingRateLimiter requestRateLimiter()
{
return NO_OP_LIMITER;
}
public void release()
{
}
@ -512,7 +524,7 @@ public class SimpleClient implements Closeable
Connection connection = ctx.channel().attr(Connection.attributeKey).get();
// The only case the connection can be null is when we send the initial STARTUP message (client side thus)
ProtocolVersion version = connection == null ? ProtocolVersion.CURRENT : connection.getVersion();
SimpleFlusher flusher = new SimpleFlusher(frameEncoder);
SimpleFlusher flusher = new SimpleFlusher(frameEncoder, largeMessageThreshold);
for (Message message : (List<Message>) msg)
flusher.enqueue(message.encode(version));
@ -522,7 +534,7 @@ public class SimpleClient implements Closeable
pipeline.remove(this);
Message.Response message = messageDecoder.decode(ctx.channel(), response);
responseConsumer.accept(channel, message, (ch, req, resp) -> null);
responseConsumer.accept(channel, message, (ch, req, resp) -> null, Overload.NONE);
}
private FrameDecoder frameDecoder(ChannelHandlerContext ctx, BufferPoolAllocator allocator)
@ -669,10 +681,17 @@ public class SimpleClient implements Closeable
final Queue<Envelope> outbound = new ConcurrentLinkedQueue<>();
final FrameEncoder frameEncoder;
private final AtomicBoolean scheduled = new AtomicBoolean(false);
private final int largeMessageThreshold;
SimpleFlusher(FrameEncoder frameEncoder, int largeMessageThreshold)
{
this.frameEncoder = frameEncoder;
this.largeMessageThreshold = largeMessageThreshold;
}
SimpleFlusher(FrameEncoder frameEncoder)
{
this.frameEncoder = frameEncoder;
this(frameEncoder, MAX_FRAMED_PAYLOAD_SIZE);
}
public void enqueue(Envelope message)
@ -709,14 +728,14 @@ public class SimpleClient implements Closeable
Envelope f;
while ((f = outbound.poll()) != null)
{
if (f.header.bodySizeInBytes > MAX_FRAMED_PAYLOAD_SIZE)
if (f.header.bodySizeInBytes > largeMessageThreshold)
{
combiner.addAll(writeLargeMessage(ctx, f));
}
else
{
int messageSize = envelopeSize(f.header);
if (bufferSize + messageSize >= MAX_FRAMED_PAYLOAD_SIZE)
if (bufferSize + messageSize >= largeMessageThreshold)
{
combiner.add(flushBuffer(ctx, buffer, bufferSize));
buffer = new ArrayList<>();
@ -751,9 +770,9 @@ public class SimpleClient implements Closeable
private FrameEncoder.Payload allocate(int size, boolean selfContained)
{
FrameEncoder.Payload payload = frameEncoder.allocator()
.allocate(selfContained, Math.min(size, MAX_FRAMED_PAYLOAD_SIZE));
if (size >= MAX_FRAMED_PAYLOAD_SIZE)
payload.buffer.limit(MAX_FRAMED_PAYLOAD_SIZE);
.allocate(selfContained, Math.min(size, largeMessageThreshold));
if (size >= largeMessageThreshold)
payload.buffer.limit(largeMessageThreshold);
return payload;
}
@ -766,14 +785,14 @@ public class SimpleClient implements Closeable
boolean firstFrame = true;
while (f.body.readableBytes() > 0 || firstFrame)
{
int payloadSize = Math.min(f.body.readableBytes(), MAX_FRAMED_PAYLOAD_SIZE);
int payloadSize = Math.min(f.body.readableBytes(), largeMessageThreshold);
payload = allocate(f.body.readableBytes(), false);
buf = payload.buffer;
// BufferPool may give us a buffer larger than we asked for.
// FrameEncoder may object if buffer.remaining is >= MAX_SIZE.
if (payloadSize >= MAX_FRAMED_PAYLOAD_SIZE)
buf.limit(MAX_FRAMED_PAYLOAD_SIZE);
if (payloadSize >= largeMessageThreshold)
buf.limit(largeMessageThreshold);
if (firstFrame)
{

188
src/java/org/apache/cassandra/utils/concurrent/NonBlockingRateLimiter.java Normal file
View File

@ -0,0 +1,188 @@
/*
* 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.utils.concurrent;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Ticker;
import javax.annotation.concurrent.ThreadSafe;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static java.lang.Math.toIntExact;
/**
* A rate limiter implementation that allows callers to reserve permits that may only be available
* in the future, delegating to them decisions about how to schedule/delay work and whether or not
* to block execution to do so.
*/
@SuppressWarnings("UnstableApiUsage")
@ThreadSafe
public class NonBlockingRateLimiter
{
public static final long NANOS_PER_SECOND = TimeUnit.SECONDS.toNanos(1);
static final long DEFAULT_BURST_NANOS = NANOS_PER_SECOND;
/** a starting time for elapsed time calculations */
private final long startedNanos;
/** nanoseconds from start time corresponding to the next available permit */
private final AtomicLong nextAvailable = new AtomicLong();
private volatile Ticker ticker;
private volatile int permitsPerSecond;
/** time in nanoseconds between permits on the timeline */
private volatile long intervalNanos;
/**
* To allow the limiter to more closely adhere to the configured rate in the face of
* unevenly distributed permits requests, it will allow a number of permits equal to
* burstNanos / intervalNanos to be issued in a "burst" before reaching a steady state.
*
* Another way to think about this is that it allows us to bring forward the permits
* from short periods of inactivity. This is especially useful when the upstream user
* of the limiter delays request processing mechanism contains overhead that is longer
* than the intervalNanos in duration.
*/
private final long burstNanos;
public NonBlockingRateLimiter(int permitsPerSecond)
{
this(permitsPerSecond, DEFAULT_BURST_NANOS, Ticker.systemTicker());
}
@VisibleForTesting
public NonBlockingRateLimiter(int permitsPerSecond, long burstNanos, Ticker ticker)
{
this.startedNanos = ticker.read();
this.burstNanos = burstNanos;
setRate(permitsPerSecond, ticker);
}
public void setRate(int permitsPerSecond)
{
setRate(permitsPerSecond, Ticker.systemTicker());
}
@VisibleForTesting
public synchronized void setRate(int permitsPerSecond, Ticker ticker)
{
Preconditions.checkArgument(permitsPerSecond > 0, "permits/second must be positive");
Preconditions.checkArgument(permitsPerSecond <= NANOS_PER_SECOND, "permits/second cannot be greater than " + NANOS_PER_SECOND);
this.ticker = ticker;
this.permitsPerSecond = permitsPerSecond;
intervalNanos = NANOS_PER_SECOND / permitsPerSecond;
nextAvailable.set(nanosElapsed());
}
/**
* @return the number of available permits per second
*/
public int getRate()
{
return permitsPerSecond;
}
/**
* Reserves a single permit slot on the timeline which may not yet be available.
*
* @return time until the reserved permit will be available (or zero if it already is) in the specified units
*/
public long reserveAndGetDelay(TimeUnit delayUnit)
{
long nowNanos = nanosElapsed();
for (;;)
{
long prev = nextAvailable.get();
long interval = this.intervalNanos;
// Push the first available permit slot up to the burst window if necessary.
long firstAvailable = Math.max(prev, nowNanos - burstNanos);
// Advance the configured interval starting from the bounded previous permit slot.
if (nextAvailable.compareAndSet(prev, firstAvailable + interval))
// If the time now is before the first available slot, return the delay.
return delayUnit.convert(Math.max(0, firstAvailable - nowNanos), TimeUnit.NANOSECONDS);
}
}
/**
* Reserves a single permit slot on the timeline, but only if one is available.
*
* @return true if a permit is available, false if one is not
*/
public boolean tryReserve()
{
long nowNanos = nanosElapsed();
for (;;)
{
long prev = nextAvailable.get();
long interval = this.intervalNanos;
// Push the first available permit slot up to the burst window if necessary.
long firstAvailable = Math.max(prev, nowNanos - burstNanos);
// If we haven't reached the time for the first available permit, we've failed to reserve.
if (nowNanos < firstAvailable)
return false;
// Advance the configured interval starting from the bounded previous permit slot.
// If another thread has already taken the next slot, retry.
if (nextAvailable.compareAndSet(prev, firstAvailable + interval))
return true;
}
}
@VisibleForTesting
public long getIntervalNanos()
{
return intervalNanos;
}
@VisibleForTesting
public long getStartedNanos()
{
return startedNanos;
}
private long nanosElapsed()
{
return ticker.read() - startedNanos;
}
public static final NonBlockingRateLimiter NO_OP_LIMITER = new NonBlockingRateLimiter(toIntExact(NANOS_PER_SECOND))
{
@Override
public long reserveAndGetDelay(TimeUnit delayUnit) {
return 0;
}
@Override
public boolean tryReserve()
{
return true;
}
};
}

9
test/burn/org/apache/cassandra/transport/BurnTestUtil.java
View File

@ -35,6 +35,9 @@ import org.apache.cassandra.net.AbstractMessageHandler;
import org.apache.cassandra.net.ResourceLimits;
import org.apache.cassandra.transport.messages.QueryMessage;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.concurrent.NonBlockingRateLimiter;
import static org.apache.cassandra.utils.concurrent.NonBlockingRateLimiter.NO_OP_LIMITER;
public class BurnTestUtil
{
@ -143,6 +146,12 @@ public class BurnTestUtil
return delegate.endpointWaitQueue();
}
@Override
public NonBlockingRateLimiter requestRateLimiter()
{
return NO_OP_LIMITER;
}
public void release()
{
delegate.release();

81
test/burn/org/apache/cassandra/transport/SimpleClientPerfTest.java
View File

@ -23,12 +23,14 @@ import java.net.ServerSocket;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.commons.math.stat.descriptive.DescriptiveStatistics;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@ -41,11 +43,13 @@ import org.apache.cassandra.auth.AllowAllAuthorizer;
import org.apache.cassandra.auth.AllowAllNetworkAuthorizer;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.metrics.ClientMetrics;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.transport.messages.QueryMessage;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.AssertUtil;
import org.apache.cassandra.utils.Throwables;
import static org.apache.cassandra.transport.BurnTestUtil.SizeCaps;
import static org.apache.cassandra.transport.BurnTestUtil.generateQueryMessage;
@ -57,7 +61,7 @@ public class SimpleClientPerfTest
@Parameterized.Parameter
public ProtocolVersion version;
@Parameterized.Parameters()
@Parameterized.Parameters(name="{0}")
public static Collection<Object[]> versions()
{
return ProtocolVersion.SUPPORTED.stream()
@ -90,6 +94,7 @@ public class SimpleClientPerfTest
}
}
@SuppressWarnings({"IOResourceOpenedButNotSafelyClosed", "resource"})
@Test
public void measureSmall() throws Throwable
{
@ -102,6 +107,7 @@ public class SimpleClientPerfTest
version);
}
@SuppressWarnings({"IOResourceOpenedButNotSafelyClosed", "resource"})
@Test
public void measureSmallWithCompression() throws Throwable
{
@ -114,6 +120,7 @@ public class SimpleClientPerfTest
version);
}
@SuppressWarnings({"IOResourceOpenedButNotSafelyClosed", "resource"})
@Test
public void measureLarge() throws Throwable
{
@ -126,6 +133,7 @@ public class SimpleClientPerfTest
version);
}
@SuppressWarnings({"IOResourceOpenedButNotSafelyClosed", "resource"})
@Test
public void measureLargeWithCompression() throws Throwable
{
@ -138,6 +146,7 @@ public class SimpleClientPerfTest
version);
}
@SuppressWarnings({"UnstableApiUsage", "UseOfSystemOutOrSystemErr", "ResultOfMethodCallIgnored"})
public void perfTest(SizeCaps requestCaps, SizeCaps responseCaps, AssertUtil.ThrowingSupplier<SimpleClient> clientSupplier, ProtocolVersion version) throws Throwable
{
ResultMessage.Rows response = generateRows(0, responseCaps);
@ -190,7 +199,9 @@ public class SimpleClientPerfTest
AtomicBoolean measure = new AtomicBoolean(false);
DescriptiveStatistics stats = new DescriptiveStatistics();
Lock lock = new ReentrantLock();
RateLimiter limiter = RateLimiter.create(2000);
AtomicLong overloadedExceptions = new AtomicLong(0);
// TODO: exercise client -> server large messages
for (int t = 0; t < threads; t++)
{
@ -203,24 +214,53 @@ public class SimpleClientPerfTest
for (int j = 0; j < 1; j++)
messages.add(requestMessage);
if (measure.get())
{
long nanoStart = System.nanoTime();
client.execute(messages);
long diff = System.nanoTime() - nanoStart;
if (measure.get())
{
try
{
limiter.acquire();
long nanoStart = System.nanoTime();
client.execute(messages);
long elapsed = System.nanoTime() - nanoStart;
lock.lock();
try
{
stats.addValue(TimeUnit.MICROSECONDS.toMillis(diff));
lock.lock();
try
{
stats.addValue(TimeUnit.NANOSECONDS.toMicros(elapsed));
}
finally
{
lock.unlock();
}
}
catch (RuntimeException e)
{
if (Throwables.anyCauseMatches(e, cause -> cause instanceof OverloadedException))
{
overloadedExceptions.incrementAndGet();
}
else
{
throw e;
}
}
}
finally
else
{
lock.unlock();
try
{
limiter.acquire();
client.execute(messages); // warm-up
}
catch (RuntimeException e)
{
// Ignore overloads during warmup...
if (!Throwables.anyCauseMatches(e, cause -> cause instanceof OverloadedException))
{
throw e;
}
}
}
}
else
client.execute(messages); // warm-up
}
}
catch (Throwable e)
@ -240,14 +280,19 @@ public class SimpleClientPerfTest
System.out.println("requestSize = " + requestSize);
System.out.println("responseSize = " + responseSize);
System.out.println("Latencies (in microseconds)");
System.out.println("Elements: " + stats.getN());
System.out.println("Mean: " + stats.getMean());
System.out.println("Variance: " + stats.getVariance());
System.out.println("Median: " + stats.getPercentile(0.5));
System.out.println("90p: " + stats.getPercentile(0.90));
System.out.println("95p: " + stats.getPercentile(0.95));
System.out.println("99p: " + stats.getPercentile(0.99));
System.out.println("Max: " + stats.getMax());
System.out.println("Failed due to overload: " + overloadedExceptions.get());
server.stop();
}
}

35
test/unit/org/apache/cassandra/cql3/CQLTester.java
View File

@ -50,6 +50,7 @@ import org.junit.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.codahale.metrics.Gauge;
import com.datastax.driver.core.*;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.ResultSet;
@ -63,6 +64,7 @@ import org.apache.cassandra.index.SecondaryIndexManager;
import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.metrics.CassandraMetricsRegistry;
import org.apache.cassandra.metrics.ClientMetrics;
import org.apache.cassandra.schema.*;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -90,7 +92,10 @@ import org.apache.cassandra.utils.JMXServerUtils;
import static com.datastax.driver.core.SocketOptions.DEFAULT_CONNECT_TIMEOUT_MILLIS;
import static com.datastax.driver.core.SocketOptions.DEFAULT_READ_TIMEOUT_MILLIS;
import static junit.framework.Assert.assertNotNull;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
* Base class for CQL tests.
@ -941,6 +946,23 @@ public abstract class CQLTester
Assert.assertEquals(expectedArgTypes != null ? Arrays.asList(expectedArgTypes) : null, schemaChange.argTypes);
}
protected static void assertWarningsContain(Message.Response response, String message)
{
List<String> warnings = response.getWarnings();
Assert.assertNotNull(warnings);
assertTrue(warnings.stream().anyMatch(s -> s.contains(message)));
}
protected static void assertNoWarningContains(Message.Response response, String message)
{
List<String> warnings = response.getWarnings();
if (warnings != null)
{
assertFalse(warnings.stream().anyMatch(s -> s.contains(message)));
}
}
protected static ResultMessage schemaChange(String query)
{
try
@ -1806,6 +1828,17 @@ public abstract class CQLTester
return clusters.get(protocolVersion).getMetadata().newTupleType(types);
}
@SuppressWarnings({ "rawtypes", "unchecked" })
protected static Gauge<Integer> getPausedConnectionsGauge()
{
String metricName = "org.apache.cassandra.metrics.Client.PausedConnections";
Map<String, Gauge> metrics = CassandraMetricsRegistry.Metrics.getGauges((name, metric) -> name.equals(metricName));
if (metrics.size() != 1)
fail(String.format("Expected a single registered metric for paused client connections, found %s",
metrics.size()));
return metrics.get(metricName);
}
// Attempt to find an AbstracType from a value (for serialization/printing sake).
// Will work as long as we use types we know of, which is good enough for testing
private static AbstractType typeFor(Object value)

2
test/unit/org/apache/cassandra/net/ResourceLimitsTest.java
View File

@ -23,9 +23,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.LongFunction;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.KillerForTests;
import org.junit.Assert;

11
test/unit/org/apache/cassandra/transport/CQLConnectionTest.java
View File

@ -30,6 +30,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.*;
import org.apache.cassandra.transport.ClientResourceLimits.Overload;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
@ -55,10 +56,12 @@ import org.apache.cassandra.transport.CQLMessageHandler.MessageConsumer;
import org.apache.cassandra.transport.messages.*;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import org.apache.cassandra.utils.concurrent.NonBlockingRateLimiter;
import static org.apache.cassandra.config.EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED;
import static org.apache.cassandra.net.FramingTest.randomishBytes;
import static org.apache.cassandra.transport.Flusher.MAX_FRAMED_PAYLOAD_SIZE;
import static org.apache.cassandra.utils.concurrent.NonBlockingRateLimiter.NO_OP_LIMITER;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@ -622,7 +625,7 @@ public class CQLConnectionTest
this.frameEncoder = frameEncoder;
}
public void accept(Channel channel, Message.Request message, Dispatcher.FlushItemConverter toFlushItem)
public void accept(Channel channel, Message.Request message, Dispatcher.FlushItemConverter toFlushItem, Overload backpressure)
{
if (flusher == null)
flusher = new SimpleClient.SimpleFlusher(frameEncoder);
@ -755,6 +758,12 @@ public class CQLConnectionTest
return delegate.endpointWaitQueue();
}
@Override
public NonBlockingRateLimiter requestRateLimiter()
{
return NO_OP_LIMITER;
}
public void release()
{
delegate.release();

99
test/unit/org/apache/cassandra/transport/ClientResourceLimitsTest.java
View File

@ -19,7 +19,6 @@
package org.apache.cassandra.transport;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
@ -29,9 +28,9 @@ import java.util.function.Supplier;
import com.google.common.collect.ImmutableList;
import com.google.common.primitives.Ints;
import org.apache.cassandra.service.StorageService;
import org.junit.*;
import com.codahale.metrics.Gauge;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.QueryOptions;
@ -40,13 +39,11 @@ import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.virtual.*;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.metrics.CassandraMetricsRegistry;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.transport.messages.QueryMessage;
import org.apache.cassandra.utils.FBUtilities;
import org.awaitility.Awaitility;
import static org.apache.cassandra.Util.spinAssertEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
@ -55,19 +52,18 @@ import static org.junit.Assert.fail;
public class ClientResourceLimitsTest extends CQLTester
{
private static final long LOW_LIMIT = 600L;
private static final long HIGH_LIMIT = 5000000000L;
private static final QueryOptions V5_DEFAULT_OPTIONS = QueryOptions.create(
QueryOptions.DEFAULT.getConsistency(),
QueryOptions.DEFAULT.getValues(),
QueryOptions.DEFAULT.skipMetadata(),
QueryOptions.DEFAULT.getPageSize(),
QueryOptions.DEFAULT.getPagingState(),
QueryOptions.DEFAULT.getSerialConsistency(),
ProtocolVersion.V5,
KEYSPACE);
private static final QueryOptions V5_DEFAULT_OPTIONS =
QueryOptions.create(QueryOptions.DEFAULT.getConsistency(),
QueryOptions.DEFAULT.getValues(),
QueryOptions.DEFAULT.skipMetadata(),
QueryOptions.DEFAULT.getPageSize(),
QueryOptions.DEFAULT.getPagingState(),
QueryOptions.DEFAULT.getSerialConsistency(),
ProtocolVersion.V5,
KEYSPACE);
@BeforeClass
public static void setUp()
@ -75,7 +71,7 @@ public class ClientResourceLimitsTest extends CQLTester
DatabaseDescriptor.setNativeTransportReceiveQueueCapacityInBytes(1);
DatabaseDescriptor.setNativeTransportMaxConcurrentRequestsInBytesPerIp(LOW_LIMIT);
DatabaseDescriptor.setNativeTransportMaxConcurrentRequestsInBytes(LOW_LIMIT);
// The driver control connections would send queries that might interfere with the tests.
requireNetworkWithoutDriver();
}
@ -124,7 +120,7 @@ public class ClientResourceLimitsTest extends CQLTester
}
}
@SuppressWarnings("resource")
@SuppressWarnings({"resource", "SameParameterValue"})
private SimpleClient client(boolean throwOnOverload, int largeMessageThreshold)
{
try
@ -184,27 +180,34 @@ public class ClientResourceLimitsTest extends CQLTester
(provider) -> provider.endpointWaitQueue().signal());
}
private void backPressureTest(Runnable limitLifter, Consumer<ClientResourceLimits.ResourceProvider> signaller)
throws Throwable
private void backPressureTest(Runnable limitLifter, Consumer<ClientResourceLimits.ResourceProvider> signaller) throws Throwable
{
final AtomicReference<Exception> error = new AtomicReference<>();
final CountDownLatch started = new CountDownLatch(1);
final CountDownLatch complete = new CountDownLatch(1);
try(SimpleClient client = client(false))
try (SimpleClient client = client(false))
{
QueryMessage queryMessage = new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)",
V5_DEFAULT_OPTIONS);
client.execute(queryMessage);
// There should be no paused client connections yet
Gauge<Integer> pausedConnections = getPausedConnectionsGauge();
int before = pausedConnections.getValue();
// The first query does not trigger backpressure/pause the connection:
QueryMessage queryMessage =
new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)", V5_DEFAULT_OPTIONS);
Message.Response belowThresholdResponse = client.execute(queryMessage);
assertEquals(0, getPausedConnectionsGauge().getValue().intValue());
assertNoWarningContains(belowThresholdResponse, "bytes in flight");
// A second query triggers backpressure but is allowed to complete...
Message.Response aboveThresholdResponse = client.execute(queryMessage());
assertEquals(1, getPausedConnectionsGauge().getValue().intValue());
assertWarningsContain(aboveThresholdResponse, "bytes in flight");
// ...and a third request is paused.
final AtomicReference<Message.Response> response = new AtomicReference<>();
Thread t = new Thread(() -> {
try
{
started.countDown();
client.execute(queryMessage());
response.set(client.execute(queryMessage()));
complete.countDown();
}
catch (Exception e)
@ -216,19 +219,13 @@ public class ClientResourceLimitsTest extends CQLTester
});
t.start();
// When the client attempts to execute the second query, the backpressure
// mechanism should report the client connection is paused
assertTrue(started.await(5, TimeUnit.SECONDS));
spinAssertEquals("Timed out after waiting 5 seconds for paused " +
"connections metric to increment due to backpressure",
before + 1, pausedConnections::getValue, 5, TimeUnit.SECONDS);
// verify the request hasn't completed
assertFalse(complete.await(1, TimeUnit.SECONDS));
// backpressure has been applied, if we increase the limits of the exhausted reserve and signal
// the appropriate WaitQueue, it should be released and the client request will complete
limitLifter.run();
// We need a ResourceProvider to get access to the WaitQueue
ClientResourceLimits.Allocator allocator = ClientResourceLimits.getAllocatorForEndpoint(FBUtilities.getJustLocalAddress());
ClientResourceLimits.ResourceProvider queueHandle = new ClientResourceLimits.ResourceProvider.Default(allocator);
@ -237,8 +234,10 @@ public class ClientResourceLimitsTest extends CQLTester
// SimpleClient has a 10 second timeout, so if we have to wait
// longer than that assume that we're not going to receive a
// reply. If all's well, the completion should happen immediately
assertTrue(complete.await(11, TimeUnit.SECONDS));
assertTrue(complete.await(SimpleClient.TIMEOUT_SECONDS + 1, TimeUnit.SECONDS));
assertNull(error.get());
assertEquals(0, getPausedConnectionsGauge().getValue().intValue());
assertNoWarningContains(response.get(), "bytes in flight");
}
}
@ -304,17 +303,6 @@ public class ClientResourceLimitsTest extends CQLTester
return new QueryMessage(query.toString(), V5_DEFAULT_OPTIONS);
}
@SuppressWarnings({"rawtypes", "unchecked"})
private Gauge<Integer> getPausedConnectionsGauge()
{
String metricName = "org.apache.cassandra.metrics.Client.PausedConnections";
Map<String, Gauge> metrics = CassandraMetricsRegistry.Metrics.getGauges((name, metric) -> name.equals(metricName));
if (metrics.size() != 1)
fail(String.format("Expected a single registered metric for paused client connections, found %s",
metrics.size()));
return metrics.get(metricName);
}
@Test
public void testQueryUpdatesConcurrentMetricsUpdate() throws Throwable
{
@ -469,4 +457,23 @@ public class ClientResourceLimitsTest extends CQLTester
client.close();
}
}
@Test
public void shouldChangeRequestsPerSecondAtRuntime()
{
StorageService.instance.setNativeTransportMaxRequestsPerSecond(100);
assertEquals(100, ClientResourceLimits.getNativeTransportMaxRequestsPerSecond(), 0);
assertEquals(100, ClientResourceLimits.GLOBAL_REQUEST_LIMITER.getRate(), 0);
assertEquals(100, StorageService.instance.getNativeTransportMaxRequestsPerSecond());
StorageService.instance.setNativeTransportMaxRequestsPerSecond(1000);
assertEquals(1000, ClientResourceLimits.getNativeTransportMaxRequestsPerSecond(), 0);
assertEquals(1000, ClientResourceLimits.GLOBAL_REQUEST_LIMITER.getRate(), 0);
assertEquals(1000, StorageService.instance.getNativeTransportMaxRequestsPerSecond());
StorageService.instance.setNativeTransportMaxRequestsPerSecond(500);
assertEquals(500, ClientResourceLimits.getNativeTransportMaxRequestsPerSecond(), 0);
assertEquals(500, ClientResourceLimits.GLOBAL_REQUEST_LIMITER.getRate(), 0);
assertEquals(500, StorageService.instance.getNativeTransportMaxRequestsPerSecond());
}
}

309
test/unit/org/apache/cassandra/transport/RateLimitingTest.java Normal file
View File

@ -0,0 +1,309 @@
/*
* 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.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import com.google.common.base.Ticker;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.StorageService;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.transport.messages.QueryMessage;
import org.apache.cassandra.utils.Throwables;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.apache.cassandra.Util.spinAssertEquals;
import static org.apache.cassandra.transport.ProtocolVersion.V4;
@SuppressWarnings("UnstableApiUsage")
@RunWith(Parameterized.class)
public class RateLimitingTest extends CQLTester
{
public static final String BACKPRESSURE_WARNING_SNIPPET = "Request breached global limit";
private static final int LARGE_PAYLOAD_THRESHOLD_BYTES = 1000;
private static final int OVERLOAD_PERMITS_PER_SECOND = 1;
@Parameterized.Parameter
public ProtocolVersion version;
@Parameterized.Parameters(name="{0}")
public static Collection<Object[]> versions()
{
return ProtocolVersion.SUPPORTED.stream()
.map(v -> new Object[]{v})
.collect(Collectors.toList());
}
private AtomicLong tick;
private Ticker ticker;
@BeforeClass
public static void setup()
{
// If we don't exceed the queue capacity, we won't actually use the global/endpoint
// bytes-in-flight limits, and the assertions we make below around releasing them would be useless.
DatabaseDescriptor.setNativeTransportReceiveQueueCapacityInBytes(1);
// The driver control connections would send queries that might interfere with the tests.
requireNetworkWithoutDriver();
}
@Before
public void resetLimits()
{
// Reset to the original start time in case a test advances the clock.
tick = new AtomicLong(ClientResourceLimits.GLOBAL_REQUEST_LIMITER.getStartedNanos());
ticker = new Ticker()
{
@Override
public long read()
{
return tick.get();
}
};
ClientResourceLimits.setGlobalLimit(Long.MAX_VALUE);
}
@Test
public void shouldThrowOnOverloadSmallMessages() throws Exception
{
int payloadSize = LARGE_PAYLOAD_THRESHOLD_BYTES / 4;
testOverload(payloadSize, true);
}
@Test
public void shouldThrowOnOverloadLargeMessages() throws Exception
{
int payloadSize = LARGE_PAYLOAD_THRESHOLD_BYTES * 2;
testOverload(payloadSize, true);
}
@Test
public void shouldBackpressureSmallMessages() throws Exception
{
int payloadSize = LARGE_PAYLOAD_THRESHOLD_BYTES / 4;
testOverload(payloadSize, false);
}
@Test
public void shouldBackpressureLargeMessages() throws Exception
{
int payloadSize = LARGE_PAYLOAD_THRESHOLD_BYTES * 2;
testOverload(payloadSize, false);
}
@Test
public void shouldReleaseSmallMessageOnBytesInFlightOverload() throws Exception
{
testBytesInFlightOverload(LARGE_PAYLOAD_THRESHOLD_BYTES / 4);
}
@Test
public void shouldReleaseLargeMessageOnBytesInFlightOverload() throws Exception
{
testBytesInFlightOverload(LARGE_PAYLOAD_THRESHOLD_BYTES * 2);
}
private void testBytesInFlightOverload(int payloadSize) throws Exception
{
try (SimpleClient client = client().connect(false, true))
{
StorageService.instance.setNativeTransportRateLimitingEnabled(false);
QueryMessage queryMessage = new QueryMessage("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".atable (pk int PRIMARY KEY, v text)", queryOptions());
client.execute(queryMessage);
StorageService.instance.setNativeTransportRateLimitingEnabled(true);
ClientResourceLimits.GLOBAL_REQUEST_LIMITER.setRate(OVERLOAD_PERMITS_PER_SECOND, ticker);
ClientResourceLimits.setGlobalLimit(1);
try
{
// The first query takes the one available permit, but should fail on the bytes in flight limit.
client.execute(queryMessage(payloadSize));
}
catch (RuntimeException e)
{
assertTrue(Throwables.anyCauseMatches(e, cause -> cause instanceof OverloadedException));
}
}
finally
{
// Sanity check bytes in flight limiter.
assertEquals(0, ClientResourceLimits.getCurrentGlobalUsage());
StorageService.instance.setNativeTransportRateLimitingEnabled(false);
}
}
private void testOverload(int payloadSize, boolean throwOnOverload) throws Exception
{
try (SimpleClient client = client().connect(false, throwOnOverload))
{
StorageService.instance.setNativeTransportRateLimitingEnabled(false);
QueryMessage queryMessage = new QueryMessage("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".atable (pk int PRIMARY KEY, v text)", queryOptions());
client.execute(queryMessage);
StorageService.instance.setNativeTransportRateLimitingEnabled(true);
ClientResourceLimits.GLOBAL_REQUEST_LIMITER.setRate(OVERLOAD_PERMITS_PER_SECOND, ticker);
if (throwOnOverload)
testThrowOnOverload(payloadSize, client);
else
{
testBackpressureOnOverload(payloadSize, client);
}
}
finally
{
// Sanity the check bytes in flight limiter.
assertEquals(0, ClientResourceLimits.getCurrentGlobalUsage());
StorageService.instance.setNativeTransportRateLimitingEnabled(false);
}
}
private void testBackpressureOnOverload(int payloadSize, SimpleClient client) throws Exception
{
// The first query takes the one available permit.
Message.Response firstResponse = client.execute(queryMessage(payloadSize));
assertEquals(0, getPausedConnectionsGauge().getValue().intValue());
assertNoWarningContains(firstResponse, BACKPRESSURE_WARNING_SNIPPET);
// The second query activates backpressure.
long overloadQueryStartTime = System.currentTimeMillis();
Message.Response response = client.execute(queryMessage(payloadSize));
// V3 does not support client warnings, but otherwise we should get one for this query.
if (version.isGreaterOrEqualTo(V4))
assertWarningsContain(response, BACKPRESSURE_WARNING_SNIPPET);
AtomicReference<Throwable> error = new AtomicReference<>();
CountDownLatch started = new CountDownLatch(1);
CountDownLatch complete = new CountDownLatch(1);
AtomicReference<Message.Response> pausedQueryResponse = new AtomicReference<>();
Thread queryRunner = new Thread(() ->
{
try
{
started.countDown();
pausedQueryResponse.set(client.execute(queryMessage(payloadSize)));
complete.countDown();
}
catch (Throwable t)
{
error.set(t);
}
});
// Advance the rater limiter so that this query will see an available permit. This also
// means it should not produce a client warning, which we verify below.
// (Note that we advance 2 intervals for the 2 prior queries.)
tick.addAndGet(2 * ClientResourceLimits.GLOBAL_REQUEST_LIMITER.getIntervalNanos());
queryRunner.start();
// ...and the request should complete without error.
assertTrue(complete.await(SimpleClient.TIMEOUT_SECONDS + 1, TimeUnit.SECONDS));
assertNull(error.get());
assertNoWarningContains(pausedQueryResponse.get(), BACKPRESSURE_WARNING_SNIPPET);
// At least the number of milliseconds in the permit interval should already have elapsed
// since the start of the query that pauses the connection.
double permitIntervalMillis = (double) TimeUnit.SECONDS.toMillis(1L) / OVERLOAD_PERMITS_PER_SECOND;
long sinceQueryStarted = System.currentTimeMillis() - overloadQueryStartTime;
long remainingMillis = ((long) permitIntervalMillis) - sinceQueryStarted;
assertTrue("Query completed before connection unpause!", remainingMillis <= 0);
spinAssertEquals("Timed out after waiting 5 seconds for paused connections metric to normalize.",
0, () -> getPausedConnectionsGauge().getValue(), 5, TimeUnit.SECONDS);
}
private void testThrowOnOverload(int payloadSize, SimpleClient client)
{
// The first query takes the one available permit...
client.execute(queryMessage(payloadSize));
try
{
// ...and the second breaches the limit....
client.execute(queryMessage(payloadSize));
}
catch (RuntimeException e)
{
assertTrue(Throwables.anyCauseMatches(e, cause -> cause instanceof OverloadedException));
}
// Advance the timeline and verify that we can take a permit again.
// (Note that we don't take one when we throw on overload.)
tick.addAndGet(ClientResourceLimits.GLOBAL_REQUEST_LIMITER.getIntervalNanos());
client.execute(queryMessage(payloadSize));
}
private QueryMessage queryMessage(int length)
{
StringBuilder query = new StringBuilder("INSERT INTO " + KEYSPACE + ".atable (pk, v) VALUES (1, '");
for (int i = 0; i < length; i++)
{
query.append('a');
}
query.append("')");
return new QueryMessage(query.toString(), queryOptions());
}
private SimpleClient client()
{
return SimpleClient.builder(nativeAddr.getHostAddress(), nativePort)
.protocolVersion(version)
.useBeta()
.largeMessageThreshold(LARGE_PAYLOAD_THRESHOLD_BYTES)
.build();
}
private QueryOptions queryOptions()
{
return QueryOptions.create(QueryOptions.DEFAULT.getConsistency(),
QueryOptions.DEFAULT.getValues(),
QueryOptions.DEFAULT.skipMetadata(),
QueryOptions.DEFAULT.getPageSize(),
QueryOptions.DEFAULT.getPagingState(),
QueryOptions.DEFAULT.getSerialConsistency(),
version,
KEYSPACE);
}
}

121
test/unit/org/apache/cassandra/utils/concurrent/NonBlockingRateLimiterTest.java Normal file
View File

@ -0,0 +1,121 @@
/*
* 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.utils.concurrent;
import com.google.common.base.Ticker;
import org.junit.Before;
import org.junit.Test;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@SuppressWarnings("UnstableApiUsage")
public class NonBlockingRateLimiterTest
{
private static final AtomicLong CLOCK = new AtomicLong(0);
private static final TimeUnit DELAY_UNIT = TimeUnit.NANOSECONDS;
private static final Ticker TICKER = new Ticker()
{
@Override
public long read() {
return CLOCK.get();
}
};
@Before
public void resetTicker()
{
CLOCK.set(0);
}
@Test
public void testUnconditionalReservation()
{
NonBlockingRateLimiter limiter = new NonBlockingRateLimiter(4, 0, TICKER);
long oneSecond = DELAY_UNIT.convert(1, TimeUnit.SECONDS);
long oneDelay = oneSecond / 4;
// Delays should begin accumulating without any ticker movement...
assertEquals(0, limiter.reserveAndGetDelay(DELAY_UNIT));
assertEquals(oneDelay, limiter.reserveAndGetDelay(DELAY_UNIT));
assertEquals(oneDelay * 2, limiter.reserveAndGetDelay(DELAY_UNIT));
assertEquals(oneDelay * 3, limiter.reserveAndGetDelay(DELAY_UNIT));
// ...but should be gone after advancing enough to free up a permit.
CLOCK.addAndGet(NonBlockingRateLimiter.NANOS_PER_SECOND);
assertEquals(0, limiter.reserveAndGetDelay(DELAY_UNIT));
}
@Test
public void testConditionalReservation()
{
NonBlockingRateLimiter limiter = new NonBlockingRateLimiter(1, 0, TICKER);
// Take the available permit, but then fail a subsequent attempt.
assertTrue(limiter.tryReserve());
assertFalse(limiter.tryReserve());
// We only need to advance one second, as the second attempt should not get a permit.
CLOCK.addAndGet(NonBlockingRateLimiter.NANOS_PER_SECOND);
assertTrue(limiter.tryReserve());
}
@Test
public void testBurstPermitConsumption()
{
// Create a limiter that produces 2 permits/second and allows 1-second bursts.
NonBlockingRateLimiter limiter = new NonBlockingRateLimiter(1, NonBlockingRateLimiter.DEFAULT_BURST_NANOS, TICKER);
// Advance the clock to create a 1-second idle period, which makes one burst permit available.
CLOCK.addAndGet(NonBlockingRateLimiter.NANOS_PER_SECOND);
// Take the burst permit.
assertTrue(limiter.tryReserve());
// Take the "normal" permit.
assertTrue(limiter.tryReserve());
// Then fail, as we've consumed both.
assertFalse(limiter.tryReserve());
// Advance 1 interval again...
CLOCK.addAndGet(NonBlockingRateLimiter.NANOS_PER_SECOND);
// ...and only one permit should be available, as we've reached a steady state.
assertTrue(limiter.tryReserve());
assertFalse(limiter.tryReserve());
}
@Test(expected = IllegalArgumentException.class)
public void testMaximumRate()
{
new NonBlockingRateLimiter(Integer.MAX_VALUE, 0, Ticker.systemTicker());
}
@Test(expected = IllegalArgumentException.class)
public void testMinimumRate()
{
new NonBlockingRateLimiter(-1, 0, Ticker.systemTicker());
}
}