add inter_dc_tcp_nodelay option

patch by Marcus Eriksson; reviewed by jbellis for CASSANDRA-5148

CHANGES.txt

@@ -1,4 +1,5 @@
 1.2.1
+ * add inter_dc_tcp_nodelay setting (CASSANDRA-5148)
  * re-allow wrapping ranges for start_token/end_token range pairing (CASSANDRA-5106)
  * fix validation compaction of empty rows (CASSADRA-5136)
  * nodetool methods to enable/disable hint storage/delivery (CASSANDRA-4750)

conf/cassandra.yaml

@@ -643,3 +643,9 @@ client_encryption_options:
 #          dc   - traffic between different datacenters is compressed
 #          none - nothing is compressed.
 internode_compression: all
+
+# Enable or disable tcp_nodelay for inter-dc communication.
+# Disabling it will result in larger (but fewer) network packets being sent,
+# reducing overhead from the TCP protocol itself, at the cost of increasing
+# latency if you block for cross-datacenter responses.
+inter_dc_tcp_nodelay: true

src/java/org/apache/cassandra/config/Config.java

@@ -162,6 +162,8 @@ public class Config
     public String row_cache_provider = SerializingCacheProvider.class.getSimpleName();
     public boolean populate_io_cache_on_flush = false;
 
+    public boolean inter_dc_tcp_nodelay = true;
+
     private static boolean loadYaml = true;
     private static boolean outboundBindAny = false;
 

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -1256,4 +1256,9 @@ public class DatabaseDescriptor
     {
         return conf.internode_compression;
     }
+
+    public static boolean getInterDCTcpNoDelay()
+    {
+        return conf.inter_dc_tcp_nodelay;
+    }
 }

src/java/org/apache/cassandra/net/OutboundTcpConnection.java

@@ -263,7 +263,14 @@ public class OutboundTcpConnection extends Thread
             {
                 socket = poolReference.newSocket();
                 socket.setKeepAlive(true);
-                socket.setTcpNoDelay(true);
+                if (isLocalDC(poolReference.endPoint()))
+                {
+                    socket.setTcpNoDelay(true);
+                }
+                else
+                {
+                    socket.setTcpNoDelay(DatabaseDescriptor.getInterDCTcpNoDelay());
+                }
                 out = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream(), 4096));
 
                 if (targetVersion >= MessagingService.VERSION_12)