test/test_epoll.rb

   1 require 'test/unit'
   2 require 'fcntl'
   3 require 'socket'
   4 $-w = true
   5
   6 require 'sleepy_penguin'
   7
   8 class TestEpoll < Test::Unit::TestCase
   9   include SleepyPenguin
  10   RBX = defined?(RUBY_ENGINE) && (RUBY_ENGINE == 'rbx')
  11
  12   def setup
  13     @rd, @wr = IO.pipe
  14     @ep = Epoll.new
  15   end
  16
  17   def test_constants
  18     Epoll.constants.each do |const|
  19       next if const.to_sym == :IO
  20       nr = Epoll.const_get(const)
  21       assert nr <= 0xffffffff, "#{const}=#{nr}"
  22     end
  23   end
  24
  25   def test_cross_thread
  26     tmp = []
  27     Thread.new { sleep 0.100; @ep.add(@wr, Epoll::OUT) }
  28     t0 = Time.now
  29     @ep.wait { |flags,obj| tmp << [ flags, obj ] }
  30     elapsed = Time.now - t0
  31     assert elapsed >= 0.100
  32     assert_equal [[Epoll::OUT, @wr]], tmp
  33   end
  34
  35   def test_fork_safe
  36     tmp = []
  37     @ep.add @rd, Epoll::IN
  38     pid = fork do
  39       @ep.wait(nil, 100) { |flags,obj| tmp << [ flags, obj ] }
  40       exit!(tmp.empty?)
  41     end
  42     @wr.syswrite "HI"
  43     _, status = Process.waitpid2(pid)
  44     assert status.success?
  45     @ep.wait(nil, 0) { |flags,obj| tmp << [ flags, obj ] }
  46     assert_equal [[Epoll::IN, @rd]], tmp
  47   end
  48
  49   def test_after_fork_usability
  50     fork { @ep.add(@rd, Epoll::IN); exit!(0) }
  51     fork { @ep.set(@rd, Epoll::IN); exit!(0) }
  52     fork { @ep.to_io; exit!(0) }
  53     fork { @ep.dup; exit!(0) }
  54     fork { @ep.clone; exit!(0) }
  55     fork { @ep.close; exit!(0) }
  56     fork { @ep.closed?; exit!(0) }
  57     fork {
  58       begin
  59         @ep.del(@rd)
  60       rescue Errno::ENOENT
  61         exit!(0)
  62       end
  63       exit!(1)
  64     }
  65     res = Process.waitall
  66     res.each { |(_,status)| assert status.success? }
  67   end
  68
  69   def test_tcp_connect_nonblock_edge
  70     epflags = Epoll::OUT | Epoll::ET
  71     host = '127.0.0.1'
  72     srv = TCPServer.new(host, 0)
  73     port = srv.addr[1]
  74     addr = Socket.pack_sockaddr_in(port, host)
  75     sock = Socket.new(Socket::AF_INET, Socket::SOCK_STREAM, 0)
  76     assert_raises(Errno::EINPROGRESS) { sock.connect_nonblock(addr) }
  77     IO.select(nil, [ sock ], [sock ])
  78     @ep.add(sock, epflags)
  79     tmp = []
  80     @ep.wait(1) { |flags, obj| tmp << [ flags, obj ] }
  81     assert_equal [ [Epoll::OUT,  sock] ], tmp
  82   end
  83
  84   def test_tcp_connect_edge
  85     epflags = Epoll::OUT | Epoll::ET
  86     host = '127.0.0.1'
  87     srv = TCPServer.new(host, 0)
  88     port = srv.addr[1]
  89     sock = TCPSocket.new(host, port)
  90     @ep.add(sock, epflags)
  91     tmp = []
  92     @ep.wait(1) { |flags, obj| tmp << [ flags, obj ] }
  93     assert_equal [ [Epoll::OUT,  sock] ], tmp
  94   end
  95
  96   def teardown
  97     assert_nothing_raised do
  98       @rd.close unless @rd.closed?
  99       @wr.close unless @wr.closed?
 100       @ep.close unless @ep.closed?
 101     end
 102   end
 103
 104   def test_max_events_big
 105     @ep.add @rd, Epoll::IN
 106     tmp = []
 107     thr = Thread.new { @ep.wait(1024) { |flags, obj| tmp << [ flags, obj ] } }
 108     Thread.pass
 109     assert tmp.empty?
 110     @wr.write '.'
 111     thr.join
 112     assert_equal([[Epoll::IN, @rd]], tmp)
 113     tmp.clear
 114     thr = Thread.new { @ep.wait { |flags, obj| tmp << [ flags, obj ] } }
 115     thr.join
 116     assert_equal([[Epoll::IN, @rd]], tmp)
 117   end
 118
 119   def test_max_events_small
 120     @ep.add @rd, Epoll::IN | Epoll::ET
 121     @ep.add @wr, Epoll::OUT | Epoll::ET
 122     @wr.write '.'
 123     tmp = []
 124     @ep.wait(1) { |flags, obj| tmp << [ flags, obj ] }
 125     assert_equal 1, tmp.size
 126     @ep.wait(1) { |flags, obj| tmp << [ flags, obj ] }
 127     assert_equal 2, tmp.size
 128   end
 129
 130   def test_signal_safe_wait_forever
 131     time = {}
 132     trap(:USR1) do
 133       time[:USR1] = Time.now
 134       sleep 0.5
 135       @wr.write '.'
 136     end
 137     @ep.add @rd, Epoll::IN
 138     tmp = []
 139     pid = fork do
 140       sleep 0.5 # slightly racy :<
 141       Process.kill(:USR1, Process.ppid)
 142     end
 143     time[:START_WAIT] = Time.now
 144     assert_nothing_raised do
 145       @ep.wait do |flags, obj|
 146         tmp << [ flags, obj ]
 147         time[:EP] = Time.now
 148       end
 149     end
 150     assert_equal([[Epoll::IN, @rd]], tmp)
 151     _, status = Process.waitpid2(pid)
 152     assert status.success?, status.inspect
 153     usr1_delay = time[:USR1] - time[:START_WAIT]
 154     assert_in_delta(0.5, usr1_delay, 0.1, "usr1_delay=#{usr1_delay}")
 155     ep_delay = time[:EP] - time[:USR1]
 156     assert_in_delta(0.5, ep_delay, 0.1, "ep1_delay=#{ep_delay}")
 157     ensure
 158       trap(:USR1, 'DEFAULT')
 159   end unless RBX
 160
 161   def test_close
 162     @ep.add @rd, Epoll::IN
 163     tmp = []
 164     thr = Thread.new { @ep.wait { |flags, obj| tmp << [ flags, obj ] } }
 165     @rd.close
 166     @wr.close
 167     assert_nil thr.join(0.01)
 168     assert thr.alive?
 169     thr.kill
 170     assert tmp.empty?
 171   end
 172
 173   def test_rdhup
 174     defined?(Epoll::RDHUP) or
 175       return warn "skipping test, EPOLLRDHUP not available"
 176     rd, wr = UNIXSocket.pair
 177     @ep.add rd, Epoll::RDHUP
 178     tmp = []
 179     thr = Thread.new { @ep.wait { |flags, obj| tmp << [ flags, obj ] } }
 180     wr.shutdown Socket::SHUT_WR
 181     thr.join
 182     assert_equal([[ Epoll::RDHUP, rd ]], tmp)
 183   end
 184
 185   def test_hup
 186     @ep.add @rd, Epoll::IN
 187     tmp = []
 188     thr = Thread.new { @ep.wait { |flags, obj| tmp << [ flags, obj ] } }
 189     @wr.close
 190     thr.join
 191     assert_equal([[ Epoll::HUP, @rd ]], tmp)
 192   end
 193
 194   def test_multiple
 195     r, w = IO.pipe
 196     assert_nothing_raised do
 197       @ep.add r, Epoll::IN
 198       @ep.add @rd, Epoll::IN
 199       @ep.add w, Epoll::OUT
 200       @ep.add @wr, Epoll::OUT
 201     end
 202     tmp = []
 203     @ep.wait { |flags, obj| tmp << [ flags, obj ] }
 204     assert_equal 2, tmp.size
 205     assert_equal [ Epoll::OUT ], tmp.map { |flags, obj| flags }.uniq
 206     ios = tmp.map { |flags, obj| obj }
 207     assert ios.include?(@wr)
 208     assert ios.include?(w)
 209   end
 210
 211   def test_gc
 212     assert_nothing_raised { 4096.times { Epoll.new } }
 213     assert ! @ep.closed?
 214   end unless RBX
 215
 216   def test_gc_to_io
 217     assert_nothing_raised do
 218       4096.times do
 219         ep = Epoll.new
 220         assert_kind_of IO, ep.to_io
 221       end
 222     end
 223     assert ! @ep.closed?
 224   end unless RBX
 225
 226   def test_clone
 227     tmp = []
 228     clone = @ep.clone
 229     assert @ep.to_io.fileno != clone.to_io.fileno
 230     clone.add @wr, Epoll::OUT
 231     @ep.wait(nil, 0) { |flags, obj| tmp << [ flags, obj ] }
 232     assert_equal([[Epoll::OUT, @wr]], tmp)
 233     assert_nothing_raised { clone.close }
 234   end
 235
 236   def test_dup
 237     tmp = []
 238     clone = @ep.dup
 239     assert @ep.to_io.fileno != clone.to_io.fileno
 240     clone.add @wr, Epoll::OUT
 241     @ep.wait(nil, 0) { |flags, obj| tmp << [ flags, obj ] }
 242     assert_equal([[Epoll::OUT, @wr]], tmp)
 243     assert_nothing_raised { clone.close }
 244   end
 245
 246   def test_set_idempotency
 247     assert_nothing_raised do
 248       @ep.set @rd, Epoll::IN
 249       @ep.set @rd, Epoll::IN
 250       @ep.set @wr, Epoll::OUT
 251       @ep.set @wr, Epoll::OUT
 252     end
 253   end
 254
 255   def test_wait_timeout
 256     t0 = Time.now
 257     assert_equal 0, @ep.wait(nil, 100) { |flags,obj| assert false }
 258     diff = Time.now - t0
 259     assert(diff >= 0.075, "#{diff} < 0.100s")
 260   end
 261
 262   def test_del
 263     assert_raises(Errno::ENOENT) { @ep.del(@rd) }
 264     assert_nothing_raised do
 265       @ep.add(@rd, Epoll::IN)
 266       @ep.del(@rd)
 267     end
 268   end
 269
 270   def test_wait_read
 271     @ep.add(@rd, Epoll::IN)
 272     assert_equal 0, @ep.wait(nil, 0) { |flags,obj| assert false }
 273     @wr.syswrite '.'
 274     i = 0
 275     nr = @ep.wait(nil, 0) do |flags,obj|
 276       assert_equal Epoll::IN, flags
 277       assert_equal obj, @rd
 278       i += 1
 279     end
 280     assert_equal 1, i
 281     assert_equal 1, nr
 282   end
 283
 284   def test_wait_write
 285     @ep.add(@wr, Epoll::OUT | Epoll::IN)
 286     i = 0
 287     nr = @ep.wait(nil, 0) do |flags, obj|
 288       assert_equal Epoll::OUT, flags
 289       assert_equal obj, @wr
 290       i += 1
 291     end
 292     assert_equal 1, nr
 293     assert_equal 1, i
 294   end
 295
 296   def test_wait_write_blocked
 297     begin
 298       @wr.write_nonblock('.' * 65536)
 299     rescue Errno::EAGAIN
 300       break
 301     end while true
 302     @ep.add(@wr, Epoll::OUT | Epoll::IN)
 303     assert_equal 0, @ep.wait(nil, 0) { |flags,event| assert false }
 304   end
 305
 306   def test_selectable
 307     tmp = nil
 308     @ep.add @rd, Epoll::IN
 309     thr = Thread.new { tmp = IO.select([ @ep ]) }
 310     thr.join 0.01
 311     assert_nil tmp
 312     @wr.write '.'
 313     thr.join
 314     assert_equal([[@ep],[],[]], tmp)
 315   end
 316
 317   def test_new_no_cloexec
 318     @ep.close
 319     io = Epoll.new(0).to_io
 320     assert((io.fcntl(Fcntl::F_GETFD) & Fcntl::FD_CLOEXEC) == 0)
 321   end
 322
 323   def test_new_cloexec
 324     @ep.close
 325     io = Epoll.new(Epoll::CLOEXEC).to_io
 326     assert((io.fcntl(Fcntl::F_GETFD) & Fcntl::FD_CLOEXEC) == Fcntl::FD_CLOEXEC)
 327   end
 328
 329   def test_new
 330     @ep.close
 331     io = Epoll.new.to_io
 332     assert_equal 0, io.fcntl(Fcntl::F_GETFD)
 333   end
 334
 335   def test_delete
 336     assert_nil @ep.delete(@rd)
 337     assert_nil @ep.delete(@wr)
 338     assert_nothing_raised { @ep.add @rd, Epoll::IN }
 339     assert_equal @rd, @ep.delete(@rd)
 340     assert_nil @ep.delete(@rd)
 341   end
 342
 343   def test_io_for
 344     @ep.add @rd, Epoll::IN
 345     assert_equal @rd, @ep.io_for(@rd.fileno)
 346     assert_equal @rd, @ep.io_for(@rd)
 347     @ep.del @rd
 348     assert_nil @ep.io_for(@rd.fileno)
 349     assert_nil @ep.io_for(@rd)
 350   end
 351
 352   def test_flags_for
 353     @ep.add @rd, Epoll::IN
 354     assert_equal Epoll::IN, @ep.flags_for(@rd.fileno)
 355     assert_equal Epoll::IN, @ep.flags_for(@rd)
 356
 357     @ep.del @rd
 358     assert_nil @ep.flags_for(@rd.fileno)
 359     assert_nil @ep.flags_for(@rd)
 360   end
 361
 362   def test_flags_for_sym
 363     @ep.add @rd, :IN
 364     assert_equal Epoll::IN, @ep.flags_for(@rd.fileno)
 365     assert_equal Epoll::IN, @ep.flags_for(@rd)
 366
 367     @ep.del @rd
 368     assert_nil @ep.flags_for(@rd.fileno)
 369     assert_nil @ep.flags_for(@rd)
 370   end
 371
 372   def test_flags_for_sym_ary
 373     @ep.add @rd, [:IN, :ET]
 374     expect = Epoll::IN | Epoll::ET
 375     assert_equal expect, @ep.flags_for(@rd.fileno)
 376     assert_equal expect, @ep.flags_for(@rd)
 377
 378     @ep.del @rd
 379     assert_nil @ep.flags_for(@rd.fileno)
 380     assert_nil @ep.flags_for(@rd)
 381   end
 382
 383   def test_include?
 384     assert ! @ep.include?(@rd)
 385     @ep.add @rd, Epoll::IN
 386     assert @ep.include?(@rd)
 387     assert @ep.include?(@rd.fileno)
 388     assert ! @ep.include?(@wr)
 389     assert ! @ep.include?(@wr.fileno)
 390   end
 391
 392   def test_cross_thread_close
 393     tmp = []
 394     thr = Thread.new { sleep(1); @ep.close }
 395     assert_raises(IOError) do
 396       @ep.wait { |flags, obj| tmp << [ flags, obj ] }
 397     end
 398     assert_nil thr.value
 399   end if RUBY_VERSION == "1.9.3"
 400 end