diff --git a/CHANGES.txt b/CHANGES.txt index 8d184c6242..d774f6dc64 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -22,6 +22,7 @@ Merged from 3.0: * Fix corrupted static collection deletions in 3.0 <-> 2.{1,2} messages (CASSANDRA-14568) * Handle failures in parallelAllSSTableOperation (cleanup/upgradesstables/etc) (CASSANDRA-14657) * Improve TokenMetaData cache populating performance avoid long locking (CASSANDRA-14660) + * Backport: Flush netty client messages immediately (not by default) (CASSANDRA-13651) * Fix static column order for SELECT * wildcard queries (CASSANDRA-14638) * sstableloader should use discovered broadcast address to connect intra-cluster (CASSANDRA-14522) * Fix reading columns with non-UTF names from schema (CASSANDRA-14468) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 7d25bef93c..2cc119a907 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -1234,4 +1234,4 @@ back_pressure_strategy: # time and queue contention while iterating the backlog of messages. # An interval of 0 disables any wait time, which is the behavior of former Cassandra versions. # -# otc_backlog_expiration_interval_ms: 200 +# otc_backlog_expiration_interval_ms: 200 \ No newline at end of file diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index a01203c8f6..603d851f7c 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -156,6 +156,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 = true; @Deprecated public int thrift_max_message_length_in_mb = 16; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 8b92c5a9ef..21ab37011d 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -1817,6 +1817,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 getCommitLogSyncBatchWindow() { return conf.commitlog_sync_batch_window_in_ms; diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java index d1f8f28c36..de048a65f0 100644 --- a/src/java/org/apache/cassandra/transport/Message.java +++ b/src/java/org/apache/cassandra/transport/Message.java @@ -423,26 +423,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() { @@ -479,8 +491,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; } } @@ -489,11 +501,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 @@ -542,7 +591,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 212839c94d..1adcdd6af4 100644 --- a/src/java/org/apache/cassandra/transport/Server.java +++ b/src/java/org/apache/cassandra/transport/Server.java @@ -291,7 +291,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;