You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2018/08/24 15:07:48 UTC
cassandra git commit: 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
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 <ya...@iki.fi>
Authored: Tue May 8 15:40:54 2018 +0300
Committer: Benedict Elliott Smith <be...@apple.com>
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: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org