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_cross_thread
  18     tmp = []
  19     Thread.new { sleep 0.100; @ep.add(@wr, Epoll::OUT) }
  20     t0 = Time.now
  21     @ep.wait { |flags,obj| tmp << [ flags, obj ] }
  22     elapsed = Time.now - t0
  23     assert elapsed >= 0.100
  24     assert_equal [[Epoll::OUT, @wr]], tmp
  25   end
  26
  27   def test_fork_safe
  28     tmp = []
  29     @ep.add @rd, Epoll::IN
  30     pid = fork do
  31       @ep.wait(nil, 100) { |flags,obj| tmp << [ flags, obj ] }
  32       exit!(tmp.empty?)
  33     end
  34     @wr.syswrite "HI"
  35     _, status = Process.waitpid2(pid)
  36     assert status.success?
  37     @ep.wait(nil, 0) { |flags,obj| tmp << [ flags, obj ] }
  38     assert_equal [[Epoll::IN, @rd]], tmp
  39   end
  40
  41   def test_after_fork_usability
  42     fork { @ep.add(@rd, Epoll::IN); exit!(0) }
  43     fork { @ep.set(@rd, Epoll::IN); exit!(0) }
  44     fork { @ep.to_io; exit!(0) }
  45     fork { @ep.dup; exit!(0) }
  46     fork { @ep.clone; exit!(0) }
  47     fork { @ep.close; exit!(0) }
  48     fork { @ep.closed?; exit!(0) }
  49     fork {
  50       begin
  51         @ep.del(@rd)
  52       rescue Errno::ENOENT
  53         exit!(0)
  54       end
  55       exit!(1)
  56     }
  57     res = Process.waitall
  58     res.each { |(pid,status)| assert status.success? }
  59   end
  60
  61   def test_tcp_connect_nonblock_edge
  62     epflags = Epoll::OUT | Epoll::ET
  63     host = '127.0.0.1'
  64     srv = TCPServer.new(host, 0)
  65     port = srv.addr[1]
  66     addr = Socket.pack_sockaddr_in(port, host)
  67     sock = Socket.new(Socket::AF_INET, Socket::SOCK_STREAM, 0)
  68     assert_raises(Errno::EINPROGRESS) { sock.connect_nonblock(addr) }
  69     IO.select(nil, [ sock ], [sock ])
  70     @ep.add(sock, epflags)
  71     tmp = []
  72     @ep.wait(1) { |flags, obj| tmp << [ flags, obj ] }
  73     assert_equal [ [Epoll::OUT,  sock] ], tmp
  74   end
  75
  76   def test_tcp_connect_edge
  77     epflags = Epoll::OUT | Epoll::ET
  78     host = '127.0.0.1'
  79     srv = TCPServer.new(host, 0)
  80     port = srv.addr[1]
  81     sock = TCPSocket.new(host, port)
  82     @ep.add(sock, epflags)
  83     tmp = []
  84     @ep.wait(1) { |flags, obj| tmp << [ flags, obj ] }
  85     assert_equal [ [Epoll::OUT,  sock] ], tmp
  86   end
  87
  88   def teardown
  89     assert_nothing_raised do
  90       @rd.close unless @rd.closed?
  91       @wr.close unless @wr.closed?
  92       @ep.close unless @ep.closed?
  93     end
  94   end
  95
  96   def test_max_events_big
  97     @ep.add @rd, Epoll::IN
  98     tmp = []
  99     thr = Thread.new { @ep.wait(1024) { |flags, obj| tmp << [ flags, obj ] } }
 100     Thread.pass
 101     assert tmp.empty?
 102     @wr.write '.'
 103     thr.join
 104     assert_equal([[Epoll::IN, @rd]], tmp)
 105     tmp.clear
 106     thr = Thread.new { @ep.wait { |flags, obj| tmp << [ flags, obj ] } }
 107     thr.join
 108     assert_equal([[Epoll::IN, @rd]], tmp)
 109   end
 110
 111   def test_max_events_small
 112     @ep.add @rd, Epoll::IN | Epoll::ET
 113     @ep.add @wr, Epoll::OUT | Epoll::ET
 114     @wr.write '.'
 115     tmp = []
 116     @ep.wait(1) { |flags, obj| tmp << [ flags, obj ] }
 117     assert_equal 1, tmp.size
 118     @ep.wait(1) { |flags, obj| tmp << [ flags, obj ] }
 119     assert_equal 2, tmp.size
 120   end
 121
 122   def test_signal_safe
 123     time = {}
 124     trap(:USR1) { time[:USR1] = Time.now; sleep 0.1; @wr.write '.' }
 125     @ep.add @rd, Epoll::IN
 126     tmp = []
 127     pid = fork do
 128       sleep 0.1 # slightly racy :<
 129       Process.kill(:USR1, Process.ppid)
 130     end
 131     time[:START_WAIT] = Time.now
 132     @ep.wait { |flags, obj| tmp << [ flags, obj ]; time[:EP] = Time.now }
 133     assert_equal([[Epoll::IN, @rd]], tmp)
 134     _, status = Process.waitpid2(pid)
 135     assert status.success?
 136     assert((time[:USR1] - time[:START_WAIT]) >= 0.1)
 137     assert((time[:USR1] - time[:START_WAIT]) < 0.15)
 138     assert((time[:EP] - time[:USR1]) >= 0.1)
 139     assert((time[:EP] - time[:USR1]) < 0.15)
 140     ensure
 141       trap(:USR1, 'DEFAULT')
 142   end unless RBX
 143
 144   def test_close
 145     @ep.add @rd, Epoll::IN
 146     tmp = []
 147     thr = Thread.new { @ep.wait { |flags, obj| tmp << [ flags, obj ] } }
 148     @rd.close
 149     @wr.close
 150     assert_nil thr.join(0.01)
 151     assert thr.alive?
 152     thr.kill
 153     assert tmp.empty?
 154   end
 155
 156   def test_rdhup
 157     rd, wr = UNIXSocket.pair
 158     @ep.add rd, Epoll::RDHUP
 159     tmp = []
 160     thr = Thread.new { @ep.wait { |flags, obj| tmp << [ flags, obj ] } }
 161     wr.shutdown Socket::SHUT_WR
 162     thr.join
 163     assert_equal([[ Epoll::RDHUP, rd ]], tmp)
 164   end
 165
 166   def test_hup
 167     @ep.add @rd, Epoll::IN
 168     tmp = []
 169     thr = Thread.new { @ep.wait { |flags, obj| tmp << [ flags, obj ] } }
 170     @wr.close
 171     thr.join
 172     assert_equal([[ Epoll::HUP, @rd ]], tmp)
 173   end
 174
 175   def test_multiple
 176     r, w = IO.pipe
 177     assert_nothing_raised do
 178       @ep.add r, Epoll::IN
 179       @ep.add @rd, Epoll::IN
 180       @ep.add w, Epoll::OUT
 181       @ep.add @wr, Epoll::OUT
 182     end
 183     tmp = []
 184     @ep.wait { |flags, obj| tmp << [ flags, obj ] }
 185     assert_equal 2, tmp.size
 186     assert_equal [ Epoll::OUT ], tmp.map { |flags, obj| flags }.uniq
 187     ios = tmp.map { |flags, obj| obj }
 188     assert ios.include?(@wr)
 189     assert ios.include?(w)
 190   end
 191
 192   def test_gc
 193     assert_nothing_raised { 4096.times { Epoll.new } }
 194     assert ! @ep.closed?
 195   end unless RBX
 196
 197   def test_gc_to_io
 198     assert_nothing_raised do
 199       4096.times do
 200         ep = Epoll.new
 201         io = ep.to_io
 202       end
 203     end
 204     assert ! @ep.closed?
 205   end unless RBX
 206
 207   def test_clone
 208     tmp = []
 209     clone = @ep.clone
 210     assert @ep.to_io.fileno != clone.to_io.fileno
 211     clone.add @wr, Epoll::OUT
 212     @ep.wait(nil, 0) { |flags, obj| tmp << [ flags, obj ] }
 213     assert_equal([[Epoll::OUT, @wr]], tmp)
 214     assert_nothing_raised { clone.close }
 215   end
 216
 217   def test_dup
 218     tmp = []
 219     clone = @ep.dup
 220     assert @ep.to_io.fileno != clone.to_io.fileno
 221     clone.add @wr, Epoll::OUT
 222     @ep.wait(nil, 0) { |flags, obj| tmp << [ flags, obj ] }
 223     assert_equal([[Epoll::OUT, @wr]], tmp)
 224     assert_nothing_raised { clone.close }
 225   end
 226
 227   def test_set_idempotency
 228     assert_nothing_raised do
 229       @ep.set @rd, Epoll::IN
 230       @ep.set @rd, Epoll::IN
 231       @ep.set @wr, Epoll::OUT
 232       @ep.set @wr, Epoll::OUT
 233     end
 234   end
 235
 236   def test_wait_timeout
 237     t0 = Time.now
 238     assert_equal 0, @ep.wait(nil, 100) { |flags,obj| assert false }
 239     diff = Time.now - t0
 240     assert(diff >= 0.075, "#{diff} < 0.100s")
 241   end
 242
 243   def test_del
 244     assert_raises(Errno::ENOENT) { @ep.del(@rd) }
 245     assert_nothing_raised do
 246       @ep.add(@rd, Epoll::IN)
 247       @ep.del(@rd)
 248     end
 249   end
 250
 251   def test_wait_read
 252     @ep.add(@rd, Epoll::IN)
 253     assert_equal 0, @ep.wait(nil, 0) { |flags,obj| assert false }
 254     @wr.syswrite '.'
 255     i = 0
 256     nr = @ep.wait(nil, 0) do |flags,obj|
 257       assert_equal Epoll::IN, flags
 258       assert_equal obj, @rd
 259       i += 1
 260     end
 261     assert_equal 1, i
 262     assert_equal 1, nr
 263   end
 264
 265   def test_wait_write
 266     @ep.add(@wr, Epoll::OUT | Epoll::IN)
 267     i = 0
 268     nr = @ep.wait(nil, 0) do |flags, obj|
 269       assert_equal Epoll::OUT, flags
 270       assert_equal obj, @wr
 271       i += 1
 272     end
 273     assert_equal 1, nr
 274     assert_equal 1, i
 275   end
 276
 277   def test_wait_write_blocked
 278     begin
 279       @wr.write_nonblock('.' * 65536)
 280     rescue Errno::EAGAIN
 281       break
 282     end while true
 283     @ep.add(@wr, Epoll::OUT | Epoll::IN)
 284     i = 0
 285     assert_equal 0, @ep.wait(nil, 0) { |flags,event| assert false }
 286   end
 287
 288   def test_selectable
 289     tmp = nil
 290     @ep.add @rd, Epoll::IN
 291     thr = Thread.new { tmp = IO.select([ @ep ]) }
 292     thr.join 0.01
 293     assert_nil tmp
 294     @wr.write '.'
 295     thr.join
 296     assert_equal([[@ep],[],[]], tmp)
 297   end
 298
 299   def test_new_no_cloexec
 300     @ep.close
 301     io = Epoll.new(0).to_io
 302     assert((io.fcntl(Fcntl::F_GETFD) & Fcntl::FD_CLOEXEC) == 0)
 303   end
 304
 305   def test_new_cloexec
 306     @ep.close
 307     io = Epoll.new(Epoll::CLOEXEC).to_io
 308     assert((io.fcntl(Fcntl::F_GETFD) & Fcntl::FD_CLOEXEC) == Fcntl::FD_CLOEXEC)
 309   end
 310
 311   def test_new
 312     @ep.close
 313     io = Epoll.new.to_io
 314     assert((io.fcntl(Fcntl::F_GETFD) & Fcntl::FD_CLOEXEC) == Fcntl::FD_CLOEXEC)
 315   end
 316
 317   def test_delete
 318     assert_nil @ep.delete(@rd)
 319     assert_nil @ep.delete(@wr)
 320     assert_nothing_raised { @ep.add @rd, Epoll::IN }
 321     assert_equal 0, @ep.delete(@rd)
 322     assert_nil @ep.delete(@rd)
 323   end
 324 end