From 96ef514917e5a4829dbe864104dbc08a7d0e0cec Mon Sep 17 00:00:00 2001 From: Michael Burman Date: Tue, 8 May 2018 15:40:54 +0300 Subject: [PATCH] Remove Netty timed batching and instead do the batch during next eventLoop invocation after a write has been enqueued. Old behavior can be enabled with setting native_transport_flush_in_batches_legacy patch by Michael Burman; reviewed by Benedict for CASSANDRA-13651 --- CHANGES.txt | 1 + conf/cassandra.yaml | 4 + .../org/apache/cassandra/config/Config.java | 1 + .../cassandra/config/DatabaseDescriptor.java | 6 ++ .../apache/cassandra/transport/Message.java | 76 +++++++++++++++---- .../apache/cassandra/transport/Server.java | 2 +- 6 files changed, 76 insertions(+), 14 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 75f41e89c7..c64881a6d7 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0 + * Flush netty client messages immediately by default (CASSANDRA-13651) * Improve read repair blocking behavior (CASSANDRA-10726) * Add a virtual table to expose settings (CASSANDRA-14573) * Fix up chunk cache handling of metrics (CASSANDRA-14628) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index c7c27851f3..12f5ab2911 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -1222,3 +1222,7 @@ audit_logging_options: # If enabled, diagnostic events can be helpful for troubleshooting operational issues. Emitted events contain details # on internal state and temporal relationships across events, accessible by clients via JMX. diagnostic_events_enabled: false + +# Define use of legacy delayed flusher for replies to TCP connections. This will increase latency, but might be beneficial for +# legacy use-cases where only a single connection is used for each Cassandra node. Default is false. +#native_transport_flush_in_batches_legacy: false diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 46dbc1c93a..f6eefc3390 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -152,6 +152,7 @@ public class Config public int native_transport_max_frame_size_in_mb = 256; public volatile long native_transport_max_concurrent_connections = -1L; public volatile long native_transport_max_concurrent_connections_per_ip = -1L; + public boolean native_transport_flush_in_batches_legacy = false; /** * Max size of values in SSTables, in MegaBytes. diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index aa5ca92833..eae6fc97d6 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -1866,6 +1866,11 @@ public class DatabaseDescriptor conf.native_transport_max_concurrent_connections_per_ip = native_transport_max_concurrent_connections_per_ip; } + public static boolean useNativeTransportLegacyFlusher() + { + return conf.native_transport_flush_in_batches_legacy; + } + public static double getCommitLogSyncGroupWindow() { return conf.commitlog_sync_group_window_in_ms; @@ -2644,4 +2649,5 @@ public class DatabaseDescriptor { conf.corrupted_tombstone_strategy = strategy; } + } diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java index 531909f310..413c94a829 100644 --- a/src/java/org/apache/cassandra/transport/Message.java +++ b/src/java/org/apache/cassandra/transport/Message.java @@ -428,26 +428,38 @@ public abstract class Message } } - private static final class Flusher implements Runnable + private static abstract class Flusher implements Runnable { final EventLoop eventLoop; final ConcurrentLinkedQueue queued = new ConcurrentLinkedQueue<>(); - final AtomicBoolean running = new AtomicBoolean(false); + final AtomicBoolean scheduled = new AtomicBoolean(false); final HashSet channels = new HashSet<>(); final List flushed = new ArrayList<>(); - int runsSinceFlush = 0; - int runsWithNoWork = 0; - private Flusher(EventLoop eventLoop) - { - this.eventLoop = eventLoop; - } + void start() { - if (!running.get() && running.compareAndSet(false, true)) + if (!scheduled.get() && scheduled.compareAndSet(false, true)) { this.eventLoop.execute(this); } } + + public Flusher(EventLoop eventLoop) + { + this.eventLoop = eventLoop; + } + } + + private static final class LegacyFlusher extends Flusher + { + int runsSinceFlush = 0; + int runsWithNoWork = 0; + + private LegacyFlusher(EventLoop eventLoop) + { + super(eventLoop); + } + public void run() { @@ -484,8 +496,8 @@ public abstract class Message // either reschedule or cancel if (++runsWithNoWork > 5) { - running.set(false); - if (queued.isEmpty() || !running.compareAndSet(false, true)) + scheduled.set(false); + if (queued.isEmpty() || !scheduled.compareAndSet(false, true)) return; } } @@ -494,11 +506,48 @@ public abstract class Message } } + private static final class ImmediateFlusher extends Flusher + { + private ImmediateFlusher(EventLoop eventLoop) + { + super(eventLoop); + } + + public void run() + { + boolean doneWork = false; + FlushItem flush; + scheduled.set(false); + + while (null != (flush = queued.poll())) + { + channels.add(flush.ctx); + flush.ctx.write(flush.response, flush.ctx.voidPromise()); + flushed.add(flush); + doneWork = true; + } + + if (doneWork) + { + for (ChannelHandlerContext channel : channels) + channel.flush(); + for (FlushItem item : flushed) + item.sourceFrame.release(); + + channels.clear(); + flushed.clear(); + } + } + } + private static final ConcurrentMap flusherLookup = new ConcurrentHashMap<>(); - public Dispatcher() + private final boolean useLegacyFlusher; + + public Dispatcher(boolean useLegacyFlusher) { super(false); + this.useLegacyFlusher = useLegacyFlusher; } @Override @@ -548,7 +597,8 @@ public abstract class Message Flusher flusher = flusherLookup.get(loop); if (flusher == null) { - Flusher alt = flusherLookup.putIfAbsent(loop, flusher = new Flusher(loop)); + Flusher created = useLegacyFlusher ? new LegacyFlusher(loop) : new ImmediateFlusher(loop); + Flusher alt = flusherLookup.putIfAbsent(loop, flusher = created); if (alt != null) flusher = alt; } diff --git a/src/java/org/apache/cassandra/transport/Server.java b/src/java/org/apache/cassandra/transport/Server.java index 45146c4d99..0c4b7b8f6f 100644 --- a/src/java/org/apache/cassandra/transport/Server.java +++ b/src/java/org/apache/cassandra/transport/Server.java @@ -346,7 +346,7 @@ public class Server implements CassandraDaemon.Server private static final Frame.Compressor frameCompressor = new Frame.Compressor(); private static final Frame.Encoder frameEncoder = new Frame.Encoder(); private static final Message.ExceptionHandler exceptionHandler = new Message.ExceptionHandler(); - private static final Message.Dispatcher dispatcher = new Message.Dispatcher(); + private static final Message.Dispatcher dispatcher = new Message.Dispatcher(DatabaseDescriptor.useNativeTransportLegacyFlusher()); private static final ConnectionLimitHandler connectionLimitHandler = new ConnectionLimitHandler(); private final Server server;