From a3419bc919b49d2c4a6efe04b375e7700f43e4b8 Mon Sep 17 00:00:00 2001
From: nobu <nobu@b2dd03c8-39d4-4d8f-98ff-823fe69b080e>
Date: Tue, 20 Jul 2010 03:50:41 +0000
Subject: * io.c (io_flush_buffer): write and buffer operations should be  
 monolithic.  [ruby-core:31348]

git-svn-id: svn+ssh://ci.ruby-lang.org/ruby/trunk@28687 b2dd03c8-39d4-4d8f-98ff-823fe69b080e
---
 ChangeLog            |  5 ++++
 io.c                 | 69 ++++++++++++++++++++++++++++------------------------
 test/ruby/test_io.rb | 14 +++++++++++
 3 files changed, 56 insertions(+), 32 deletions(-)

diff --git a/ChangeLog b/ChangeLog
index e6244cbd46..ecc0958f16 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,8 @@
+Tue Jul 20 12:50:37 2010  Nobuyoshi Nakada  <nobu@ruby-lang.org>
+
+	* io.c (io_flush_buffer): write and buffer operations should be
+	  monolithic.  [ruby-core:31348]
+
 Tue Jul 20 12:27:56 2010  Nobuyoshi Nakada  <nobu@ruby-lang.org>
 
 	* lib/fileutils.rb (FileUtils::Entry_#copy): check file name
diff --git a/io.c b/io.c
index 4e239f8601..58a23c9a19 100644
--- a/io.c
+++ b/io.c
@@ -610,52 +610,57 @@ io_writable_length(rb_io_t *fptr, long l)
 }
 
 static VALUE
-io_flush_buffer(VALUE arg)
+io_flush_buffer_sync(void *arg)
 {
-    rb_io_t *fptr = (rb_io_t *)arg;
+    rb_io_t *fptr = arg;
     long l = io_writable_length(fptr, fptr->wbuf_len);
-    return rb_write_internal(fptr->fd, fptr->wbuf+fptr->wbuf_off, l);
+    ssize_t r = write(fptr->fd, fptr->wbuf+fptr->wbuf_off, (size_t)l);
+
+    if (fptr->wbuf_len <= r) {
+	fptr->wbuf_off = 0;
+	fptr->wbuf_len = 0;
+	return 0;
+    }
+    if (0 <= r) {
+	fptr->wbuf_off += (int)r;
+	fptr->wbuf_len -= (int)r;
+	errno = EAGAIN;
+    }
+    return (VALUE)-1;
 }
 
-static int
-io_fflush(rb_io_t *fptr)
+static VALUE
+io_flush_buffer_async(VALUE arg)
 {
-    long r;
+    return rb_thread_blocking_region(io_flush_buffer_sync, (void *)arg, RUBY_UBF_IO, 0);
+}
 
-    rb_io_check_closed(fptr);
-    if (fptr->wbuf_len == 0)
-        return 0;
-    if (!rb_thread_fd_writable(fptr->fd)) {
-        rb_io_check_closed(fptr);
-    }
-  retry:
-    if (fptr->wbuf_len == 0)
-        return 0;
+static inline int
+io_flush_buffer(rb_io_t *fptr)
+{
     if (fptr->write_lock) {
-	r = rb_mutex_synchronize(fptr->write_lock, io_flush_buffer, (VALUE)fptr);
+	return (int)rb_mutex_synchronize(fptr->write_lock, io_flush_buffer_async, (VALUE)fptr);
     }
     else {
-	long l = io_writable_length(fptr, fptr->wbuf_len);
-	r = rb_write_internal(fptr->fd, fptr->wbuf+fptr->wbuf_off, l);
+	return (int)io_flush_buffer_async((VALUE)fptr);
     }
-    /* xxx: Other threads may modify wbuf.
-     * A lock is required, definitely. */
+}
+
+static int
+io_fflush(rb_io_t *fptr)
+{
     rb_io_check_closed(fptr);
-    if (fptr->wbuf_len <= r) {
-        fptr->wbuf_off = 0;
-        fptr->wbuf_len = 0;
+    if (fptr->wbuf_len == 0)
         return 0;
+    if (!rb_thread_fd_writable(fptr->fd)) {
+        rb_io_check_closed(fptr);
     }
-    if (0 <= r) {
-        fptr->wbuf_off += (int)r;
-        fptr->wbuf_len -= (int)r;
-        errno = EAGAIN;
-    }
-    if (rb_io_wait_writable(fptr->fd)) {
+    while (fptr->wbuf_len > 0 && io_flush_buffer(fptr) != 0) {
+	if (!rb_io_wait_writable(fptr->fd))
+	    return -1;
         rb_io_check_closed(fptr);
-        goto retry;
     }
-    return -1;
+    return 0;
 }
 
 #ifdef HAVE_RB_FD_INIT
@@ -3512,9 +3517,9 @@ rb_io_fptr_finalize(rb_io_t *fptr)
 {
     if (!fptr) return 0;
     fptr->pathv = Qnil;
-    fptr->write_lock = 0;
     if (0 <= fptr->fd)
 	rb_io_fptr_cleanup(fptr, TRUE);
+    fptr->write_lock = 0;
     if (fptr->rbuf) {
         free(fptr->rbuf);
         fptr->rbuf = 0;
diff --git a/test/ruby/test_io.rb b/test/ruby/test_io.rb
index 41d3640105..bd4326060c 100644
--- a/test/ruby/test_io.rb
+++ b/test/ruby/test_io.rb
@@ -1609,4 +1609,18 @@ End
     t.close
     assert_raise(IOError) {t.binmode}
   end
+
+  def test_threaded_flush
+    bug3585 = '[ruby-core:31348]'
+    src = %q{\
+      t = Thread.new { sleep 3 }
+      Thread.new {sleep 1; t.kill; p 'hi!'}
+      t.join
+    }.gsub(/^\s+/, '')
+    10.times.map do
+      Thread.start do
+        assert_in_out_err([], src, [%q["hi!"]])
+      end
+    end.each {|th| th.join}
+  end
 end
-- 
cgit v1.2.3