test/test_linux.rb

   1 # -*- encoding: binary -*-
   2 require 'test/unit'
   3 require 'tempfile'
   4 require 'raindrops'
   5 require 'socket'
   6 require 'pp'
   7 $stderr.sync = $stdout.sync = true
   8
   9 class TestLinux < Test::Unit::TestCase
  10   include Raindrops::Linux
  11
  12   TEST_ADDR = ENV['UNICORN_TEST_ADDR'] || '127.0.0.1'
  13
  14   def test_unix
  15     tmp = Tempfile.new("\xde\xad\xbe\xef") # valid path, really :)
  16     File.unlink(tmp.path)
  17     us = UNIXServer.new(tmp.path)
  18     stats = unix_listener_stats([tmp.path])
  19     assert_equal 1, stats.size
  20     assert_equal 0, stats[tmp.path].active
  21     assert_equal 0, stats[tmp.path].queued
  22
  23     uc0 = UNIXSocket.new(tmp.path)
  24     stats = unix_listener_stats([tmp.path])
  25     assert_equal 1, stats.size
  26     assert_equal 0, stats[tmp.path].active
  27     assert_equal 1, stats[tmp.path].queued
  28
  29     uc1 = UNIXSocket.new(tmp.path)
  30     stats = unix_listener_stats([tmp.path])
  31     assert_equal 1, stats.size
  32     assert_equal 0, stats[tmp.path].active
  33     assert_equal 2, stats[tmp.path].queued
  34
  35     ua0 = us.accept
  36     stats = unix_listener_stats([tmp.path])
  37     assert_equal 1, stats.size
  38     assert_equal 1, stats[tmp.path].active
  39     assert_equal 1, stats[tmp.path].queued
  40   end
  41
  42   def test_tcp
  43     s = TCPServer.new(TEST_ADDR, 0)
  44     port = s.addr[1]
  45     addr = "#{TEST_ADDR}:#{port}"
  46     addrs = [ addr ]
  47     stats = tcp_listener_stats(addrs)
  48     assert_equal 1, stats.size
  49     assert_equal 0, stats[addr].queued
  50     assert_equal 0, stats[addr].active
  51
  52     c = TCPSocket.new(TEST_ADDR, port)
  53     stats = tcp_listener_stats(addrs)
  54     assert_equal 1, stats.size
  55     assert_equal 1, stats[addr].queued
  56     assert_equal 0, stats[addr].active
  57
  58     sc = s.accept
  59     stats = tcp_listener_stats(addrs)
  60     assert_equal 1, stats.size
  61     assert_equal 0, stats[addr].queued
  62     assert_equal 1, stats[addr].active
  63   end
  64
  65   def test_tcp_reuse_sock
  66     nlsock = Raindrops::InetDiagSocket.new
  67     s = TCPServer.new(TEST_ADDR, 0)
  68     port = s.addr[1]
  69     addr = "#{TEST_ADDR}:#{port}"
  70     addrs = [ addr ]
  71     stats = tcp_listener_stats(addrs, nlsock)
  72     assert_equal 1, stats.size
  73     assert_equal 0, stats[addr].queued
  74     assert_equal 0, stats[addr].active
  75
  76     c = TCPSocket.new(TEST_ADDR, port)
  77     stats = tcp_listener_stats(addrs, nlsock)
  78     assert_equal 1, stats.size
  79     assert_equal 1, stats[addr].queued
  80     assert_equal 0, stats[addr].active
  81
  82     sc = s.accept
  83     stats = tcp_listener_stats(addrs, nlsock)
  84     assert_equal 1, stats.size
  85     assert_equal 0, stats[addr].queued
  86     assert_equal 1, stats[addr].active
  87     ensure
  88       nlsock.close
  89   end
  90
  91   def test_tcp_multi
  92     s1 = TCPServer.new(TEST_ADDR, 0)
  93     s2 = TCPServer.new(TEST_ADDR, 0)
  94     port1, port2 = s1.addr[1], s2.addr[1]
  95     addr1, addr2 = "#{TEST_ADDR}:#{port1}", "#{TEST_ADDR}:#{port2}"
  96     addrs = [ addr1, addr2 ]
  97     stats = tcp_listener_stats(addrs)
  98     assert_equal 2, stats.size
  99     assert_equal 0, stats[addr1].queued
 100     assert_equal 0, stats[addr1].active
 101     assert_equal 0, stats[addr2].queued
 102     assert_equal 0, stats[addr2].active
 103
 104     c1 = TCPSocket.new(TEST_ADDR, port1)
 105     stats = tcp_listener_stats(addrs)
 106     assert_equal 2, stats.size
 107     assert_equal 1, stats[addr1].queued
 108     assert_equal 0, stats[addr1].active
 109     assert_equal 0, stats[addr2].queued
 110     assert_equal 0, stats[addr2].active
 111
 112     sc1 = s1.accept
 113     stats = tcp_listener_stats(addrs)
 114     assert_equal 2, stats.size
 115     assert_equal 0, stats[addr1].queued
 116     assert_equal 1, stats[addr1].active
 117     assert_equal 0, stats[addr2].queued
 118     assert_equal 0, stats[addr2].active
 119
 120     c2 = TCPSocket.new(TEST_ADDR, port2)
 121     stats = tcp_listener_stats(addrs)
 122     assert_equal 2, stats.size
 123     assert_equal 0, stats[addr1].queued
 124     assert_equal 1, stats[addr1].active
 125     assert_equal 1, stats[addr2].queued
 126     assert_equal 0, stats[addr2].active
 127
 128     c3 = TCPSocket.new(TEST_ADDR, port2)
 129     stats = tcp_listener_stats(addrs)
 130     assert_equal 2, stats.size
 131     assert_equal 0, stats[addr1].queued
 132     assert_equal 1, stats[addr1].active
 133     assert_equal 2, stats[addr2].queued
 134     assert_equal 0, stats[addr2].active
 135
 136     sc2 = s2.accept
 137     stats = tcp_listener_stats(addrs)
 138     assert_equal 2, stats.size
 139     assert_equal 0, stats[addr1].queued
 140     assert_equal 1, stats[addr1].active
 141     assert_equal 1, stats[addr2].queued
 142     assert_equal 1, stats[addr2].active
 143
 144     sc1.close
 145     stats = tcp_listener_stats(addrs)
 146     assert_equal 0, stats[addr1].queued
 147     assert_equal 0, stats[addr1].active
 148     assert_equal 1, stats[addr2].queued
 149     assert_equal 1, stats[addr2].active
 150   end
 151
 152   # tries to overflow buffers
 153   def test_tcp_stress_test
 154     nr_proc = 32
 155     nr_sock = 500
 156     s = TCPServer.new(TEST_ADDR, 0)
 157     port = s.addr[1]
 158     addr = "#{TEST_ADDR}:#{port}"
 159     addrs = [ addr ]
 160     rda, wra = IO.pipe
 161     rdb, wrb = IO.pipe
 162
 163     nr_proc.times do
 164       fork do
 165         rda.close
 166         wrb.close
 167         socks = (1..nr_sock).map { s.accept }
 168         wra.syswrite('.')
 169         wra.close
 170         rdb.sysread(1) # wait for parent to nuke us
 171       end
 172     end
 173
 174     nr_proc.times do
 175       fork do
 176         rda.close
 177         wrb.close
 178         socks = (1..nr_sock).map { TCPSocket.new(TEST_ADDR, port) }
 179         wra.syswrite('.')
 180         wra.close
 181         rdb.sysread(1) # wait for parent to nuke us
 182       end
 183     end
 184
 185     assert_equal('.' * (nr_proc * 2), rda.read(nr_proc * 2))
 186
 187     rda.close
 188     stats = tcp_listener_stats(addrs)
 189     expect = { addr => Raindrops::ListenStats[nr_sock * nr_proc, 0] }
 190     assert_equal expect, stats
 191
 192     uno_mas = TCPSocket.new(TEST_ADDR, port)
 193     stats = tcp_listener_stats(addrs)
 194     expect = { addr => Raindrops::ListenStats[nr_sock * nr_proc, 1] }
 195     assert_equal expect, stats
 196
 197     if ENV["BENCHMARK"].to_i != 0
 198       require 'benchmark'
 199       puts(Benchmark.measure{1000.times { tcp_listener_stats(addrs) }})
 200     end
 201
 202     wrb.syswrite('.' * (nr_proc * 2)) # broadcast a wakeup
 203     statuses = Process.waitall
 204     statuses.each { |(pid,status)| assert status.success?, status.inspect }
 205   end if ENV["STRESS"].to_i != 0
 206 end if RUBY_PLATFORM =~ /linux/