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