Repository: cassandra Updated Branches: refs/heads/trunk ffde38a25 -> 96ef51491
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 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/96ef5149 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/96ef5149 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/96ef5149 Branch: refs/heads/trunk Commit: 96ef514917e5a4829dbe864104dbc08a7d0e0cec Parents: ffde38a Author: Michael Burman <[email protected]> Authored: Tue May 8 15:40:54 2018 +0300 Committer: Benedict Elliott Smith <[email protected]> Committed: Fri Aug 24 16:07:39 2018 +0100 ---------------------------------------------------------------------- CHANGES.txt | 1 + conf/cassandra.yaml | 4 ++ .../org/apache/cassandra/config/Config.java | 1 + .../cassandra/config/DatabaseDescriptor.java | 6 ++ .../org/apache/cassandra/transport/Message.java | 76 ++++++++++++++++---- .../org/apache/cassandra/transport/Server.java | 2 +- 6 files changed, 76 insertions(+), 14 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/96ef5149/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 75f41e8..c64881a 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) http://git-wip-us.apache.org/repos/asf/cassandra/blob/96ef5149/conf/cassandra.yaml ---------------------------------------------------------------------- diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index c7c2785..12f5ab2 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 http://git-wip-us.apache.org/repos/asf/cassandra/blob/96ef5149/src/java/org/apache/cassandra/config/Config.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 46dbc1c..f6eefc3 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. http://git-wip-us.apache.org/repos/asf/cassandra/blob/96ef5149/src/java/org/apache/cassandra/config/DatabaseDescriptor.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index aa5ca92..eae6fc9 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; } + } http://git-wip-us.apache.org/repos/asf/cassandra/blob/96ef5149/src/java/org/apache/cassandra/transport/Message.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java index 531909f..413c94a 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<FlushItem> queued = new ConcurrentLinkedQueue<>(); - final AtomicBoolean running = new AtomicBoolean(false); + final AtomicBoolean scheduled = new AtomicBoolean(false); final HashSet<ChannelHandlerContext> channels = new HashSet<>(); final List<FlushItem> 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<EventLoop, Flusher> 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; } http://git-wip-us.apache.org/repos/asf/cassandra/blob/96ef5149/src/java/org/apache/cassandra/transport/Server.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/transport/Server.java b/src/java/org/apache/cassandra/transport/Server.java index 45146c4..0c4b7b8 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; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
