Socket: add {snd,rcv}buf opts to bind_listen

bind_listen takes a hash as its second parameter now, allowing
the addition of :sndbuf and :rcvbuf options to specify the size
of the buffers in bytes.  These correspond to the SO_SNDBUF and
SO_RCVBUF options via setsockopt(2) respectively.

This also adds support for per-listener backlogs to be used.

However, this is only an internal API change and the changes
have not yet been exposed to the user via Unicorn::Configurator,
yet.

Also add a bunch of SocketHelper tests

4 files changed, 184 insertions, 6 deletions

diff --git a/Manifest b/Manifest
index 0889fc7..3487b13 100644
--- a/Manifest
+++ b/Manifest
@@ -21,6 +21,9 @@ ext/unicorn/http11/http11_parser.rl
 ext/unicorn/http11/http11_parser_common.rl
 lib/unicorn.rb
 lib/unicorn/app/exec_cgi.rb
+lib/unicorn/app/old_rails.rb
+lib/unicorn/app/old_rails/static.rb
+lib/unicorn/cgi_wrapper.rb
 lib/unicorn/configurator.rb
 lib/unicorn/const.rb
 lib/unicorn/http_request.rb
@@ -30,9 +33,11 @@ lib/unicorn/socket.rb
 lib/unicorn/util.rb
 setup.rb
 test/aggregate.rb
-test/benchmark/previous.rb
-test/benchmark/simple.rb
-test/benchmark/utils.rb
+test/benchmark/README
+test/benchmark/big_request.rb
+test/benchmark/dd.ru
+test/benchmark/request.rb
+test/benchmark/response.rb
 test/exec/README
 test/exec/test_exec.rb
 test/test_helper.rb
@@ -42,4 +47,5 @@ test/unit/test_http_parser.rb
 test/unit/test_request.rb
 test/unit/test_response.rb
 test/unit/test_server.rb
+test/unit/test_socket_helper.rb
 test/unit/test_upload.rb
diff --git a/lib/unicorn.rb b/lib/unicorn.rb
index eefbfc1..e36cb1e 100644
--- a/lib/unicorn.rb
+++ b/lib/unicorn.rb
@@ -135,7 +135,7 @@ module Unicorn
     def listen(address)
       return if String === address && listener_names.include?(address)
 
-      if io = bind_listen(address, @backlog)
+      if io = bind_listen(address, { :backlog => @backlog })
         if Socket == io.class
           @io_purgatory << io
           io = server_cast(io)
diff --git a/lib/unicorn/socket.rb b/lib/unicorn/socket.rb
index 9519448..4870133 100644
--- a/lib/unicorn/socket.rb
+++ b/lib/unicorn/socket.rb
@@ -62,10 +62,17 @@ module Unicorn
       end
     end
 
+    def log_buffer_sizes(sock, pfx = '')
+      respond_to?(:logger) or return
+      rcvbuf = sock.getsockopt(SOL_SOCKET, SO_RCVBUF).unpack('i')
+      sndbuf = sock.getsockopt(SOL_SOCKET, SO_SNDBUF).unpack('i')
+      logger.info "#{pfx}#{sock_name(sock)} rcvbuf=#{rcvbuf} sndbuf=#{sndbuf}"
+    end
+
     # creates a new server, socket. address may be a HOST:PORT or
     # an absolute path to a UNIX socket.  address can even be a Socket
     # object in which case it is immediately returned
-    def bind_listen(address = '0.0.0.0:8080', backlog = 1024)
+    def bind_listen(address = '0.0.0.0:8080', opt = { :backlog => 1024 })
       return address unless String === address
 
       domain, bind_addr = if address[0..0] == "/"
@@ -95,7 +102,13 @@ module Unicorn
         sock.close rescue nil
         return nil
       end
-      sock.listen(backlog)
+      if opt[:rcvbuf] || opt[:sndbuf]
+        log_buffer_sizes(sock, "before: ")
+        sock.setsockopt(SOL_SOCKET, SO_RCVBUF, opt[:rcvbuf]) if opt[:rcvbuf]
+        sock.setsockopt(SOL_SOCKET, SO_SNDBUF, opt[:sndbuf]) if opt[:sndbuf]
+        log_buffer_sizes(sock, " after: ")
+      end
+      sock.listen(opt[:backlog] || 1024)
       set_server_sockopt(sock) if domain == AF_INET
       sock
     end
diff --git a/test/unit/test_socket_helper.rb b/test/unit/test_socket_helper.rb
new file mode 100644
index 0000000..23fa44c
--- /dev/null
+++ b/test/unit/test_socket_helper.rb
@@ -0,0 +1,159 @@
+require 'test/test_helper'
+require 'tempfile'
+
+class TestSocketHelper < Test::Unit::TestCase
+  include Unicorn::SocketHelper
+  attr_reader :logger
+  GET_SLASH = "GET / HTTP/1.0\r\n\r\n".freeze
+
+  def setup
+    @log_tmp = Tempfile.new 'logger'
+    @logger = Logger.new(@log_tmp.path)
+    @test_addr = ENV['UNICORN_TEST_ADDR'] || '127.0.0.1'
+  end
+
+  def test_bind_listen_tcp
+    port = unused_port @test_addr
+    @tcp_listener_name = "#@test_addr:#{port}"
+    @tcp_listener = bind_listen(@tcp_listener_name)
+    assert Socket === @tcp_listener
+    assert_equal @tcp_listener_name, sock_name(@tcp_listener)
+  end
+
+  def test_bind_listen_options
+    port = unused_port @test_addr
+    tcp_listener_name = "#@test_addr:#{port}"
+    tmp = Tempfile.new 'unix.sock'
+    unix_listener_name = tmp.path
+    File.unlink(tmp.path)
+    [ { :backlog => 5 }, { :sndbuf => 4096 }, { :rcvbuf => 4096 },
+      { :backlog => 16, :rcvbuf => 4096, :sndbuf => 4096 }
+    ].each do |opts|
+      assert_nothing_raised do
+        tcp_listener = bind_listen(tcp_listener_name, opts)
+        assert Socket === tcp_listener
+        tcp_listener.close
+        unix_listener = bind_listen(unix_listener_name, opts)
+        assert Socket === unix_listener
+        unix_listener.close
+      end
+    end
+    #system('cat', @log_tmp.path)
+  end
+
+  def test_bind_listen_unix
+    tmp = Tempfile.new 'unix.sock'
+    @unix_listener_path = tmp.path
+    File.unlink(@unix_listener_path)
+    @unix_listener = bind_listen(@unix_listener_path)
+    assert Socket === @unix_listener
+    assert_equal @unix_listener_path, sock_name(@unix_listener)
+  end
+
+  def test_bind_listen_unix_idempotent
+    test_bind_listen_unix
+    a = bind_listen(@unix_listener)
+    assert_equal a.fileno, @unix_listener.fileno
+    unix_server = server_cast(@unix_listener)
+    a = bind_listen(unix_server)
+    assert_equal a.fileno, unix_server.fileno
+    assert_equal a.fileno, @unix_listener.fileno
+  end
+
+  def test_bind_listen_tcp_idempotent
+    test_bind_listen_tcp
+    a = bind_listen(@tcp_listener)
+    assert_equal a.fileno, @tcp_listener.fileno
+    tcp_server = server_cast(@tcp_listener)
+    a = bind_listen(tcp_server)
+    assert_equal a.fileno, tcp_server.fileno
+    assert_equal a.fileno, @tcp_listener.fileno
+  end
+
+  def test_bind_listen_unix_rebind
+    test_bind_listen_unix
+    new_listener = bind_listen(@unix_listener_path)
+    assert Socket === new_listener
+    assert new_listener.fileno != @unix_listener.fileno
+    assert_equal sock_name(new_listener), sock_name(@unix_listener)
+    assert_equal @unix_listener_path, sock_name(new_listener)
+    pid = fork do
+      client = server_cast(new_listener).accept
+      client.syswrite('abcde')
+      exit 0
+    end
+    s = UNIXSocket.new(@unix_listener_path)
+    IO.select([s])
+    assert_equal 'abcde', s.sysread(5)
+    pid, status = Process.waitpid2(pid)
+    assert status.success?
+  end
+
+  def test_server_cast
+    assert_nothing_raised do
+      test_bind_listen_unix
+      test_bind_listen_tcp
+    end
+    @unix_server = server_cast(@unix_listener)
+    assert_equal @unix_listener.fileno, @unix_server.fileno
+    assert UNIXServer === @unix_server
+    assert File.socket?(@unix_server.path)
+    assert_equal @unix_listener_path, sock_name(@unix_server)
+
+    @tcp_server = server_cast(@tcp_listener)
+    assert_equal @tcp_listener.fileno, @tcp_server.fileno
+    assert TCPServer === @tcp_server
+    assert_equal @tcp_listener_name, sock_name(@tcp_server)
+  end
+
+  def test_sock_name
+    test_server_cast
+    sock_name(@unix_server)
+  end
+
+  def test_tcp_unicorn_peeraddr
+    test_bind_listen_tcp
+    @tcp_server = server_cast(@tcp_listener)
+    tmp = Tempfile.new 'shared'
+    pid = fork do
+      client = @tcp_server.accept
+      IO.select([client])
+      assert_equal GET_SLASH, client.sysread(GET_SLASH.size)
+      tmp.syswrite "#{client.unicorn_peeraddr}"
+      exit 0
+    end
+    host, port = sock_name(@tcp_server).split(/:/)
+    client = TCPSocket.new(host, port.to_i)
+    client.syswrite(GET_SLASH)
+
+    pid, status = Process.waitpid2(pid)
+    assert_nothing_raised { client.close }
+    assert status.success?
+    tmp.sysseek 0
+    assert_equal @test_addr, tmp.sysread(4096)
+    tmp.sysseek 0
+  end
+
+  def test_unix_unicorn_peeraddr
+    test_bind_listen_unix
+    @unix_server = server_cast(@unix_listener)
+    tmp = Tempfile.new 'shared'
+    pid = fork do
+      client = @unix_server.accept
+      IO.select([client])
+      assert_equal GET_SLASH, client.sysread(4096)
+      tmp.syswrite "#{client.unicorn_peeraddr}"
+      exit 0
+    end
+    client = UNIXSocket.new(@unix_listener_path)
+    client.syswrite(GET_SLASH)
+
+    pid, status = Process.waitpid2(pid)
+    assert_nothing_raised { client.close }
+    assert status.success?
+    tmp.sysseek 0
+    assert_equal '127.0.0.1', tmp.sysread(4096)
+    tmp.sysseek 0
+  end
+
+end