[RFC/WIP] support non-blocking listeners, too

[RFC/WIP] support non-blocking listeners, too

[Eric Wong]

While a major design factor of this server was based
load-balancing with blocking accept() in a dedicated thread,
non-blocking accept() has become fairer with EPOLLEXCLUSIVE in
Linux 4.5+.

Additionally, most C10K servers rely on non-blocking listen
sockets.  For users of systemd (or similar) socket activation,
it is now possible to run yahns alongside another C10K server
while sharing the same socket with no change to the socket
itself.

This means yahns and another server can split traffic without
extra load balancing.

The main purpose of this change is to make zero downtime migrations,
both to and from yahns possible without an external load balancer.

This also allows users of small installations (zero or one
worker process who don't need load balancing) to avoid needing a
dedicated thread to accept connections.
---
  Needs more tests...

 Documentation/yahns_config.pod | 18 ++++++++++++++++++
 lib/yahns.rb                   |  1 +
 lib/yahns/acceptor.rb          | 23 +++++++++++++++++++++++
 lib/yahns/config.rb            |  2 +-
 lib/yahns/queue_epoll.rb       | 10 ++++++++++
 lib/yahns/queue_kqueue.rb      |  7 +++++++
 lib/yahns/server.rb            | 24 +++++++++++++++++++++---
 lib/yahns/socket_helper.rb     |  1 +
 test/test_server.rb            | 33 +++++++++++++++++++++++++++++++++
 9 files changed, 115 insertions(+), 4 deletions(-)

diff --git a/Documentation/yahns_config.pod b/Documentation/yahns_config.pod
index 08c2e27..359a4af 100644
--- a/Documentation/yahns_config.pod
+++ b/Documentation/yahns_config.pod
@@ -481,6 +481,24 @@ This has no effect on TCP listeners.
 
 Default: 0000 (world-read/writable)
 
+=item nonblock: BOOLEAN
+
+Bind a non-blocking listen socket.  Do not set this if you
+are using multiple worker_processes as it can cause unfair
+load balancing between workers.
+
+For small deployments with zero or one worker process, this can
+save a few megabytes of memory by avoiding a dedicated listener
+thread.
+
+This option does NOT apply to inherited sockets because chaing that
+flag can break (non-yahns) servers it shares a listen socket with.
+yahns 2.0+ supports inheriting either blocking or nonblocking
+listeners so it can inherit (from systemd or similar) sockets meant
+for other servers to ease migrations from/to yahns.
+
+Default: false
+
 =back
 
 =item logger LOGGER
diff --git a/lib/yahns.rb b/lib/yahns.rb
index 4cf911e..766b7d3 100644
--- a/lib/yahns.rb
+++ b/lib/yahns.rb
@@ -6,6 +6,7 @@
 require 'unicorn' # pulls in raindrops, kgio, fcntl, etc, stringio, and logger
 require 'sleepy_penguin'
 require 'io/wait'
+require 'io/nonblock'
 
 # kill off some unicorn internals we don't need
 # we'll probably just make kcar into a server parser so we don't depend
diff --git a/lib/yahns/acceptor.rb b/lib/yahns/acceptor.rb
index 7340a1a..471166f 100644
--- a/lib/yahns/acceptor.rb
+++ b/lib/yahns/acceptor.rb
@@ -41,6 +41,29 @@ def ac_quit
     return __ac_quit_done?
   end
 
+  # only for non-blocking sockets
+  def yahns_listen_init(ctx)
+    @ctx = ctx # aka client_class
+  end
+
+  # runs if and only if non-blocking (and ideally with EPOLLEXCLUSIVE)
+  def yahns_step
+    if c = kgio_tryaccept(@ctx, Kgio::SOCK_NONBLOCK | Kgio::SOCK_CLOEXEC)
+      c.yahns_init
+
+      # it is not safe to touch client in this thread after this,
+      # a worker thread may grab client right away
+      @ctx.queue.queue_add(c, @ctx.superclass::QEV_FLAGS)
+    end
+  rescue Errno::EMFILE, Errno::ENFILE => e
+    logger.error("#{e.message}, consider raising open file limits")
+    @ctx.queue.fdmap.desperate_expire(5)
+  rescue => e
+    Yahns::Log.exception(logger, 'accept (yahns_step)', e)
+  ensure
+    return :ignore
+  end
+
   def spawn_acceptor(nr, logger, client_class)
     @quit = false
     @thrs = nr.times.map do
diff --git a/lib/yahns/config.rb b/lib/yahns/config.rb
index 441d3f9..50cc735 100644
--- a/lib/yahns/config.rb
+++ b/lib/yahns/config.rb
@@ -204,7 +204,7 @@ def listen(address, options = {})
        value = options[key] and _check_int(key, value, 1)
     end
 
-    [ :ipv6only, :reuseport ].each do |key|
+    [ :ipv6only, :reuseport, :nonblock ].each do |key|
       (value = options[key]).nil? and next
       [ true, false ].include?(value) or
         raise ArgumentError, "#{var}: not boolean: #{key}=#{value.inspect}"
diff --git a/lib/yahns/queue_epoll.rb b/lib/yahns/queue_epoll.rb
index 9e4271a..343edfb 100644
--- a/lib/yahns/queue_epoll.rb
+++ b/lib/yahns/queue_epoll.rb
@@ -28,6 +28,16 @@ def queue_add(io, flags)
     epoll_ctl(Epoll::CTL_ADD, io, flags)
   end
 
+  # non-blocking listeners are level-trigger,
+  def queue_add_acceptor(io)
+    epoll_ctl(Epoll::CTL_ADD, io, Epoll::IN | Epoll::EXCLUSIVE)
+    true
+    # caller won't warn
+  rescue Errno::EINVAL, NameError
+    epoll_ctl(Epoll::CTL_ADD, io, Epoll::IN)
+    false # caller warns
+  end
+
   def queue_mod(io, flags)
     epoll_ctl(Epoll::CTL_MOD, io, flags)
   end
diff --git a/lib/yahns/queue_kqueue.rb b/lib/yahns/queue_kqueue.rb
index 3c4c51c..27c68fc 100644
--- a/lib/yahns/queue_kqueue.rb
+++ b/lib/yahns/queue_kqueue.rb
@@ -31,6 +31,13 @@ def queue_add(io, flags)
     kevent(Kevent[io.fileno, flags, fflags, 0, 0, io])
   end
 
+  # non-blocking listeners are level-trigger
+  def queue_add_acceptor(io)
+    kevent(Kevent[io.fileno, EvFilt::READ, Ev::ADD, 0, 0, io])
+    # no EPOLLEXCLUSIVE analogy, so assume thundering herds :<
+    false
+  end
+
   def queue_mod(io, flags)
     kevent(Kevent[io.fileno, flags, ADD_ONESHOT, 0, 0, io])
   end
diff --git a/lib/yahns/server.rb b/lib/yahns/server.rb
index d13c57e..071500e 100644
--- a/lib/yahns/server.rb
+++ b/lib/yahns/server.rb
@@ -328,7 +328,14 @@ def inherit_listeners!
       io = server_cast(io, opts)
       set_server_sockopt(io, opts)
       name = sock_name(io)
-      @logger.info "inherited addr=#{name} fd=#{io.fileno}"
+      nb = io.nonblock?
+      @logger.info "inherited addr=#{name} fd=#{io.fileno} nonblock=#{nb}"
+      case c = opts[:nonblock]
+      when false, true
+        @logger.warn "inherited nonblock=#{nb}, but nonblock=#{c} in config"
+        @logger.warn "ignoring config, leaving as nonblock=#{nb}"
+        @logger.warn 'we cannot safely change nonblock flag on shared sockets'
+      end
       @config.register_inherited(name)
       io
     end
@@ -402,8 +409,19 @@ def fdmap_init
         ssl_ctx.setup
       end
       ctx_list << ctx
-      # acceptors feed the the queues
-      l.spawn_acceptor(opts[:threads] || 1, @logger, ctx)
+
+      # our whole design is was based on BLOCKING listeners; back in 2010
+      # However, Linux 4.5 (2016-03-13) added EPOLLEXCLUSIVE, which
+      # (at least on Linux) allows non-blocking listeners to solve the
+      # same problem we solved by using blocking listeners.
+      if l.nonblock?
+        l.yahns_listen_init(ctx)
+        next if ctx.queue.queue_add_acceptor(l)
+        ((@worker_processes || 0) > 1) and @logger.warn(
+'non-blocking listener w/o EPOLLEXCLUSIVE, balance degraded')
+      else # our original design, acceptors feed the the queues
+        l.spawn_acceptor(opts[:threads] || 1, @logger, ctx)
+      end
     end
     fdmap
   end
diff --git a/lib/yahns/socket_helper.rb b/lib/yahns/socket_helper.rb
index 963c9fa..02f2d15 100644
--- a/lib/yahns/socket_helper.rb
+++ b/lib/yahns/socket_helper.rb
@@ -79,6 +79,7 @@ def bind_listen(address, opt)
     else
       raise ArgumentError, "Don't know how to bind: #{address}"
     end
+    sock.nonblock = opt[:nonblock] || false
     set_server_sockopt(sock, opt)
     sock
   end
diff --git a/test/test_server.rb b/test/test_server.rb
index 75e1857..6083a2f 100644
--- a/test/test_server.rb
+++ b/test/test_server.rb
@@ -903,4 +903,37 @@ def test_inherit_tcp_nodelay_set
   ensure
     quit_wait(pid)
   end
+
+  def kernel_major_minor
+    major, minor = Etc.uname[:release].split('.')[0,2].map(&:to_i)
+    ver_int(major, minor)
+  end
+
+  def ver_int(major, minor)
+    (major << 24) | (minor << 16)
+  end
+
+  def test_inherit_nonblocking
+    err = @err
+    cfg = Yahns::Config.new
+    host, port = @srv.addr[3], @srv.addr[1]
+    @srv.nonblock = true
+    cfg.instance_eval do
+      ru = lambda { |_| [ 200, { 'Content-Length' => '2' } , [ 'HI' ] ] }
+      GTL.synchronize { app(:rack, ru) { listen "#{host}:#{port}" } }
+      logger(Logger.new(err.path))
+    end
+    pid = mkserver(cfg, @srv) { ENV["YAHNS_FD"] = "#{@srv.fileno}" }
+    run_client(host, port) { |res| assert_equal "HI", res.body }
+    assert_predicate @srv, :nonblock?
+    unless defined?(SleepyPenguin::Epoll::EXCLUSIVE) &&
+           RUBY_PLATFORM =~ /linux/ &&
+           kernel_major_minor >= ver_int(4,5)
+      err.flush
+      err.rewind
+      err.truncate(0)
+    end
+  ensure
+    quit_wait(pid)
+  end
 end
-- 
EW

Re: [RFC/WIP] support non-blocking listeners, too

[Eric Wong]

Eric Wong <e@80x24.org> wrote:
> While a major design factor of this server was based
> load-balancing with blocking accept() in a dedicated thread,
> non-blocking accept() has become fairer with EPOLLEXCLUSIVE in
> Linux 4.5+.

Actually... not sure if it matters, even.  EPOLLEXCLUSIVE still
seems worse off than a blocking accept4() thread.  The process
which gets woken from epoll_wait can still be stuck dealing with
application processing or other I/O (the kernel has no way of
knowing that).

Blocking on accept4() is fairer in that regard since it
round-robins the processes.

On a related note, having a multi_accept parameter (like nginx)
should be beneficial to single-process deployments.