[RUBY-90] add tcp nodelay option

Author: Bulat Shakirzyanov

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 Features:
 
+* [RUBY-90] Add support for disabling nagle algorithm (tcp nodelay), enabled by default.
 * [RUBY-70] Add support for client-side timestamps, disabled by default.
 * [RUBY-114] Add support for serial consistency in batch requests.
 

lib/cassandra.rb

@@ -59,6 +59,9 @@ module Cassandra
   #
   # @option options [Integer] :port (9042) cassandra native protocol port.
   #
+  # @option options [Boolean] :nodelay (false) when set to `true`, disables
+  #   nagle algorithm.
+  #
   # @option options [String] :datacenter (nil) name of current datacenter.
   #   First datacenter found will be assumed current by default. Note that you
   #   can skip this option if you specify only hosts from the local datacenter
@@ -435,6 +438,10 @@ def self.cluster_async(options = {})
       Util.assert_one_of(CONSISTENCIES, consistency) { ":consistency must be one of #{CONSISTENCIES.inspect}, #{consistency.inspect} given" }
     end
 
+    if options.has_key?(:nodelay)
+      options[:nodelay] = !!options[:nodelay]
+    end
+
     if options.has_key?(:trace)
       options[:trace] = !!options[:trace]
     end

lib/cassandra/cluster/connector.rb

@@ -114,6 +114,13 @@ def refresh_status(host)
       def do_connect(host)
         @reactor.connect(host.ip.to_s, @connection_options.port, {:timeout => @connection_options.connect_timeout, :ssl => @connection_options.ssl}) do |connection|
           raise Errors::ClientError, 'Not connected, reactor stopped' unless connection
+
+          if @connection_options.nodelay?
+            connection.to_io.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)
+          else
+            connection.to_io.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 0)
+          end
+
           Protocol::CqlProtocolHandler.new(connection, @reactor, @connection_options.protocol_version, @connection_options.compressor, @connection_options.heartbeat_interval, @connection_options.idle_timeout)
         end.flat_map do |connection|
           f = request_options(connection)

lib/cassandra/cluster/options.rb

@@ -25,7 +25,7 @@ class Options
                   :schema_refresh_delay, :schema_refresh_timeout
       attr_accessor :protocol_version
 
-      def initialize(protocol_version, credentials, auth_provider, compressor, port, connect_timeout, ssl, connections_per_local_node, connections_per_remote_node, heartbeat_interval, idle_timeout, synchronize_schema, schema_refresh_delay, schema_refresh_timeout, client_timestamps)
+      def initialize(protocol_version, credentials, auth_provider, compressor, port, connect_timeout, ssl, connections_per_local_node, connections_per_remote_node, heartbeat_interval, idle_timeout, synchronize_schema, schema_refresh_delay, schema_refresh_timeout, client_timestamps, nodelay)
         @protocol_version       = protocol_version
         @credentials            = credentials
         @auth_provider          = auth_provider
@@ -39,6 +39,7 @@ def initialize(protocol_version, credentials, auth_provider, compressor, port, c
         @schema_refresh_delay   = schema_refresh_delay
         @schema_refresh_timeout = schema_refresh_timeout
         @client_timestamps      = client_timestamps
+        @nodelay                = nodelay
 
         @connections_per_local_node  = connections_per_local_node
         @connections_per_remote_node = connections_per_remote_node
@@ -52,6 +53,10 @@ def client_timestamps?
         @client_timestamps
       end
 
+      def nodelay?
+        @nodelay
+      end
+
       def compression
         @compressor && @compressor.algorithm
       end

lib/cassandra/driver.rb

@@ -87,7 +87,8 @@ def self.let(name, &block)
         synchronize_schema,
         schema_refresh_delay,
         schema_refresh_timeout,
-        client_timestamps
+        client_timestamps,
+        nodelay
       )
     end
 
@@ -116,6 +117,7 @@ def self.let(name, &block)
     let(:thread_pool_size)          { 4 }
     let(:shuffle_replicas)          { true }
     let(:client_timestamps)         { false }
+    let(:nodelay)                   { false }
 
     let(:connections_per_local_node)  { 2 }
     let(:connections_per_remote_node) { 1 }

spec/support/stub_io_reactor.rb

@@ -17,6 +17,16 @@
 #++
 
 class StubIoReactor
+  class NullObject
+    def method_missing(method, *args, &block)
+      self
+    end
+
+    def nil?
+      true
+    end
+  end
+
   class Connection
     attr_reader :host
     attr_reader :port
@@ -30,6 +40,7 @@ def initialize(host, port, timeout, ssl)
       @ssl       = ssl
       @blocked   = false
       @connected = true
+      @io        = NullObject.new
 
       @closed_listeners = []
       @incoming = ::Queue.new
@@ -46,6 +57,10 @@ def initialize(host, port, timeout, ssl)
       end
     end
 
+    def to_io
+      @io
+    end
+
     def connected?
       @connected
     end