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