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 test_tcp
  16     s = TCPServer.new(TEST_ADDR, 0)
  17     port = s.addr[1]
  18     addr = "[#{TEST_ADDR}]:#{port}"
  19     addrs = [ addr ]
  20     stats = tcp_listener_stats(addrs)
  21     assert_equal 1, stats.size
  22     assert_equal 0, stats[addr].queued
  23     assert_equal 0, stats[addr].active
  24
  25     @to_close << TCPSocket.new(TEST_ADDR, port)
  26     stats = tcp_listener_stats(addrs)
  27     assert_equal 1, stats.size
  28     assert_equal 1, stats[addr].queued
  29     assert_equal 0, stats[addr].active
  30
  31     @to_close << s.accept
  32     stats = tcp_listener_stats(addrs)
  33     assert_equal 1, stats.size
  34     assert_equal 0, stats[addr].queued
  35     assert_equal 1, stats[addr].active
  36   end
  37
  38   def test_tcp_multi
  39     s1 = TCPServer.new(TEST_ADDR, 0)
  40     s2 = TCPServer.new(TEST_ADDR, 0)
  41     port1, port2 = s1.addr[1], s2.addr[1]
  42     addr1, addr2 = "[#{TEST_ADDR}]:#{port1}", "[#{TEST_ADDR}]:#{port2}"
  43     addrs = [ addr1, addr2 ]
  44     stats = tcp_listener_stats(addrs)
  45     assert_equal 2, stats.size
  46     assert_equal 0, stats[addr1].queued
  47     assert_equal 0, stats[addr1].active
  48     assert_equal 0, stats[addr2].queued
  49     assert_equal 0, stats[addr2].active
  50
  51     @to_close << TCPSocket.new(TEST_ADDR, port1)
  52     stats = tcp_listener_stats(addrs)
  53     assert_equal 2, stats.size
  54     assert_equal 1, 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     sc1 = s1.accept
  60     stats = tcp_listener_stats(addrs)
  61     assert_equal 2, stats.size
  62     assert_equal 0, stats[addr1].queued
  63     assert_equal 1, stats[addr1].active
  64     assert_equal 0, stats[addr2].queued
  65     assert_equal 0, stats[addr2].active
  66
  67     @to_close << TCPSocket.new(TEST_ADDR, port2)
  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 1, 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 2, stats[addr2].queued
  81     assert_equal 0, stats[addr2].active
  82
  83     @to_close << s2.accept
  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 1, stats[addr2].queued
  89     assert_equal 1, stats[addr2].active
  90
  91     sc1.close
  92     stats = tcp_listener_stats(addrs)
  93     assert_equal 0, stats[addr1].queued
  94     assert_equal 0, stats[addr1].active
  95     assert_equal 1, stats[addr2].queued
  96     assert_equal 1, stats[addr2].active
  97   end
  98
  99   def test_invalid_addresses
 100     assert_raises(ArgumentError) { tcp_listener_stats(%w([1:::5)) }
 101     assert_raises(ArgumentError) { tcp_listener_stats(%w([1:::]5)) }
 102   end
 103
 104   # tries to overflow buffers
 105   def test_tcp_stress_test
 106     nr_proc = 32
 107     nr_sock = 500
 108     s = TCPServer.new(TEST_ADDR, 0)
 109     port = s.addr[1]
 110     addr = "[#{TEST_ADDR}]:#{port}"
 111     addrs = [ addr ]
 112     rda, wra = IO.pipe
 113     rdb, wrb = IO.pipe
 114
 115     nr_proc.times do
 116       fork do
 117         rda.close
 118         wrb.close
 119         @to_close.concat((1..nr_sock).map { s.accept })
 120         wra.syswrite('.')
 121         wra.close
 122         rdb.sysread(1) # wait for parent to nuke us
 123       end
 124     end
 125
 126     nr_proc.times do
 127       fork do
 128         rda.close
 129         wrb.close
 130         @to_close.concat((1..nr_sock).map { TCPSocket.new(TEST_ADDR, port) })
 131         wra.syswrite('.')
 132         wra.close
 133         rdb.sysread(1) # wait for parent to nuke us
 134       end
 135     end
 136
 137     assert_equal('.' * (nr_proc * 2), rda.read(nr_proc * 2))
 138
 139     rda.close
 140     stats = tcp_listener_stats(addrs)
 141     expect = { addr => Raindrops::ListenStats[nr_sock * nr_proc, 0] }
 142     assert_equal expect, stats
 143
 144     @to_close << TCPSocket.new(TEST_ADDR, port)
 145     stats = tcp_listener_stats(addrs)
 146     expect = { addr => Raindrops::ListenStats[nr_sock * nr_proc, 1] }
 147     assert_equal expect, stats
 148
 149     if ENV["BENCHMARK"].to_i != 0
 150       require 'benchmark'
 151       puts(Benchmark.measure{1000.times { tcp_listener_stats(addrs) }})
 152     end
 153
 154     wrb.syswrite('.' * (nr_proc * 2)) # broadcast a wakeup
 155     statuses = Process.waitall
 156     statuses.each { |(_,status)| assert status.success?, status.inspect }
 157   end if ENV["STRESS"].to_i != 0
 158 end if RUBY_PLATFORM =~ /linux/ && ipv6_enabled?