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     c = 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     sc = 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     c1 = 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     c2 = 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     c3 = 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     sc2 = 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   # tries to overflow buffers
 100   def test_tcp_stress_test
 101     nr_proc = 32
 102     nr_sock = 500
 103     s = TCPServer.new(TEST_ADDR, 0)
 104     port = s.addr[1]
 105     addr = "[#{TEST_ADDR}]:#{port}"
 106     addrs = [ addr ]
 107     rda, wra = IO.pipe
 108     rdb, wrb = IO.pipe
 109
 110     nr_proc.times do
 111       fork do
 112         rda.close
 113         wrb.close
 114         socks = (1..nr_sock).map { s.accept }
 115         wra.syswrite('.')
 116         wra.close
 117         rdb.sysread(1) # wait for parent to nuke us
 118       end
 119     end
 120
 121     nr_proc.times do
 122       fork do
 123         rda.close
 124         wrb.close
 125         socks = (1..nr_sock).map { TCPSocket.new(TEST_ADDR, port) }
 126         wra.syswrite('.')
 127         wra.close
 128         rdb.sysread(1) # wait for parent to nuke us
 129       end
 130     end
 131
 132     assert_equal('.' * (nr_proc * 2), rda.read(nr_proc * 2))
 133
 134     rda.close
 135     stats = tcp_listener_stats(addrs)
 136     expect = { addr => Raindrops::ListenStats[nr_sock * nr_proc, 0] }
 137     assert_equal expect, stats
 138
 139     uno_mas = TCPSocket.new(TEST_ADDR, port)
 140     stats = tcp_listener_stats(addrs)
 141     expect = { addr => Raindrops::ListenStats[nr_sock * nr_proc, 1] }
 142     assert_equal expect, stats
 143
 144     if ENV["BENCHMARK"].to_i != 0
 145       require 'benchmark'
 146       puts(Benchmark.measure{1000.times { tcp_listener_stats(addrs) }})
 147     end
 148
 149     wrb.syswrite('.' * (nr_proc * 2)) # broadcast a wakeup
 150     statuses = Process.waitall
 151     statuses.each { |(pid,status)| assert status.success?, status.inspect }
 152   end if ENV["STRESS"].to_i != 0
 153 end if RUBY_PLATFORM =~ /linux/ && ipv6_enabled?