% svn diff --diff-cmd diff -x '-u -p' Index: io.c =================================================================== --- io.c (revision 31777) +++ io.c (working copy) @@ -13,6 +13,7 @@ #include "ruby/ruby.h" #include "ruby/io.h" +#include "vm_core.h" #include "dln.h" #include #include @@ -8384,13 +8385,57 @@ struct copy_stream_struct { VALUE th; }; +static void * +exec_interrupts(void *arg) +{ + rb_thread_t *th = arg; + rb_threadptr_execute_interrupts(th); + return NULL; +} + +/* + * returns TRUE if the preceding system call was interrupted + * so we can continue. If the thread was interrupted, we + * reacquire the GVL to execute interrupts before continuing. + */ static int -maygvl_copy_stream_wait_read(struct copy_stream_struct *stp) +maygvl_copy_stream_continue_p(int has_gvl, struct copy_stream_struct *stp) +{ + switch (errno) { + case EINTR: +#if defined(ERESTART) + case ERESTART: +#endif + if (rb_thread_interrupted(stp->th)) + if (has_gvl) + rb_threadptr_execute_interrupts((rb_thread_t *)stp->th); + else + rb_thread_call_with_gvl(exec_interrupts, (void *)stp->th); + return TRUE; + } + return FALSE; +} + +static int +maygvl_select(int has_gvl, int n, rb_fdset_t *rfds, rb_fdset_t *wfds, rb_fdset_t *efds, struct timeval *timeout) +{ + if (has_gvl) + return rb_thread_fd_select(n, rfds, wfds, efds, timeout); + else + return rb_fd_select(n, rfds, wfds, efds, timeout); +} + +static int +maygvl_copy_stream_wait_read(int has_gvl, struct copy_stream_struct *stp) { int ret; - rb_fd_zero(&stp->fds); - rb_fd_set(stp->src_fd, &stp->fds); - ret = rb_fd_select(rb_fd_max(&stp->fds), &stp->fds, NULL, NULL, NULL); + + do { + rb_fd_zero(&stp->fds); + rb_fd_set(stp->src_fd, &stp->fds); + ret = maygvl_select(has_gvl, rb_fd_max(&stp->fds), &stp->fds, NULL, NULL, NULL); + } while (ret == -1 && maygvl_copy_stream_continue_p(has_gvl, stp)); + if (ret == -1) { stp->syserr = "select"; stp->error_no = errno; @@ -8403,9 +8448,13 @@ static int nogvl_copy_stream_wait_write(struct copy_stream_struct *stp) { int ret; - rb_fd_zero(&stp->fds); - rb_fd_set(stp->dst_fd, &stp->fds); - ret = rb_fd_select(rb_fd_max(&stp->fds), NULL, &stp->fds, NULL, NULL); + + do { + rb_fd_zero(&stp->fds); + rb_fd_set(stp->dst_fd, &stp->fds); + ret = rb_fd_select(rb_fd_max(&stp->fds), NULL, &stp->fds, NULL, NULL); + } while (ret == -1 && maygvl_copy_stream_continue_p(0, stp)); + if (ret == -1) { stp->syserr = "select"; stp->error_no = errno; @@ -8467,13 +8516,13 @@ simple_sendfile(int out_fd, int in_fd, o #ifdef USE_SENDFILE static int -maygvl_copy_stream_wait_readwrite(struct copy_stream_struct *stp) +maygvl_copy_stream_wait_readwrite(int has_gvl, struct copy_stream_struct *stp) { int ret; rb_fd_zero(&stp->fds); rb_fd_set(stp->src_fd, &stp->fds); rb_fd_set(stp->dst_fd, &stp->fds); - ret = rb_fd_select(rb_fd_max(&stp->fds), &stp->fds, NULL, NULL, NULL); + ret = maygvl_select(has_gvl, rb_fd_max(&stp->fds), &stp->fds, NULL, NULL, NULL); if (ret == -1) { stp->syserr = "select"; stp->error_no = errno; @@ -8552,6 +8601,8 @@ nogvl_copy_stream_sendfile(struct copy_s } } if (ss == -1) { + if (maygvl_copy_stream_continue_p(0, stp)) + goto retry_sendfile; switch (errno) { case EINVAL: #ifdef ENOSYS @@ -8562,9 +8613,7 @@ nogvl_copy_stream_sendfile(struct copy_s #if defined(EWOULDBLOCK) && EWOULDBLOCK != EAGAIN case EWOULDBLOCK: #endif - if (maygvl_copy_stream_wait_readwrite(stp) == -1) - return -1; - if (rb_thread_interrupted(stp->th)) + if (maygvl_copy_stream_wait_readwrite(0, stp) == -1) return -1; goto retry_sendfile; } @@ -8577,12 +8626,22 @@ nogvl_copy_stream_sendfile(struct copy_s #endif static ssize_t -maygvl_copy_stream_read(struct copy_stream_struct *stp, char *buf, size_t len, off_t offset) +maygvl_read(int has_gvl, int fd, void *buf, size_t count) +{ + if (has_gvl) + return rb_read_internal(fd, buf, count); + else + return read(fd, buf, count); +} + +static ssize_t +maygvl_copy_stream_read(int has_gvl, struct copy_stream_struct *stp, char *buf, size_t len, off_t offset) { ssize_t ss; retry_read: - if (offset == (off_t)-1) - ss = read(stp->src_fd, buf, len); + if (offset == (off_t)-1) { + ss = maygvl_read(has_gvl, stp->src_fd, buf, len); + } else { #ifdef HAVE_PREAD ss = pread(stp->src_fd, buf, len, offset); @@ -8595,12 +8654,14 @@ maygvl_copy_stream_read(struct copy_stre return 0; } if (ss == -1) { + if (maygvl_copy_stream_continue_p(has_gvl, stp)) + goto retry_read; switch (errno) { case EAGAIN: #if defined(EWOULDBLOCK) && EWOULDBLOCK != EAGAIN case EWOULDBLOCK: #endif - if (maygvl_copy_stream_wait_read(stp) == -1) + if (maygvl_copy_stream_wait_read(has_gvl, stp) == -1) return -1; goto retry_read; #ifdef ENOSYS @@ -8624,6 +8685,8 @@ nogvl_copy_stream_write(struct copy_stre while (len) { ss = write(stp->dst_fd, buf+off, len); if (ss == -1) { + if (maygvl_copy_stream_continue_p(0, stp)) + continue; if (errno == EAGAIN || errno == EWOULDBLOCK) { if (nogvl_copy_stream_wait_write(stp) == -1) return -1; @@ -8678,12 +8741,12 @@ nogvl_copy_stream_read_write(struct copy len = sizeof(buf); } if (use_pread) { - ss = maygvl_copy_stream_read(stp, buf, len, src_offset); + ss = maygvl_copy_stream_read(0, stp, buf, len, src_offset); if (0 < ss) src_offset += ss; } else { - ss = maygvl_copy_stream_read(stp, buf, len, (off_t)-1); + ss = maygvl_copy_stream_read(0, stp, buf, len, (off_t)-1); } if (ss <= 0) /* EOF or error */ return; @@ -8694,9 +8757,6 @@ nogvl_copy_stream_read_write(struct copy if (!use_eof) copy_length -= ss; - - if (rb_thread_interrupted(stp->th)) - return; } } @@ -8757,7 +8817,7 @@ copy_stream_fallback_body(VALUE arg) ssize_t ss; rb_thread_wait_fd(stp->src_fd); rb_str_resize(buf, buflen); - ss = maygvl_copy_stream_read(stp, RSTRING_PTR(buf), l, off); + ss = maygvl_copy_stream_read(1, stp, RSTRING_PTR(buf), l, off); if (ss == -1) return Qnil; if (ss == 0) Index: test/ruby/test_io.rb =================================================================== --- test/ruby/test_io.rb (revision 31777) +++ test/ruby/test_io.rb (working copy) @@ -78,6 +78,14 @@ class TestIO < Test::Unit::TestCase } end + def trapping_usr1 + @usr1_rcvd = 0 + trap(:USR1) { @usr1_rcvd += 1 } + yield + ensure + trap(:USR1, "DEFAULT") + end + def test_pipe r, w = IO.pipe assert_instance_of(IO, r) @@ -594,6 +602,30 @@ class TestIO < Test::Unit::TestCase result = t.value assert_equal(megacontent, result) } + with_socketpair {|s1, s2| + begin + s1.nonblock = true + rescue Errno::EBADF + skip "nonblocking IO for pipe is not implemented" + end + trapping_usr1 do + nr = 10 + pid = fork do + s1.close + IO.select([s2]) + Process.kill(:USR1, Process.ppid) + s2.read + end + s2.close + nr.times do + assert_equal megacontent.bytesize, IO.copy_stream("megasrc", s1) + end + assert_equal(1, @usr1_rcvd) + s1.close + _, status = Process.waitpid2(pid) + assert status.success?, status.inspect + end + } end } end