* 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

@@ -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

io.c

@@ -610,37 +610,12 @@ 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);
-}
 
-static int
-io_fflush(rb_io_t *fptr)
-{
-    long r;
-
-    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;
-    if (fptr->write_lock) {
-	r = rb_mutex_synchronize(fptr->write_lock, io_flush_buffer, (VALUE)fptr);
-    }
-    else {
-	long l = io_writable_length(fptr, fptr->wbuf_len);
-	r = rb_write_internal(fptr->fd, fptr->wbuf+fptr->wbuf_off, l);
-    }
-    /* xxx: Other threads may modify wbuf.
-     * A lock is required, definitely. */
-    rb_io_check_closed(fptr);
     if (fptr->wbuf_len <= r) {
 	fptr->wbuf_off = 0;
 	fptr->wbuf_len = 0;
@@ -651,11 +626,41 @@ io_fflush(rb_io_t *fptr)
 	fptr->wbuf_len -= (int)r;
 	errno = EAGAIN;
     }
-    if (rb_io_wait_writable(fptr->fd)) {
+    return (VALUE)-1;
-        rb_io_check_closed(fptr);
-        goto retry;
 }
+
+static VALUE
+io_flush_buffer_async(VALUE arg)
+{
+    return rb_thread_blocking_region(io_flush_buffer_sync, (void *)arg, RUBY_UBF_IO, 0);
+}
+
+static inline int
+io_flush_buffer(rb_io_t *fptr)
+{
+    if (fptr->write_lock) {
+	return (int)rb_mutex_synchronize(fptr->write_lock, io_flush_buffer_async, (VALUE)fptr);
+    }
+    else {
+	return (int)io_flush_buffer_async((VALUE)fptr);
+    }
+}
+
+static int
+io_fflush(rb_io_t *fptr)
+{
+    rb_io_check_closed(fptr);
+    if (fptr->wbuf_len == 0)
+        return 0;
+    if (!rb_thread_fd_writable(fptr->fd)) {
+        rb_io_check_closed(fptr);
+    }
+    while (fptr->wbuf_len > 0 && io_flush_buffer(fptr) != 0) {
+	if (!rb_io_wait_writable(fptr->fd))
 	    return -1;
+        rb_io_check_closed(fptr);
+    }
+    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;

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