test/test_server.rb

   1 # Copyright (C) 2013-2016 all contributors <yahns-public@yhbt.net>
   2 # License: GPL-3.0+ (https://www.gnu.org/licenses/gpl-3.0.txt)
   3 # frozen_string_literal: true
   4 require_relative 'server_helper'
   5
   6 class TestServer < Testcase
   7   ENV["N"].to_i > 1 and parallelize_me!
   8   include ServerHelper
   9
  10   alias setup server_helper_setup
  11   alias teardown server_helper_teardown
  12
  13   def test_single_process
  14     err = @err
  15     cfg = Yahns::Config.new
  16     host, port = @srv.addr[3], @srv.addr[1]
  17     cfg.instance_eval do
  18       ru = lambda { |_| [ 200, {'Content-Length'=>'2'}, ['HI'] ] }
  19       GTL.synchronize { app(:rack, ru) { listen "#{host}:#{port}" } }
  20       logger(Logger.new(err.path))
  21     end
  22     pid = mkserver(cfg)
  23     run_client(host, port) { |res| assert_equal "HI", res.body }
  24     c = get_tcp_client(host, port)
  25
  26     # test pipelining
  27     r = "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"
  28     c.write(r + r)
  29     buf = ''.dup
  30     Timeout.timeout(10) do
  31       until buf =~ /HI.+HI/m
  32         buf << c.readpartial(4096)
  33       end
  34     end
  35
  36     # trickle pipelining
  37     c.write(r + "GET ")
  38     buf = ''.dup
  39     Timeout.timeout(10) do
  40       until buf =~ /HI\z/
  41         buf << c.readpartial(4096)
  42       end
  43     end
  44     c.write("/ HTTP/1.1\r\nHost: example.com\r\n\r\n")
  45     Timeout.timeout(10) do
  46       until buf =~ /HI.+HI/m
  47         buf << c.readpartial(4096)
  48       end
  49     end
  50     Process.kill(:QUIT, pid)
  51     "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n".each_byte do |x|
  52       sleep(0.01)
  53       c.write(x.chr)
  54     end
  55     buf = Timeout.timeout(30) { c.read }
  56     assert_match(/Connection: close/, buf)
  57     _, status = Timeout.timeout(10) { Process.waitpid2(pid) }
  58     assert status.success?, status.inspect
  59     c.close
  60   end
  61
  62   def test_input_body_true; input_body(true); end
  63   def test_input_body_false; input_body(false); end
  64   def test_input_body_lazy; input_body(:lazy); end
  65
  66   def input_body(btype)
  67     err = @err
  68     cfg = Yahns::Config.new
  69     host, port = @srv.addr[3], @srv.addr[1]
  70     cfg.instance_eval do
  71       ru = lambda {|e|[ 200, {'Content-Length'=>'2'},[e["rack.input"].read]]}
  72       GTL.synchronize do
  73         app(:rack, ru) do
  74           listen "#{host}:#{port}"
  75           input_buffering btype
  76         end
  77       end
  78       logger(Logger.new(err.path))
  79     end
  80     pid = mkserver(cfg)
  81     c = get_tcp_client(host, port)
  82     buf = "PUT / HTTP/1.0\r\nContent-Length: 2\r\n\r\nHI"
  83     c.write(buf)
  84     IO.select([c], nil, nil, 5)
  85     rv = c.read(666)
  86     head, body = rv.split(/\r\n\r\n/)
  87     assert_match(%r{^Content-Length: 2\r\n}, head)
  88     assert_equal "HI", body, "#{rv.inspect} - #{btype.inspect}"
  89     c.close
  90
  91     # pipelined oneshot
  92     buf = "PUT / HTTP/1.1\r\nContent-Length: 2\r\n\r\nHI"
  93     c = get_tcp_client(host, port)
  94     c.write(buf + buf)
  95     buf = ''.dup
  96     Timeout.timeout(10) do
  97       until buf =~ /HI.+HI/m
  98         buf << c.readpartial(4096)
  99       end
 100     end
 101     assert buf.gsub!(/Date:[^\r\n]+\r\n/, ""), "kill differing Date"
 102     rv = buf.sub!(/\A(HTTP.+?\r\n\r\nHI)/m, "")
 103     first = $1
 104     assert rv
 105     assert_equal first, buf
 106
 107     # pipelined trickle
 108     buf = "PUT / HTTP/1.1\r\nContent-Length: 5\r\n\r\nHIBYE"
 109     (buf + buf).each_byte do |b|
 110       c.write(b.chr)
 111       sleep(0.01) if b.chr == ":"
 112       Thread.pass
 113     end
 114     buf = ''.dup
 115     Timeout.timeout(10) do
 116       until buf =~ /HIBYE.+HIBYE/m
 117         buf << c.readpartial(4096)
 118       end
 119     end
 120     assert buf.gsub!(/Date:[^\r\n]+\r\n/, ""), "kill differing Date"
 121     rv = buf.sub!(/\A(HTTP.+?\r\n\r\nHIBYE)/m, "")
 122     first = $1
 123     assert rv
 124     assert_equal first, buf
 125   ensure
 126     c.close if c
 127     quit_wait(pid)
 128   end
 129
 130   def test_trailer_true; trailer(true); end
 131   def test_trailer_false; trailer(false); end
 132   def test_trailer_lazy; trailer(:lazy); end
 133   def test_slow_trailer_true; trailer(true, 0.02); end
 134   def test_slow_trailer_false; trailer(false, 0.02); end
 135   def test_slow_trailer_lazy; trailer(:lazy, 0.02); end
 136
 137   def trailer(btype, delay = false)
 138     err = @err
 139     cfg = Yahns::Config.new
 140     host, port = @srv.addr[3], @srv.addr[1]
 141     cfg.instance_eval do
 142       ru = lambda do |e|
 143         body = e["rack.input"].read
 144         s = e["HTTP_XBT"] + "\n" + body
 145         [ 200, {'Content-Length'=>s.size.to_s}, [ s ] ]
 146       end
 147       GTL.synchronize do
 148         app(:rack, ru) do
 149           listen "#{host}:#{port}"
 150           input_buffering btype
 151         end
 152       end
 153       logger(Logger.new(err.path))
 154     end
 155     pid = mkserver(cfg)
 156     c = get_tcp_client(host, port)
 157     buf = "PUT / HTTP/1.0\r\nTrailer:xbt\r\nTransfer-Encoding: chunked\r\n\r\n"
 158     c.write(buf)
 159     xbt = btype.to_s
 160     sleep(delay) if delay
 161     c.write(sprintf("%x\r\n", xbt.size))
 162     sleep(delay) if delay
 163     c.write(xbt)
 164     sleep(delay) if delay
 165     c.write("\r\n")
 166     sleep(delay) if delay
 167     c.write("0\r\nXBT: ")
 168     sleep(delay) if delay
 169     c.write("#{xbt}\r\n\r\n")
 170     IO.select([c], nil, nil, 5000) or raise "timed out"
 171     rv = c.read(666)
 172     _, body = rv.split(/\r\n\r\n/)
 173     a, b = body.split(/\n/)
 174     assert_equal xbt, a
 175     assert_equal xbt, b
 176   ensure
 177     c.close if c
 178     quit_wait(pid)
 179   end
 180
 181   def test_check_client_connection
 182     tmpdir = yahns_mktmpdir
 183     sock = "#{tmpdir}/sock"
 184     unix_srv = UNIXServer.new(sock)
 185     msgs = %w(ZZ zz)
 186     err = @err
 187     cfg = Yahns::Config.new
 188     bpipe = IO.pipe
 189     cfg.instance_eval do
 190       ru = lambda { |e|
 191         case e['PATH_INFO']
 192         when '/sleep'
 193           a = Object.new
 194           a.instance_variable_set(:@bpipe, bpipe)
 195           a.instance_variable_set(:@msgs, msgs)
 196           def a.each
 197             @msgs.each do |msg|
 198               yield @bpipe[0].read(msg.size)
 199             end
 200           end
 201         when '/cccfail'
 202           # we should not get here if check_client_connection worked
 203           abort "CCCFAIL"
 204         else
 205           a = %w(HI)
 206         end
 207         [ 200, {'Content-Length'=>'2'}, a ]
 208       }
 209       GTL.synchronize {
 210         app(:rack, ru) {
 211           listen sock
 212           check_client_connection true
 213           # needed to avoid concurrency with check_client_connection
 214           queue { worker_threads 1 }
 215           output_buffering false
 216         }
 217       }
 218       logger(Logger.new(err.path))
 219     end
 220     srv = Yahns::Server.new(cfg)
 221
 222     # ensure we set worker_threads correctly
 223     eggs = srv.instance_variable_get(:@config).qeggs
 224     assert_equal 1, eggs.size
 225     assert_equal 1, eggs.first[1].instance_variable_get(:@worker_threads)
 226
 227     pid = xfork do
 228       bpipe[1].close
 229       ENV["YAHNS_FD"] = unix_srv.fileno.to_s
 230       unix_srv.autoclose = false
 231       srv.start.join
 232     end
 233     bpipe[0].close
 234     a = UNIXSocket.new(sock)
 235     b = UNIXSocket.new(sock)
 236     a.write("GET /sleep HTTP/1.0\r\n\r\n")
 237     r = IO.select([a], nil, nil, 4)
 238     assert r, "nothing ready"
 239     assert_equal a, r[0][0]
 240     buf = a.read(8)
 241     assert_equal "HTTP/1.1", buf
 242
 243     # hope the kernel sees this before it sees the bpipe ping-ponging below
 244     b.write("GET /cccfail HTTP/1.0\r\n\r\n")
 245     b.shutdown
 246     b.close
 247
 248     # ping-pong a bit to stall the server
 249     msgs.each do |msg|
 250       bpipe[1].write(msg)
 251       Timeout.timeout(10) { buf << a.readpartial(10) until buf =~ /#{msg}/ }
 252     end
 253     bpipe[1].close
 254     assert_equal msgs.join, buf.split(/\r\n\r\n/)[1]
 255
 256     # do things still work?
 257     c = UNIXSocket.new(sock)
 258     c.write "GET /\r\n\r\n"
 259     assert_equal "HI", c.read
 260     c.close
 261     a.close
 262   rescue => e
 263     warn e.class
 264     warn e.message
 265     warn e.backtrace.join("\n")
 266   ensure
 267     unix_srv.close
 268     quit_wait(pid)
 269     FileUtils.rm_rf(tmpdir)
 270   end
 271
 272   def test_mp
 273     pid, host, port = new_mp_server
 274     wpid = nil
 275     run_client(host, port) do |res|
 276       wpid ||= res.body.to_i
 277     end
 278   ensure
 279     quit_wait(pid)
 280     if wpid
 281       assert_raises(Errno::ESRCH) { Process.kill(:KILL, wpid) }
 282       assert_raises(Errno::ECHILD) { Process.waitpid2(wpid) }
 283     end
 284   end
 285
 286   def test_mp_worker_die
 287     pid, host, port = new_mp_server
 288     wpid1 = wpid2 = nil
 289     run_client(host, port) do |res|
 290       wpid1 ||= res.body.to_i
 291     end
 292     Process.kill(:QUIT, wpid1)
 293     poke_until_dead(wpid1)
 294     run_client(host, port) do |res|
 295       wpid2 ||= res.body.to_i
 296     end
 297     refute_equal wpid2, wpid1
 298   ensure
 299     quit_wait(pid)
 300     assert_raises(Errno::ESRCH) { Process.kill(:KILL, wpid2) } if wpid2
 301   end
 302
 303   def test_mp_dead_parent
 304     pid, host, port = new_mp_server(1)
 305     wpid = nil
 306     run_client(host, port) do |res|
 307       wpid ||= res.body.to_i
 308     end
 309     Process.kill(:KILL, pid)
 310     _, status = Process.waitpid2(pid)
 311     assert status.signaled?, status.inspect
 312     poke_until_dead(wpid)
 313   end
 314
 315   def run_client(host, port)
 316     c = get_tcp_client(host, port)
 317     Net::HTTP.start(host, port) do |http|
 318       res = http.request(Net::HTTP::Get.new("/"))
 319       assert_equal 200, res.code.to_i
 320       assert_equal "keep-alive", res["Connection"]
 321       yield res
 322       res = http.request(Net::HTTP::Get.new("/"))
 323       assert_equal 200, res.code.to_i
 324       assert_equal "keep-alive", res["Connection"]
 325       yield res
 326     end
 327     c.write "GET / HTTP/1.0\r\n\r\n"
 328     res = Timeout.timeout(10) { c.read }
 329     head, _ = res.split(/\r\n\r\n/)
 330     head = head.split(/\r\n/)
 331     assert_equal "HTTP/1.1 200 OK", head[0]
 332     assert_equal "Connection: close", head[-1]
 333     c.close
 334   end
 335
 336   def new_mp_server(nr = 2)
 337     ru = @ru = tmpfile(%w(config .ru))
 338     @ru.puts('a = $$.to_s')
 339     @ru.puts('run lambda { |_| [ 200, {"Content-Length"=>a.size.to_s},[a]]}')
 340     err = @err
 341     cfg = Yahns::Config.new
 342     host, port = @srv.addr[3], @srv.addr[1]
 343     cfg.instance_eval do
 344       worker_processes nr
 345       GTL.synchronize { app(:rack, ru.path) { listen "#{host}:#{port}" } }
 346       logger(Logger.new(File.open(err.path, "a")))
 347     end
 348     pid = mkserver(cfg)
 349     [ pid, host, port ]
 350   end
 351
 352   def test_nonpersistent
 353     err = @err
 354     cfg = Yahns::Config.new
 355     host, port = @srv.addr[3], @srv.addr[1]
 356     cfg.instance_eval do
 357       ru = lambda { |_| [ 200, {'Content-Length'=>'2'}, ['HI'] ] }
 358       GTL.synchronize {
 359         app(:rack, ru) {
 360           listen "#{host}:#{port}"
 361           persistent_connections false
 362         }
 363       }
 364       logger(Logger.new(err.path))
 365     end
 366     pid = mkserver(cfg)
 367     c = get_tcp_client(host, port)
 368     c.write("GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
 369     buf = Timeout.timeout(10) { c.read }
 370     assert_match(/Connection: close/, buf)
 371     c.close
 372   ensure
 373     quit_wait(pid)
 374   end
 375
 376   def test_ttin_ttou
 377     err, cfg, host, port = @err, Yahns::Config.new, @srv.addr[3], @srv.addr[1]
 378     ru = lambda { |_|
 379       b = "#$$"
 380       [ 200, {'Content-Length'=>b.size.to_s}, [b] ]
 381     }
 382     cfg.instance_eval do
 383       GTL.synchronize { app(:rack, ru) { listen "#{host}:#{port}" } }
 384       worker_processes 1
 385       stderr_path err.path
 386     end
 387     pid = mkserver(cfg)
 388
 389     read_pid = lambda do
 390       c = get_tcp_client(host, port)
 391       c.write("GET /\r\n\r\n")
 392       body = Timeout.timeout(10) { c.read }
 393       c.close
 394       assert_match(/\A\d+\z/, body)
 395       body
 396     end
 397
 398     orig_worker_pid = read_pid.call.to_i
 399     assert_equal 1, Process.kill(0, orig_worker_pid)
 400
 401     Process.kill(:TTOU, pid)
 402     poke_until_dead(orig_worker_pid)
 403
 404     Process.kill(:TTIN, pid)
 405     second_worker_pid = read_pid.call.to_i
 406
 407     # PID recycling is rare, hope it doesn't fail here
 408     refute_equal orig_worker_pid, second_worker_pid
 409   ensure
 410     quit_wait(pid)
 411   end
 412
 413   def test_mp_hooks
 414     err = @err
 415     out = tmpfile(%w(mp_hooks .out))
 416     cfg = Yahns::Config.new
 417     host, port = @srv.addr[3], @srv.addr[1]
 418     cfg.instance_eval do
 419       ru = lambda {|_|x="#$$";[200,{'Content-Length'=>x.size.to_s },[x]]}
 420       GTL.synchronize {
 421         app(:rack, ru) {
 422           listen "#{host}:#{port}"
 423           persistent_connections false
 424           atfork_child { warn "INFO hihi from app.atfork_child" }
 425         }
 426         worker_processes(1) do
 427           atfork_child { puts "af #$$ worker is running" }
 428           atfork_prepare { puts "af #$$ parent about to spawn" }
 429           atfork_parent { puts "af #$$ parent done spawning" }
 430         end
 431       }
 432       stderr_path err.path
 433       stdout_path out.path
 434     end
 435     master_pid = pid = mkserver(cfg)
 436     c = get_tcp_client(host, port)
 437     c.write("GET / HTTP/1.0\r\nHost: example.com\r\n\r\n")
 438     buf = Timeout.timeout(10) { c.read }
 439     c.close
 440     head, body = buf.split(/\r\n\r\n/)
 441     assert_match(/200 OK/, head)
 442     assert_match(/\A\d+\z/, body)
 443     worker_pid = body.to_i
 444
 445     # ensure atfork_parent has run
 446     quit_wait(master_pid)
 447     master_pid = nil
 448
 449     lines = out.readlines.map!(&:chomp!)
 450     out.close!
 451
 452     assert_match %r{INFO hihi from app\.atfork_child}, File.read(err.path)
 453
 454     assert_equal 3, lines.size, lines.join("\n")
 455     assert_equal("af #{pid} parent about to spawn", lines.shift)
 456
 457     # child/parent ordering is not guaranteed
 458     assert_equal 1, lines.grep(/\Aaf #{pid} parent done spawning\z/).size
 459     assert_equal 1, lines.grep(/\Aaf #{worker_pid} worker is running\z/).size
 460   ensure
 461     quit_wait(master_pid)
 462   end
 463
 464   def test_mp_hooks_worker_nr
 465     err = @err
 466     out = tmpfile(%w(mp_hooks .out))
 467     cfg = Yahns::Config.new
 468     host, port = @srv.addr[3], @srv.addr[1]
 469     cfg.instance_eval do
 470       ru = lambda {|_|x="#$$";[200,{'Content-Length'=>x.size.to_s },[x]]}
 471       GTL.synchronize {
 472         app(:rack, ru) {
 473           listen "#{host}:#{port}"
 474           persistent_connections false
 475           atfork_child { |nr| warn "INFO hihi.#{nr} from app.atfork_child" }
 476         }
 477         worker_processes(1) do
 478           atfork_child { |nr| puts "af.#{nr} #$$ worker is running" }
 479           atfork_prepare { |nr| puts "af.#{nr} #$$ parent about to spawn" }
 480           atfork_parent { |nr| puts "af.#{nr} #$$ parent done spawning" }
 481         end
 482       }
 483       stderr_path err.path
 484       stdout_path out.path
 485     end
 486     pid = mkserver(cfg)
 487     c = get_tcp_client(host, port)
 488     c.write("GET / HTTP/1.0\r\nHost: example.com\r\n\r\n")
 489     buf = Timeout.timeout(10) { c.read }
 490     c.close
 491     head, body = buf.split(/\r\n\r\n/)
 492     assert_match(/200 OK/, head)
 493     assert_match(/\A\d+\z/, body)
 494     worker_pid = body.to_i
 495     lines = out.readlines.map!(&:chomp!)
 496     out.close!
 497
 498     assert_match %r{INFO hihi\.0 from app\.atfork_child}, File.read(err.path)
 499     assert_equal 3, lines.size
 500     assert_equal("af.0 #{pid} parent about to spawn", lines.shift)
 501
 502     # child/parent ordering is not guaranteed
 503     assert_equal 1,
 504         lines.grep(/\Aaf\.0 #{pid} parent done spawning\z/).size
 505     assert_equal 1,
 506         lines.grep(/\Aaf\.0 #{worker_pid} worker is running\z/).size
 507   ensure
 508     quit_wait(pid)
 509   end
 510
 511   def test_pidfile_usr2
 512     tmpdir = yahns_mktmpdir
 513     pidf = "#{tmpdir}/pid"
 514     old = "#{pidf}.oldbin"
 515     err = @err
 516     cfg = Yahns::Config.new
 517     host, port = @srv.addr[3], @srv.addr[1]
 518     cfg.instance_eval do
 519       GTL.synchronize {
 520         app(:rack, lambda { |_| [ 200, {}, [] ] }) { listen "#{host}:#{port}" }
 521         pid pidf
 522       }
 523       stderr_path err.path
 524     end
 525     pid = mkserver(cfg) do
 526       Yahns::START[0] = "sh"
 527       Yahns::START[:argv] = [ '-c', "echo $$ > #{pidf}; sleep 10" ]
 528     end
 529
 530     # ensure server is running
 531     c = get_tcp_client(host, port)
 532     c.write("GET / HTTP/1.0\r\n\r\n")
 533     buf = Timeout.timeout(10) { c.read }
 534     assert_match(/Connection: close/, buf)
 535     c.close
 536
 537     assert_equal pid, File.read(pidf).to_i
 538     before = File.stat(pidf)
 539
 540     # start the upgrade
 541     Process.kill(:USR2, pid)
 542     Timeout.timeout(10) { sleep(0.01) until File.exist?(old) }
 543     after = File.stat(old)
 544     assert_equal after.ino, before.ino
 545     Timeout.timeout(10) { sleep(0.01) until File.exist?(pidf) }
 546     new = File.read(pidf).to_i
 547     refute_equal pid, new
 548
 549     # abort the upgrade (just wait for it to finish)
 550     Process.kill(:TERM, new)
 551     poke_until_dead(new)
 552
 553     # ensure reversion is OK
 554     Timeout.timeout(10) { sleep(0.01) while File.exist?(old) }
 555     after = File.stat(pidf)
 556     assert_equal before.ino, after.ino
 557     assert_equal before.mtime, after.mtime
 558     assert_equal pid, File.read(pidf).to_i
 559
 560     lines = File.readlines(err.path).grep(/ERROR/)
 561     assert_equal 1, lines.size
 562     assert_match(/reaped/, lines[0], lines)
 563     File.truncate(err.path, 0)
 564   ensure
 565     quit_wait(pid)
 566     FileUtils.rm_rf(tmpdir)
 567   end
 568
 569   module MockSwitchUser
 570     def self.included(cls)
 571       cls.__send__(:remove_method, :switch_user)
 572       cls.__send__(:alias_method, :switch_user, :mock_switch_user)
 573     end
 574
 575     def mock_switch_user(user, group = nil)
 576       $yahns_user = [ $$, user, group ]
 577     end
 578   end
 579
 580   def test_user_no_workers
 581     refute defined?($yahns_user), "$yahns_user global should be undefined"
 582     err = @err
 583     cfg = Yahns::Config.new
 584     host, port = @srv.addr[3], @srv.addr[1]
 585     cfg.instance_eval do
 586       ru = lambda do |_|
 587         b = $yahns_user.inspect
 588         [ 200, {'Content-Length'=>b.size.to_s }, [b] ]
 589       end
 590       GTL.synchronize { app(:rack, ru) { listen "#{host}:#{port}" } }
 591       user "nobody"
 592       stderr_path err.path
 593     end
 594     pid = mkserver(cfg) { Yahns::Server.__send__(:include, MockSwitchUser) }
 595     expect = [ pid, "nobody", nil ].inspect
 596     run_client(host, port) { |res| assert_equal expect, res.body }
 597     refute defined?($yahns_user), "$yahns_user global should be undefined"
 598   ensure
 599     quit_wait(pid)
 600   end
 601
 602   def test_user_workers
 603     refute defined?($yahns_user), "$yahns_user global should be undefined"
 604     err = @err
 605     cfg = Yahns::Config.new
 606     host, port = @srv.addr[3], @srv.addr[1]
 607     cfg.instance_eval do
 608       ru = lambda do |_|
 609         b = $yahns_user.inspect
 610         [ 200, {'Content-Length'=>b.size.to_s, 'X-Pid' => "#$$" }, [b] ]
 611       end
 612       GTL.synchronize { app(:rack, ru) { listen "#{host}:#{port}" } }
 613       user "nobody"
 614       stderr_path err.path
 615       worker_processes 1
 616     end
 617     pid = mkserver(cfg) { Yahns::Server.__send__(:include, MockSwitchUser) }
 618     run_client(host, port) do |res|
 619       worker_pid = res["X-Pid"].to_i
 620       assert_operator worker_pid, :>, 0
 621       refute_equal pid, worker_pid
 622       refute_equal $$, worker_pid
 623       expect = [ worker_pid, "nobody", nil ].inspect
 624       assert_equal expect, res.body
 625     end
 626     refute defined?($yahns_user), "$yahns_user global should be undefined"
 627   ensure
 628     quit_wait(pid)
 629   end
 630
 631   def test_working_directory
 632     err, cfg, host, port = @err, Yahns::Config.new, @srv.addr[3], @srv.addr[1]
 633     ru = lambda { |_|
 634       [ 200, {'Content-Length'=>Dir.pwd.size.to_s }, [Dir.pwd] ]
 635     }
 636     yahns_mktmpdir do |tmpdir|
 637       begin
 638         pid = mkserver(cfg) do
 639           $LOAD_PATH << File.expand_path("lib")
 640           cfg.instance_eval do
 641             working_directory tmpdir
 642             app(:rack, ru) { listen "#{host}:#{port}" }
 643             stderr_path err.path
 644           end
 645         end
 646         refute_equal Dir.pwd, tmpdir
 647         Net::HTTP.start(host, port) do |http|
 648           assert_equal tmpdir, http.request(Net::HTTP::Get.new("/")).body
 649         end
 650       ensure
 651         quit_wait(pid)
 652       end
 653     end
 654   end
 655
 656   def test_errors
 657     tmpdir = yahns_mktmpdir
 658     sock = "#{tmpdir}/sock"
 659     err, cfg, host, port = @err, Yahns::Config.new, @srv.addr[3], @srv.addr[1]
 660     re = tmpfile(%w(rack .errors))
 661     ru = lambda { |e|
 662       e["rack.errors"].write "INFO HIHI\n"
 663       [ 200, {'Content-Length'=>'2' }, %w(OK) ]
 664     }
 665     cfg.instance_eval do
 666       GTL.synchronize {
 667         app(:rack, ru) {
 668           listen "#{host}:#{port}"
 669           errors re.path
 670         }
 671         app(:rack, ru) { listen sock }
 672       }
 673       stderr_path err.path
 674     end
 675     pid = mkserver(cfg)
 676     Net::HTTP.start(host, port) do |http|
 677       assert_equal "OK", http.request(Net::HTTP::Get.new("/")).body
 678     end
 679     assert_equal "INFO HIHI\n", re.read
 680
 681     c = UNIXSocket.new(sock)
 682     c.write "GET /\r\n\r\n"
 683     assert_equal c, c.wait(30)
 684     assert_equal "OK", c.read
 685     c.close
 686     assert_match %r{INFO HIHI}, File.read(err.path)
 687   ensure
 688     re.close!
 689     quit_wait(pid)
 690     FileUtils.rm_rf(tmpdir)
 691   end
 692
 693   def test_persistent_shutdown_timeout; _persistent_shutdown(nil); end
 694   def test_persistent_shutdown_timeout_mp; _persistent_shutdown(1); end
 695
 696   def _persistent_shutdown(nr_workers)
 697     err, cfg, host, port = @err, Yahns::Config.new, @srv.addr[3], @srv.addr[1]
 698     pid = mkserver(cfg) do
 699       ru = lambda { |e| [ 200, {'Content-Length'=>'2'}, %w(OK) ] }
 700       cfg.instance_eval do
 701         app(:rack, ru) { listen "#{host}:#{port}" }
 702         stderr_path err.path
 703         shutdown_timeout 1
 704         worker_processes(nr_workers) if nr_workers
 705       end
 706     end
 707     c = get_tcp_client(host, port)
 708     c.write "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"
 709     assert_equal c, c.wait(30)
 710     buf = ''.dup
 711     re = /\r\n\r\nOK\z/
 712     Timeout.timeout(30) do
 713       begin
 714         buf << c.readpartial(666)
 715       end until re =~ buf
 716     end
 717     refute_match %r{Connection: close}, buf
 718     assert_nil c.wait(0.001), "connection should still be alive"
 719     Process.kill(:QUIT, pid)
 720     _, status = Timeout.timeout(5) { Process.waitpid2(pid) }
 721     assert status.success?, status.inspect
 722     assert_nil c.read(666)
 723   end
 724
 725   def test_slow_shutdown_timeout; _slow_shutdown(nil); end
 726   def test_slow_shutdown_timeout_mp; _slow_shutdown(1); end
 727
 728   def _slow_shutdown(nr_workers)
 729     err, cfg, host, port = @err, Yahns::Config.new, @srv.addr[3], @srv.addr[1]
 730     pid = mkserver(cfg) do
 731       ru = lambda { |e| [ 200, {'Content-Length'=>'2'}, %w(OK) ] }
 732       cfg.instance_eval do
 733         app(:rack, ru) { listen "#{host}:#{port}" }
 734         stderr_path err.path
 735         worker_processes(nr_workers) if nr_workers
 736       end
 737     end
 738     c = get_tcp_client(host, port)
 739     c.write 'G'
 740     100000.times { Thread.pass }
 741     Process.kill(:QUIT, pid)
 742     "ET / HTTP/1.1\r\nHost: example.com\r\n\r\n".each_byte do |x|
 743       Thread.pass
 744       c.write(x.chr)
 745       Thread.pass
 746     end
 747     assert_equal c, c.wait(30)
 748     buf = ''.dup
 749     re = /\r\n\r\nOK\z/
 750     Timeout.timeout(30) do
 751       begin
 752         buf << c.readpartial(666)
 753       end until re =~ buf
 754     end
 755     c.close
 756     _, status = Timeout.timeout(5) { Process.waitpid2(pid) }
 757     assert status.success?, status.inspect
 758   end
 759
 760   def test_before_exec
 761     err, cfg, host, port = @err, Yahns::Config.new, @srv.addr[3], @srv.addr[1]
 762     ru = lambda { |e| [ 200, {'Content-Length'=>'2' }, %w(OK) ] }
 763     tmp = tmpfile(%w(exec .pid))
 764     x = "echo $$ >> #{tmp.path}"
 765     pid = mkserver(cfg) do
 766       cfg.instance_eval do
 767         app(:rack, ru) { listen "#{host}:#{port}" }
 768         before_exec do |exec_cmd|
 769           exec_cmd.replace(%W(/bin/sh -c #{x}))
 770         end
 771         stderr_path err.path
 772       end
 773     end
 774
 775     # did we start properly?
 776     Net::HTTP.start(host, port) do |http|
 777       assert_equal "OK", http.request(Net::HTTP::Get.new("/")).body
 778     end
 779
 780     Process.kill(:USR2, pid)
 781     Timeout.timeout(30) { sleep(0.01) until tmp.size > 0 }
 782     buf = tmp.read
 783     assert_match %r{\A\d+}, buf
 784     exec_pid = buf.to_i
 785     poke_until_dead exec_pid
 786
 787     # ensure it recovered
 788     Net::HTTP.start(host, port) do |http|
 789       assert_equal "OK", http.request(Net::HTTP::Get.new("/")).body
 790     end
 791     assert_match %r{reaped}, err.read
 792     err.truncate(0)
 793   ensure
 794     tmp.close!
 795     quit_wait(pid)
 796   end
 797
 798   def test_app_controls_close
 799     err, cfg, host, port = @err, Yahns::Config.new, @srv.addr[3], @srv.addr[1]
 800     pid = mkserver(cfg) do
 801       cfg.instance_eval do
 802         ru = lambda { |env|
 803           h = { 'Content-Length' => '2' }
 804           if env["PATH_INFO"] =~ %r{\A/(.+)}
 805             h["Connection"] = $1
 806           end
 807           [ 200, h, ['HI'] ]
 808         }
 809         app(:rack, ru) { listen "#{host}:#{port}" }
 810         stderr_path err.path
 811       end
 812     end
 813     c = get_tcp_client(host, port)
 814
 815     # normal response
 816     c.write "GET /keep-alive HTTP/1.1\r\nHost: example.com\r\n\r\n"
 817     buf = ''.dup
 818     Timeout.timeout(30) do
 819       buf << c.readpartial(4096) until buf =~ /HI\z/
 820     end
 821     assert_match %r{^Connection: keep-alive}, buf
 822     assert_raises(Errno::EAGAIN,IO::WaitReadable) { c.read_nonblock(666) }
 823
 824     # we allow whatever in the response, but don't send it
 825     c.write "GET /whatever HTTP/1.1\r\nHost: example.com\r\n\r\n"
 826     buf = ''.dup
 827     Timeout.timeout(30) do
 828       buf << c.readpartial(4096) until buf =~ /HI\z/
 829     end
 830     assert_match %r{^Connection: keep-alive}, buf
 831     assert_raises(Errno::EAGAIN,IO::WaitReadable) { c.read_nonblock(666) }
 832
 833     c.write "GET /close HTTP/1.1\r\nHost: example.com\r\n\r\n"
 834     buf = ''.dup
 835     Timeout.timeout(30) do
 836       buf << c.readpartial(4096) until buf =~ /HI\z/
 837     end
 838     assert_match %r{^Connection: close}, buf
 839     assert_equal c, IO.select([c], nil, nil, 30)[0][0]
 840     assert_raises(EOFError) { c.readpartial(666) }
 841     c.close
 842   ensure
 843     quit_wait(pid)
 844   end
 845
 846   def test_inherit_too_many
 847     err = @err
 848     s2 = TCPServer.new(ENV["TEST_HOST"] || "127.0.0.1", 0)
 849     cfg = Yahns::Config.new
 850     host, port = @srv.addr[3], @srv.addr[1]
 851     cfg.instance_eval do
 852       ru = lambda { |_| [ 200, {'Content-Length'=>'2'}, ['HI'] ] }
 853       GTL.synchronize { app(:rack, ru) { listen "#{host}:#{port}" } }
 854       logger(Logger.new(err.path))
 855     end
 856     pid = mkserver(cfg, @srv) do
 857       s2.autoclose = false
 858       ENV["YAHNS_FD"] = "#{@srv.fileno},#{s2.fileno}"
 859     end
 860     run_client(host, port) { |res| assert_equal "HI", res.body }
 861     th = Thread.new do
 862       c = s2.accept
 863       c.readpartial(1234)
 864       c.write "HTTP/1.0 666 OK\r\n\r\nGO AWAY"
 865       c.close
 866       :OK
 867     end
 868     Thread.pass
 869     s2host, s2port = s2.addr[3], s2.addr[1]
 870     Net::HTTP.start(s2host, s2port) do |http|
 871       res = http.request(Net::HTTP::Get.new("/"))
 872       assert_equal 666, res.code.to_i
 873       assert_equal "GO AWAY", res.body
 874     end
 875     assert_equal :OK, th.value
 876     tmpc = TCPSocket.new(s2host, s2port)
 877     a2 = s2.accept
 878     assert_nil IO.select([a2], nil, nil, 0.05)
 879     tmpc.close
 880     assert_nil a2.read(1)
 881     a2.close
 882     s2.close
 883   ensure
 884     quit_wait(pid)
 885   end
 886
 887   def test_inherit_tcp_nodelay_set
 888     err = @err
 889     cfg = Yahns::Config.new
 890     host, port = @srv.addr[3], @srv.addr[1]
 891     @srv.setsockopt(:IPPROTO_TCP, :TCP_NODELAY, 0)
 892     assert_equal 0, @srv.getsockopt(:IPPROTO_TCP, :TCP_NODELAY).int
 893     cfg.instance_eval do
 894       ru = lambda { |_| [ 200, { 'Content-Length' => '2' } , [ 'HI' ] ] }
 895       GTL.synchronize { app(:rack, ru) { listen "#{host}:#{port}" } }
 896       logger(Logger.new(err.path))
 897     end
 898     pid = mkserver(cfg, @srv) { ENV["YAHNS_FD"] = "#{@srv.fileno}" }
 899     run_client(host, port) { |res| assert_equal "HI", res.body }
 900
 901     # TCP socket option is shared at file level, not FD level:
 902     assert_equal 1, @srv.getsockopt(:IPPROTO_TCP, :TCP_NODELAY).int
 903   ensure
 904     quit_wait(pid)
 905   end
 906 end