Allow configuration of internode socket buffer.

Patch by Michael Michalski and Jason Brown for CASSANDRA-3389, reviewed
by Jason Brown.

CHANGES.txt

@@ -5,6 +5,7 @@
  * Validate login for USE queries (CASSANDRA-5207)
  * cli: remove default username and password (CASSANDRA-5208)
  * configure populate_io_cache_on_flush per-CF (CASSANDRA-4694)
+ * allow configuration of internode socket buffer (CASSANDRA-3378)
 
 1.2.1
  * stream undelivered hints on decommission (CASSANDRA-5128)

conf/cassandra.yaml

@@ -370,6 +370,10 @@ rpc_server_type: sync
 # rpc_send_buff_size_in_bytes:
 # rpc_recv_buff_size_in_bytes:
 
+# uncomment to set socket buffer size for internode communication
+# internode_send_buff_size_in_bytes:
+# internode_recv_buff_size_in_bytes:
+
 # Frame size for thrift (maximum field length).
 thrift_framed_transport_size_in_mb: 15
 

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

@@ -87,6 +87,8 @@ public class Config
     public Integer rpc_max_threads = null;
     public Integer rpc_send_buff_size_in_bytes;
     public Integer rpc_recv_buff_size_in_bytes;
+    public Integer internode_send_buff_size_in_bytes;
+    public Integer internode_recv_buff_size_in_bytes;
 
     public Boolean start_native_transport = false;
     public Integer native_transport_port = 9042;

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

@@ -982,6 +982,16 @@ public class DatabaseDescriptor
         return conf.rpc_recv_buff_size_in_bytes;
     }
 
+    public static Integer getInternodeSendBufferSize()
+    {
+        return conf.internode_send_buff_size_in_bytes;
+    }
+
+    public static Integer getInternodeRecvBufferSize()
+    {
+        return conf.internode_recv_buff_size_in_bytes;
+    }
+
     public static boolean startNativeTransport()
     {
         return conf.start_native_transport;

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

@@ -20,6 +20,7 @@ package org.apache.cassandra.net;
 import java.io.*;
 import java.net.InetAddress;
 import java.net.Socket;
+import java.net.SocketException;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -42,6 +43,17 @@ public class IncomingTcpConnection extends Thread
     {
         assert socket != null;
         this.socket = socket;
+        if (DatabaseDescriptor.getInternodeRecvBufferSize() != null)
+        {
+            try
+            {
+                this.socket.setReceiveBufferSize(DatabaseDescriptor.getInternodeRecvBufferSize().intValue());
+            }
+            catch (SocketException se)
+            {
+                logger.warn("Failed to set receive buffer size on internode socket.", se);
+            }
+        }
     }
 
     /**

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

@@ -23,6 +23,7 @@ import java.io.DataOutputStream;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.Socket;
+import java.net.SocketException;
 import java.nio.ByteBuffer;
 import java.util.UUID;
 import java.util.concurrent.BlockingQueue;
@@ -275,6 +276,17 @@ public class OutboundTcpConnection extends Thread
                 {
                     socket.setTcpNoDelay(DatabaseDescriptor.getInterDCTcpNoDelay());
                 }
+                if (DatabaseDescriptor.getInternodeSendBufferSize() != null)
+                {
+                    try
+                    {
+                        socket.setSendBufferSize(DatabaseDescriptor.getInternodeSendBufferSize().intValue());
+                    }
+                    catch (SocketException se)
+                    {
+                        logger.warn("Failed to set send buffer size on internode socket.", se);
+                    }
+                }
                 out = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream(), 4096));
 
                 if (targetVersion >= MessagingService.VERSION_12)