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 setup
  15     @to_close = []
  16   end
  17
  18   def teardown
  19     @to_close.each { |io| io.close unless io.closed? }
  20   end
  21
  22   def test_unix
  23     tmp = Tempfile.new("\xde\xad\xbe\xef") # valid path, really :)
  24     File.unlink(tmp.path)
  25     us = UNIXServer.new(tmp.path)
  26     stats = unix_listener_stats([tmp.path])
  27     assert_equal 1, stats.size
  28     assert_equal 0, stats[tmp.path].active
  29     assert_equal 0, stats[tmp.path].queued
  30
  31     @to_close << UNIXSocket.new(tmp.path)
  32     stats = unix_listener_stats([tmp.path])
  33     assert_equal 1, stats.size
  34     assert_equal 0, stats[tmp.path].active
  35     assert_equal 1, stats[tmp.path].queued
  36
  37     @to_close << UNIXSocket.new(tmp.path)
  38     stats = unix_listener_stats([tmp.path])
  39     assert_equal 1, stats.size
  40     assert_equal 0, stats[tmp.path].active
  41     assert_equal 2, stats[tmp.path].queued
  42
  43     @to_close << us.accept
  44     stats = unix_listener_stats([tmp.path])
  45     assert_equal 1, stats.size
  46     assert_equal 1, stats[tmp.path].active
  47     assert_equal 1, stats[tmp.path].queued
  48   end
  49
  50   def test_unix_all
  51     tmp = Tempfile.new("\xde\xad\xbe\xef") # valid path, really :)
  52     File.unlink(tmp.path)
  53     us = UNIXServer.new(tmp.path)
  54     @to_close << UNIXSocket.new(tmp.path)
  55     stats = unix_listener_stats
  56     assert_equal 0, stats[tmp.path].active
  57     assert_equal 1, stats[tmp.path].queued
  58
  59     @to_close << UNIXSocket.new(tmp.path)
  60     stats = unix_listener_stats
  61     assert_equal 0, stats[tmp.path].active
  62     assert_equal 2, stats[tmp.path].queued
  63
  64     @to_close << us.accept
  65     stats = unix_listener_stats
  66     assert_equal 1, stats[tmp.path].active
  67     assert_equal 1, stats[tmp.path].queued
  68   end
  69
  70   def test_unix_resolves_symlinks
  71     tmp = Tempfile.new("\xde\xad\xbe\xef") # valid path, really :)
  72     File.unlink(tmp.path)
  73     us = UNIXServer.new(tmp.path)
  74
  75     # Create a symlink
  76     link = Tempfile.new("somethingelse")
  77     File.unlink(link.path) # We need an available name, not an actual file
  78     File.symlink(tmp.path, link.path)
  79
  80     @to_close << UNIXSocket.new(tmp.path)
  81     stats = unix_listener_stats
  82     assert_equal 0, stats[tmp.path].active
  83     assert_equal 1, stats[tmp.path].queued
  84
  85     @to_close << UNIXSocket.new(link.path)
  86     stats = unix_listener_stats([link.path])
  87     assert_equal 0, stats[link.path].active
  88     assert_equal 2, stats[link.path].queued
  89
  90     assert_equal stats[link.path].object_id, stats[tmp.path].object_id
  91
  92     @to_close << us.accept
  93     stats = unix_listener_stats
  94     assert_equal 1, stats[tmp.path].active
  95     assert_equal 1, stats[tmp.path].queued
  96   end
  97
  98   def test_tcp
  99     s = TCPServer.new(TEST_ADDR, 0)
 100     port = s.addr[1]
 101     addr = "#{TEST_ADDR}:#{port}"
 102     addrs = [ addr ]
 103     stats = tcp_listener_stats(addrs)
 104     assert_equal 1, stats.size
 105     assert_equal 0, stats[addr].queued
 106     assert_equal 0, stats[addr].active
 107
 108     @to_close << TCPSocket.new(TEST_ADDR, port)
 109     stats = tcp_listener_stats(addrs)
 110     assert_equal 1, stats.size
 111     assert_equal 1, stats[addr].queued
 112     assert_equal 0, stats[addr].active
 113
 114     @to_close << s.accept
 115     stats = tcp_listener_stats(addrs)
 116     assert_equal 1, stats.size
 117     assert_equal 0, stats[addr].queued
 118     assert_equal 1, stats[addr].active
 119   end
 120
 121   def test_tcp_reuse_sock
 122     nlsock = Raindrops::InetDiagSocket.new
 123     s = TCPServer.new(TEST_ADDR, 0)
 124     port = s.addr[1]
 125     addr = "#{TEST_ADDR}:#{port}"
 126     addrs = [ addr ]
 127     stats = tcp_listener_stats(addrs, nlsock)
 128     assert_equal 1, stats.size
 129     assert_equal 0, stats[addr].queued
 130     assert_equal 0, stats[addr].active
 131
 132     @to_close << TCPSocket.new(TEST_ADDR, port)
 133     stats = tcp_listener_stats(addrs, nlsock)
 134     assert_equal 1, stats.size
 135     assert_equal 1, stats[addr].queued
 136     assert_equal 0, stats[addr].active
 137
 138     @to_close << s.accept
 139     stats = tcp_listener_stats(addrs, nlsock)
 140     assert_equal 1, stats.size
 141     assert_equal 0, stats[addr].queued
 142     assert_equal 1, stats[addr].active
 143     ensure
 144       nlsock.close
 145   end
 146
 147   def test_tcp_multi
 148     s1 = TCPServer.new(TEST_ADDR, 0)
 149     s2 = TCPServer.new(TEST_ADDR, 0)
 150     port1, port2 = s1.addr[1], s2.addr[1]
 151     addr1, addr2 = "#{TEST_ADDR}:#{port1}", "#{TEST_ADDR}:#{port2}"
 152     addrs = [ addr1, addr2 ]
 153     stats = tcp_listener_stats(addrs)
 154     assert_equal 2, stats.size
 155     assert_equal 0, stats[addr1].queued
 156     assert_equal 0, stats[addr1].active
 157     assert_equal 0, stats[addr2].queued
 158     assert_equal 0, stats[addr2].active
 159
 160     @to_close << TCPSocket.new(TEST_ADDR, port1)
 161     stats = tcp_listener_stats(addrs)
 162     assert_equal 2, stats.size
 163     assert_equal 1, stats[addr1].queued
 164     assert_equal 0, stats[addr1].active
 165     assert_equal 0, stats[addr2].queued
 166     assert_equal 0, stats[addr2].active
 167
 168     sc1 = s1.accept
 169     stats = tcp_listener_stats(addrs)
 170     assert_equal 2, stats.size
 171     assert_equal 0, stats[addr1].queued
 172     assert_equal 1, stats[addr1].active
 173     assert_equal 0, stats[addr2].queued
 174     assert_equal 0, stats[addr2].active
 175
 176     @to_close << TCPSocket.new(TEST_ADDR, port2)
 177     stats = tcp_listener_stats(addrs)
 178     assert_equal 2, stats.size
 179     assert_equal 0, stats[addr1].queued
 180     assert_equal 1, stats[addr1].active
 181     assert_equal 1, stats[addr2].queued
 182     assert_equal 0, stats[addr2].active
 183
 184     @to_close << TCPSocket.new(TEST_ADDR, port2)
 185     stats = tcp_listener_stats(addrs)
 186     assert_equal 2, stats.size
 187     assert_equal 0, stats[addr1].queued
 188     assert_equal 1, stats[addr1].active
 189     assert_equal 2, stats[addr2].queued
 190     assert_equal 0, stats[addr2].active
 191
 192     @to_close << s2.accept
 193     stats = tcp_listener_stats(addrs)
 194     assert_equal 2, stats.size
 195     assert_equal 0, stats[addr1].queued
 196     assert_equal 1, stats[addr1].active
 197     assert_equal 1, stats[addr2].queued
 198     assert_equal 1, stats[addr2].active
 199
 200     sc1.close
 201     stats = tcp_listener_stats(addrs)
 202     assert_equal 0, stats[addr1].queued
 203     assert_equal 0, stats[addr1].active
 204     assert_equal 1, stats[addr2].queued
 205     assert_equal 1, stats[addr2].active
 206   end
 207
 208   # tries to overflow buffers
 209   def test_tcp_stress_test
 210     nr_proc = 32
 211     nr_sock = 500
 212     s = TCPServer.new(TEST_ADDR, 0)
 213     port = s.addr[1]
 214     addr = "#{TEST_ADDR}:#{port}"
 215     addrs = [ addr ]
 216     rda, wra = IO.pipe
 217     rdb, wrb = IO.pipe
 218
 219     nr_proc.times do
 220       fork do
 221         rda.close
 222         wrb.close
 223         @to_close.concat((1..nr_sock).map { s.accept })
 224         wra.syswrite('.')
 225         wra.close
 226         rdb.sysread(1) # wait for parent to nuke us
 227       end
 228     end
 229
 230     nr_proc.times do
 231       fork do
 232         rda.close
 233         wrb.close
 234         @to_close.concat((1..nr_sock).map { TCPSocket.new(TEST_ADDR, port) })
 235         wra.syswrite('.')
 236         wra.close
 237         rdb.sysread(1) # wait for parent to nuke us
 238       end
 239     end
 240
 241     assert_equal('.' * (nr_proc * 2), rda.read(nr_proc * 2))
 242
 243     rda.close
 244     stats = tcp_listener_stats(addrs)
 245     expect = { addr => Raindrops::ListenStats[nr_sock * nr_proc, 0] }
 246     assert_equal expect, stats
 247
 248     @to_close << TCPSocket.new(TEST_ADDR, port)
 249     stats = tcp_listener_stats(addrs)
 250     expect = { addr => Raindrops::ListenStats[nr_sock * nr_proc, 1] }
 251     assert_equal expect, stats
 252
 253     if ENV["BENCHMARK"].to_i != 0
 254       require 'benchmark'
 255       puts(Benchmark.measure{1000.times { tcp_listener_stats(addrs) }})
 256     end
 257
 258     wrb.syswrite('.' * (nr_proc * 2)) # broadcast a wakeup
 259     statuses = Process.waitall
 260     statuses.each { |(_,status)| assert status.success?, status.inspect }
 261   end if ENV["STRESS"].to_i != 0
 262 end if RUBY_PLATFORM =~ /linux/