test/test_linux_ipv6.rb

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