test/test_linux_ipv6.rb

   1 # -*- encoding: binary -*-
   2 require 'test/unit'
   3 require 'tempfile'
   4 require 'raindrops'
   5 require 'socket'
   6 require 'pp'
   7 require "./test/ipv6_enabled"
   8 $stderr.sync = $stdout.sync = true
   9
  10 class TestLinuxIPv6 < Test::Unit::TestCase
  11   include Raindrops::Linux
  12
  13   TEST_ADDR = ENV["TEST_HOST6"] || "::1"
  14
  15   def setup
  16     @to_close = []
  17   end
  18
  19   def teardown
  20     @to_close.each { |io| io.close unless io.closed? }
  21   end
  22
  23   def test_tcp
  24     s = TCPServer.new(TEST_ADDR, 0)
  25     port = s.addr[1]
  26     addr = "[#{TEST_ADDR}]:#{port}"
  27     addrs = [ addr ]
  28     stats = tcp_listener_stats(addrs)
  29     assert_equal 1, stats.size
  30     assert_equal 0, stats[addr].queued
  31     assert_equal 0, stats[addr].active
  32
  33     @to_close << TCPSocket.new(TEST_ADDR, port)
  34     stats = tcp_listener_stats(addrs)
  35     assert_equal 1, stats.size
  36     assert_equal 1, stats[addr].queued
  37     assert_equal 0, stats[addr].active
  38
  39     @to_close << s.accept
  40     stats = tcp_listener_stats(addrs)
  41     assert_equal 1, stats.size
  42     assert_equal 0, stats[addr].queued
  43     assert_equal 1, stats[addr].active
  44   end
  45
  46   def test_tcp_multi
  47     s1 = TCPServer.new(TEST_ADDR, 0)
  48     s2 = TCPServer.new(TEST_ADDR, 0)
  49     port1, port2 = s1.addr[1], s2.addr[1]
  50     addr1, addr2 = "[#{TEST_ADDR}]:#{port1}", "[#{TEST_ADDR}]:#{port2}"
  51     addrs = [ addr1, addr2 ]
  52     stats = tcp_listener_stats(addrs)
  53     assert_equal 2, stats.size
  54     assert_equal 0, stats[addr1].queued
  55     assert_equal 0, stats[addr1].active
  56     assert_equal 0, stats[addr2].queued
  57     assert_equal 0, stats[addr2].active
  58
  59     @to_close << TCPSocket.new(TEST_ADDR, port1)
  60     stats = tcp_listener_stats(addrs)
  61     assert_equal 2, stats.size
  62     assert_equal 1, stats[addr1].queued
  63     assert_equal 0, stats[addr1].active
  64     assert_equal 0, stats[addr2].queued
  65     assert_equal 0, stats[addr2].active
  66
  67     sc1 = s1.accept
  68     stats = tcp_listener_stats(addrs)
  69     assert_equal 2, stats.size
  70     assert_equal 0, stats[addr1].queued
  71     assert_equal 1, stats[addr1].active
  72     assert_equal 0, stats[addr2].queued
  73     assert_equal 0, stats[addr2].active
  74
  75     @to_close << TCPSocket.new(TEST_ADDR, port2)
  76     stats = tcp_listener_stats(addrs)
  77     assert_equal 2, stats.size
  78     assert_equal 0, stats[addr1].queued
  79     assert_equal 1, stats[addr1].active
  80     assert_equal 1, stats[addr2].queued
  81     assert_equal 0, stats[addr2].active
  82
  83     @to_close << TCPSocket.new(TEST_ADDR, port2)
  84     stats = tcp_listener_stats(addrs)
  85     assert_equal 2, stats.size
  86     assert_equal 0, stats[addr1].queued
  87     assert_equal 1, stats[addr1].active
  88     assert_equal 2, stats[addr2].queued
  89     assert_equal 0, stats[addr2].active
  90
  91     @to_close << s2.accept
  92     stats = tcp_listener_stats(addrs)
  93     assert_equal 2, stats.size
  94     assert_equal 0, stats[addr1].queued
  95     assert_equal 1, stats[addr1].active
  96     assert_equal 1, stats[addr2].queued
  97     assert_equal 1, stats[addr2].active
  98
  99     sc1.close
 100     stats = tcp_listener_stats(addrs)
 101     assert_equal 0, stats[addr1].queued
 102     assert_equal 0, stats[addr1].active
 103     assert_equal 1, stats[addr2].queued
 104     assert_equal 1, stats[addr2].active
 105   end
 106
 107   def test_invalid_addresses
 108     assert_raises(ArgumentError) { tcp_listener_stats(%w([1:::5)) }
 109     assert_raises(ArgumentError) { tcp_listener_stats(%w([1:::]5)) }
 110   end
 111
 112   # tries to overflow buffers
 113   def test_tcp_stress_test
 114     nr_proc = 32
 115     nr_sock = 500
 116     s = TCPServer.new(TEST_ADDR, 0)
 117     port = s.addr[1]
 118     addr = "[#{TEST_ADDR}]:#{port}"
 119     addrs = [ addr ]
 120     rda, wra = IO.pipe
 121     rdb, wrb = IO.pipe
 122
 123     nr_proc.times do
 124       fork do
 125         rda.close
 126         wrb.close
 127         @to_close.concat((1..nr_sock).map { s.accept })
 128         wra.syswrite('.')
 129         wra.close
 130         rdb.sysread(1) # wait for parent to nuke us
 131       end
 132     end
 133
 134     nr_proc.times do
 135       fork do
 136         rda.close
 137         wrb.close
 138         @to_close.concat((1..nr_sock).map { TCPSocket.new(TEST_ADDR, port) })
 139         wra.syswrite('.')
 140         wra.close
 141         rdb.sysread(1) # wait for parent to nuke us
 142       end
 143     end
 144
 145     assert_equal('.' * (nr_proc * 2), rda.read(nr_proc * 2))
 146
 147     rda.close
 148     stats = tcp_listener_stats(addrs)
 149     expect = { addr => Raindrops::ListenStats[nr_sock * nr_proc, 0] }
 150     assert_equal expect, stats
 151
 152     @to_close << TCPSocket.new(TEST_ADDR, port)
 153     stats = tcp_listener_stats(addrs)
 154     expect = { addr => Raindrops::ListenStats[nr_sock * nr_proc, 1] }
 155     assert_equal expect, stats
 156
 157     if ENV["BENCHMARK"].to_i != 0
 158       require 'benchmark'
 159       puts(Benchmark.measure{1000.times { tcp_listener_stats(addrs) }})
 160     end
 161
 162     wrb.syswrite('.' * (nr_proc * 2)) # broadcast a wakeup
 163     statuses = Process.waitall
 164     statuses.each { |(_,status)| assert status.success?, status.inspect }
 165   end if ENV["STRESS"].to_i != 0
 166 end if RUBY_PLATFORM =~ /linux/ && ipv6_enabled?