test/test_io_splice.rb

   1 # -*- encoding: binary -*-
   2 require 'test/unit'
   3 require 'tempfile'
   4 require 'socket'
   5 require 'io/nonblock'
   6 require 'timeout'
   7 $-w = true
   8 require 'io/splice'
   9
  10 # unused_port provides an unused port on +addr+ usable for TCP that is
  11 # guaranteed to be unused across all unicorn builds on that system.  It
  12 # prevents race conditions by using a lock file other unicorn builds
  13 # will see.  This is required if you perform several builds in parallel
  14 # with a continuous integration system or run tests in parallel via
  15 # gmake.  This is NOT guaranteed to be race-free if you run other
  16 # processes that bind to random ports for testing (but the window
  17 # for a race condition is very small).
  18 def unused_port(addr = '127.0.0.1')
  19   retries = 100
  20   base = 5000
  21   port = sock = nil
  22   begin
  23     sock = TCPServer.new(addr, 0)
  24     port = sock.addr[1]
  25
  26     # since we'll end up closing the random port we just got, there's a race
  27     # condition could allow the random port we just chose to reselect itself
  28     # when running tests in parallel with gmake.  Create a lock file while
  29     # we have the port here to ensure that does not happen .
  30     lock_path = "#{Dir::tmpdir}/unicorn_test.#{addr}:#{port}.lock"
  31     File.open(lock_path, File::WRONLY|File::CREAT|File::EXCL, 0600).close
  32     at_exit { File.unlink(lock_path) rescue nil }
  33   rescue Errno::EEXIST
  34     sock.close rescue nil
  35     retry
  36   end
  37   sock.close rescue nil
  38   port
  39 end
  40
  41 class Test_IO_Splice < Test::Unit::TestCase
  42
  43   def test_splice
  44     str = 'abcde'
  45     size = 5
  46     rd, wr = IO.pipe
  47     tmp = Tempfile.new('ruby_io_splice')
  48
  49     assert_nothing_raised {
  50       tmp.syswrite(str)
  51       tmp.sysseek(0)
  52     }
  53
  54     nr = IO.splice(tmp.fileno, nil, wr.fileno, nil, size, 0)
  55     assert_equal size, nr
  56     assert_equal str, rd.sysread(size)
  57   end
  58
  59   def test_splice_io
  60     str = 'abcde'
  61     size = 5
  62     rd, wr = IO.pipe
  63     tmp = Tempfile.new('ruby_io_splice')
  64
  65     assert_nothing_raised {
  66       tmp.syswrite(str)
  67       tmp.sysseek(0)
  68     }
  69
  70     nr = IO.splice(tmp, nil, wr, nil, size, 0)
  71     assert_equal size, nr
  72     assert_equal str, rd.sysread(size)
  73   end
  74
  75   def test_splice_io_ish
  76     str = 'abcde'
  77     size = 5
  78     rd, wr = IO.pipe
  79     tmp = Tempfile.new('ruby_io_splice')
  80     io_ish = [ tmp ]
  81     def io_ish.to_io
  82       first.to_io
  83     end
  84
  85     assert_nothing_raised {
  86       tmp.syswrite(str)
  87       tmp.sysseek(0)
  88     }
  89
  90     nr = IO.splice(io_ish, nil, wr, nil, size, 0)
  91     assert_equal size, nr
  92     assert_equal str, rd.sysread(size)
  93   end
  94
  95   def test_splice_in_offset
  96     str = 'abcde'
  97     off = 3
  98     len = 2
  99     rd, wr = IO.pipe
 100     tmp = Tempfile.new('ruby_io_splice')
 101
 102     assert_nothing_raised {
 103       tmp.syswrite(str)
 104       tmp.sysseek(0)
 105     }
 106
 107     nr = IO.splice(tmp.fileno, off, wr.fileno, nil, len, 0)
 108     assert_equal len, nr
 109     assert_equal 'de', rd.sysread(len)
 110   end
 111
 112   def test_splice_out_offset
 113     str = 'abcde'
 114     rd, wr = IO.pipe
 115     tmp = Tempfile.new('ruby_io_splice')
 116
 117     assert_nothing_raised { wr.syswrite(str) }
 118     nr = IO.splice(rd.fileno, nil, tmp.fileno, 3, str.size, 0)
 119     assert_equal 5, nr
 120     assert_nothing_raised { tmp.sysseek(0) }
 121     assert_equal "\0\0\0abcde", tmp.sysread(9)
 122   end
 123
 124   def test_splice_nonblock
 125     rd, wr = IO.pipe
 126     tmp = Tempfile.new('ruby_io_splice')
 127
 128     assert_raises(Errno::EAGAIN) {
 129       IO.splice(rd.fileno, nil, tmp.fileno, 0, 5, IO::Splice::F_NONBLOCK)
 130     }
 131   end
 132
 133   def test_splice_eof
 134     rd, wr = IO.pipe
 135     tmp = Tempfile.new('ruby_io_splice')
 136     wr.syswrite 'abc'
 137     wr.close
 138
 139     nr = IO.splice(rd.fileno, nil, tmp.fileno, 0, 5, IO::Splice::F_NONBLOCK)
 140     assert_equal 3, nr
 141     assert_raises(EOFError) {
 142       IO.splice(rd.fileno, nil, tmp.fileno, 0, 5, IO::Splice::F_NONBLOCK)
 143     }
 144   end
 145
 146   def test_splice_nonblock_socket
 147     port = unused_port
 148     server = TCPServer.new('127.0.0.1', port)
 149     rp, wp = IO.pipe
 150     rs = TCPSocket.new('127.0.0.1', port)
 151     rs.nonblock = true
 152     assert_raises(Errno::EAGAIN) { IO.splice(rs, nil, wp, nil, 1024, 0) }
 153     rs.close
 154     server.close
 155   end
 156
 157   def test_tee
 158     str = 'abcde'
 159     size = 5
 160     rda, wra = IO.pipe
 161     rdb, wrb = IO.pipe
 162
 163     assert_nothing_raised { wra.syswrite(str) }
 164     nr = IO.tee(rda.fileno, wrb.fileno, size, 0)
 165     assert_equal 5, nr
 166     assert_equal str, rdb.sysread(5)
 167     assert_equal str, rda.sysread(5)
 168   end
 169
 170   def test_tee_eof
 171     rda, wra = IO.pipe
 172     rdb, wrb = IO.pipe
 173     wra.close
 174     assert_raises(EOFError) { IO.tee(rda.fileno, wrb.fileno, 4096, 0) }
 175   end
 176
 177   def test_tee_nonblock
 178     rda, wra = IO.pipe
 179     rdb, wrb = IO.pipe
 180     assert_raises(Errno::EAGAIN) {
 181       IO.tee(rda.fileno, wrb.fileno, 4096, IO::Splice::F_NONBLOCK)
 182     }
 183   end
 184
 185   def test_tee_io
 186     str = 'abcde'
 187     size = 5
 188     rda, wra = IO.pipe
 189     rdb, wrb = IO.pipe
 190
 191     assert_nothing_raised { wra.syswrite(str) }
 192     nr = IO.tee(rda, wrb, size, 0)
 193     assert_equal 5, nr
 194     assert_equal str, rdb.sysread(5)
 195     assert_equal str, rda.sysread(5)
 196   end
 197
 198   def test_vmsplice_array
 199     data = %w(hello world how are you today)
 200     r, w = IO.pipe
 201     n = IO.vmsplice(w.fileno, data, 0)
 202     assert_equal data.join('').size, n
 203     assert_equal data.join(''), r.readpartial(16384)
 204   end
 205
 206   def test_vmsplice_string
 207     r, w = IO.pipe
 208     assert_equal 5, IO.vmsplice(w, 'hello', 0)
 209     assert_equal 'hello', r.read(5)
 210   end
 211
 212   def test_vmsplice_array_io
 213     data = %w(hello world how are you today)
 214     r, w = IO.pipe
 215     n = IO.vmsplice(w, data, 0)
 216     assert_equal data.join('').size, n
 217     assert_equal data.join(''), r.readpartial(16384)
 218   end
 219
 220   def test_vmsplice_nonblock
 221     data = %w(hello world how are you today)
 222     r, w = IO.pipe
 223     w.syswrite('.' * IO::Splice::PIPE_CAPA)
 224     assert_raises(Errno::EAGAIN) {
 225       IO.vmsplice(w.fileno, data, IO::Splice::F_NONBLOCK)
 226     }
 227   end
 228
 229   def test_vmsplice_in_full
 230     empty = ""
 231
 232     # bs * count should be > PIPE_BUF
 233     [ [ 512, 512 ], [ 131073, 3 ], [ 4098, 64 ] ].each do |(bs,count)|
 234       rd, wr = IO.pipe
 235       buf = File.open('/dev/urandom', 'rb') { |fp| fp.sysread(bs) }
 236
 237       vec = (1..count).map { buf }
 238       pid = fork do
 239         wr.close
 240         tmp = []
 241         begin
 242           sleep 0.005
 243           tmp << rd.readpartial(8192)
 244         rescue EOFError
 245           break
 246         end while true
 247         ok = (vec.join(empty) == tmp.join(empty))
 248         exit! ok
 249       end
 250       assert_nothing_raised { rd.close }
 251       assert_equal(bs * count, IO.vmsplice(wr.fileno, vec, 0))
 252       assert_nothing_raised { wr.close }
 253       _, status = Process.waitpid2(pid)
 254       assert status.success?
 255     end
 256   end
 257
 258   def test_vmsplice_nil
 259     data = %w(hello world how are you today)
 260     assert_raises(TypeError) { IO.vmsplice(nil, data, 0) }
 261   end
 262
 263   def test_constants
 264     assert IO::Splice::PIPE_BUF > 0
 265     %w(move nonblock more gift).each { |x|
 266       assert Integer === IO::Splice.const_get("F_#{x.upcase}")
 267     }
 268     assert IO::Splice::PIPE_CAPA >= IO::Splice::PIPE_BUF
 269   end
 270
 271   def test_splice_copy_stream_file_to_file_small
 272     a, b = Tempfile.new('a'), Tempfile.new('b')
 273     a.syswrite 'hello world'
 274     a.sysseek(0)
 275     IO::Splice.copy_stream(a, b)
 276     b.rewind
 277     assert_equal 'hello world', b.read
 278   end
 279
 280   def test_splice_copy_stream_file_to_file_big
 281     buf = ('ab' * IO::Splice::PIPE_CAPA) + 'hi'
 282     a, b = Tempfile.new('a'), Tempfile.new('b')
 283     a.syswrite buf
 284     a.sysseek(0)
 285     IO::Splice.copy_stream(a, b)
 286     b.rewind
 287     assert_equal buf, b.read
 288   end
 289
 290   def test_splice_copy_stream_file_to_file_big_partial
 291     nr = IO::Splice::PIPE_CAPA
 292     buf = ('ab' * nr) + 'hi'
 293     a, b = Tempfile.new('a'), Tempfile.new('b')
 294     a.syswrite buf
 295     a.sysseek(0)
 296     assert_equal nr, IO::Splice.copy_stream(a, b, nr)
 297     b.rewind
 298     assert_equal('ab' * (nr/2), b.read)
 299   end
 300
 301   def test_splice_copy_stream_file_to_file_len
 302     a, b = Tempfile.new('a'), Tempfile.new('b')
 303     a.syswrite 'hello world'
 304     a.sysseek(0)
 305     IO::Splice.copy_stream(a, b, 5)
 306     b.rewind
 307     assert_equal 'hello', b.read
 308   end
 309
 310   def test_splice_copy_stream_pipe_to_file_len
 311     a = Tempfile.new('a')
 312     r, w = IO.pipe
 313     w.syswrite 'hello world'
 314     IO::Splice.copy_stream(r, a, 5)
 315     a.rewind
 316     assert_equal 'hello', a.read
 317   end
 318
 319   def test_splice_copy_stream_paths
 320     a = Tempfile.new('a')
 321     b = Tempfile.new('a')
 322     a.syswrite('hello world')
 323     IO::Splice.copy_stream(a.path, b.path, 5)
 324     assert_equal 'hello', b.read
 325   end
 326
 327   def test_splice_copy_stream_src_offset
 328     a = Tempfile.new('a')
 329     b = Tempfile.new('a')
 330     a.syswrite('hello world')
 331     IO::Splice.copy_stream(a.path, b.path, 5, 6)
 332     assert_equal 'world', b.read
 333   end
 334
 335   def test_copy_stream_nonblock_src
 336     port = unused_port
 337     server = TCPServer.new('127.0.0.1', port)
 338     rp, wp = IO.pipe
 339     rs = TCPSocket.new('127.0.0.1', port)
 340     rs.nonblock = true
 341     nr = 0
 342     assert_raises(Timeout::Error) do
 343       timeout(0.05) { nr += IO::Splice.copy_stream(rs, wp, 5) }
 344     end
 345     assert_equal 0, nr
 346     rs.close
 347     server.close
 348   end
 349
 350   def test_copy_stream_nonblock_dst
 351     port = unused_port
 352     server = TCPServer.new('127.0.0.1', port)
 353     rp, wp = IO.pipe
 354     rs = TCPSocket.new('127.0.0.1', port)
 355     rs.nonblock = true
 356     client = server.accept
 357     buf = ' ' * IO::Splice::PIPE_CAPA
 358     nr = 0
 359     assert_raises(Timeout::Error) do
 360       loop do
 361         begin
 362           wp.write_nonblock(buf)
 363         rescue Errno::EAGAIN
 364         end
 365         timeout(0.05) do
 366           nr += IO::Splice.copy_stream(rp, rs, IO::Splice::PIPE_CAPA)
 367         end
 368       end
 369     end
 370     assert_equal nr, client.read(nr).size
 371     rs.close
 372     server.close
 373   end
 374
 375   def test_copy_stream_eof
 376     r, w = IO.pipe
 377     w.syswrite 'hello world'
 378     w.close
 379     a = Tempfile.new('a')
 380     assert_equal 11, IO::Splice.copy_stream(r, a)
 381     a.rewind
 382     assert_equal 'hello world', a.read
 383   end
 384
 385   def test_pipe_size
 386     r, w = IO.pipe
 387     assert Integer, r.pipe_size
 388     assert(r.pipe_size >= 512)
 389     assert_nothing_raised { w.pipe_size = 8192 }
 390     assert 8192, r.pipe_size
 391
 392     w.write('*' * 4097)
 393     assert_raises(Errno::EBUSY) { r.pipe_size = 4096 }
 394
 395     pipe_max_size = File.read("/proc/sys/fs/pipe-max-size").to_i
 396     assert_nothing_raised { r.pipe_size = pipe_max_size }
 397     assert_raises(Errno::EPERM) { r.pipe_size = pipe_max_size * 2 }
 398   end if IO.method_defined?(:pipe_size)
 399
 400 end