Project

General

Profile

Download (9.42 KB)

Bug #4683 » copy_stream_interrupt_handling-2.patch

akr (Akira Tanaka), 06/02/2011 08:20 PM

View differences:

vm_core.h (working copy)
void rb_thread_start_timer_thread(void);
void rb_thread_stop_timer_thread(void);
void rb_thread_reset_timer_thread(void);
void *rb_thread_call_with_gvl(void *(*func)(void *), void *data1);
int ruby_thread_has_gvl_p(void);
VALUE rb_make_backtrace(void);
typedef int rb_backtrace_iter_func(void *, VALUE, int, VALUE);
io.c (working copy)
#include "ruby/ruby.h"
#include "ruby/io.h"
#include "dln.h"
#include "internal.h"
#include <ctype.h>
#include <errno.h>
......
VALUE th;
};
static void *
exec_interrupts(void *arg)
{
VALUE th = (VALUE)arg;
rb_thread_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_thread_execute_interrupts(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;
......
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;
......
#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;
......
}
}
if (ss == -1) {
if (maygvl_copy_stream_continue_p(0, stp))
goto retry_sendfile;
switch (errno) {
case EINVAL:
#ifdef ENOSYS
......
#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;
}
......
#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);
......
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
......
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;
......
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;
......
if (!use_eof)
copy_length -= ss;
if (rb_thread_interrupted(stp->th))
return;
}
}
......
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)
thread.c (working copy)
#include "eval_intern.h"
#include "gc.h"
#include "internal.h"
#include "ruby/io.h"
#ifndef USE_NATIVE_THREAD_PRIORITY
......
}
void
rb_thread_execute_interrupts(VALUE th)
{
rb_threadptr_execute_interrupts_rec((rb_thread_t *)th, 0);
}
void
rb_gc_mark_threads(void)
{
rb_bug("deprecated function rb_gc_mark_threads is called");
internal.h (working copy)
VALUE rb_big_uminus(VALUE x);
void rb_thread_execute_interrupts(VALUE th);
void *rb_thread_call_with_gvl(void *(*func)(void *), void *data1);
#if defined(__cplusplus)
#if 0
{ /* satisfy cc-mode */
test/ruby/test_io.rb (working copy)
}
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)
......
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
(4-4/4)