jruby · Nov 9, 2014 · Nov 9, 2014
Showing with 113 additions and 86 deletions.

+1 −1 .travis.yml

+4 −10 core/src/main/java/org/jruby/RubyIO.java

+8 −3 core/src/main/java/org/jruby/embed/jsr223/Utils.java

+100 −72 core/src/main/java/org/jruby/util/io/OpenFile.java
diff --git a/.travis.yml b/.travis.yml
@@ -67,7 +67,7 @@ branches:
 before_install:
   - if [[ $TRAVIS_JDK_VERSION = 'oraclejdk8' ]]; then sudo apt-get update; sudo apt-get install oracle-java8-installer; else true; fi
 
-script: travis_retry mvn install $TARGET -Dinvoker.skip=false | egrep -v 'Download|\\[exec\\] [[:digit:]]+/[[:digit:]]+|^[[:space:]]*\\[exec\\][[:space:]]*$' ; [ ${PIPESTATUS[0]} == 0 ]
+script: mvn package && travis_retry mvn install $TARGET -Dinvoker.skip=false | egrep -v 'Download|\\[exec\\] [[:digit:]]+/[[:digit:]]+|^[[:space:]]*\\[exec\\][[:space:]]*$' ; [ ${PIPESTATUS[0]} == 0 ]
 install: /bin/true
 notifications:
   irc:

diff --git a/core/src/main/java/org/jruby/RubyIO.java b/core/src/main/java/org/jruby/RubyIO.java
@@ -480,8 +480,8 @@ protected RubyIO reopenIO(ThreadContext context, RubyIO nfile) {
                 // TODO: clear interrupts waiting on this IO?
                 //            rb_thread_fd_close(fd);
                 if (orig.isReadable() && pos >= 0) {
-                    checkReopenSeek(context, runtime, fptr, pos);
-                    checkReopenSeek(context, runtime, orig, pos);
+                    fptr.checkReopenSeek(context, runtime, pos);
+                    orig.checkReopenSeek(context, runtime, pos);
                 }
             }
 
@@ -498,14 +498,8 @@ protected RubyIO reopenIO(ThreadContext context, RubyIO nfile) {
         return this;
     }
 
-    private void checkReopenSeek(ThreadContext context, Ruby runtime, OpenFile fptr, long pos) {
-        if (fptr.seek(context, pos, PosixShim.SEEK_SET) < 0 && fptr.errno() != null) {
-            throw runtime.newErrnoFromErrno(fptr.errno(), fptr.getPath());
-        }
-    }
-
     private void checkReopenCloexecDup2(Ruby runtime, OpenFile orig, ChannelFD oldfd, ChannelFD newfd) {
-        orig.cloexecDup2(runtime, oldfd, newfd);
+        OpenFile.cloexecDup2(new PosixShim(runtime.getPosix()), oldfd, newfd);
     }
 
     // rb_io_binmode
@@ -641,7 +635,7 @@ public IRubyObject reopen(ThreadContext context, IRubyObject[] args) {
             } else {
                 ChannelFD tmpfd = sysopen(runtime, fptr.getPath(), oflags_p[0], 0666);
                 Errno err = null;
-                if (fptr.cloexecDup2(runtime, tmpfd, fptr.fd()) < 0)
+                if (fptr.cloexecDup2(fptr.posix, tmpfd, fptr.fd()) < 0)
                     err = fptr.errno();
 
                 if (err != null) {

diff --git a/core/src/main/java/org/jruby/embed/jsr223/Utils.java b/core/src/main/java/org/jruby/embed/jsr223/Utils.java
@@ -177,9 +177,14 @@ private static void setErrorWriter(ScriptingContainer container, Writer writer)
     private static RubyIO getRubyIO(Ruby runtime, Writer writer) throws IOException, BadDescriptorException {
         PrintStream pstream = new PrintStream(new WriterOutputStream(writer), true);
         RubyIO io = new RubyIO(runtime, pstream, false);
-        io.getOpenFile().setSync(true);
-        io.getOpenFile().io_fflush(runtime.getCurrentContext());
-        return io;
+        boolean locked = io.getOpenFile().lock();
+        try {
+            io.getOpenFile().setSync(true);
+            io.getOpenFile().io_fflush(runtime.getCurrentContext());
+            return io;
+        } finally {
+            if (locked) io.getOpenFile().unlock();
+        }
     }
 
     static void postEval(ScriptingContainer container, ScriptContext context) {

diff --git a/core/src/main/java/org/jruby/util/io/OpenFile.java b/core/src/main/java/org/jruby/util/io/OpenFile.java
@@ -99,6 +99,12 @@ public void ascii8bitBinmode(Ruby runtime) {
         clearCodeConversion();
     }
 
+    public void checkReopenSeek(ThreadContext context, Ruby runtime, long pos) {
+        if (seek(context, pos, PosixShim.SEEK_SET) < 0 && errno() != null) {
+            throw runtime.newErrnoFromErrno(errno(), getPath());
+        }
+    }
+
     public static interface Finalizer {
         public void finalize(Ruby runtime, OpenFile fptr, boolean noraise);
     }
@@ -409,43 +415,49 @@ public void checkReadable(ThreadContext context) {
 
     // io_fflush
     public int io_fflush(ThreadContext context) {
-        assert lock.isHeldByCurrentThread();
-
-        checkClosed();
+        boolean locked = lock();
+        try {
+            checkClosed();
 
-        if (wbuf.len == 0) return 0;
+            if (wbuf.len == 0) return 0;
 
-        checkClosed();
+            checkClosed();
 
-        while (wbuf.len > 0 && flushBuffer() != 0) {
-            if (!waitWritable(context)) {
-                return -1;
+            while (wbuf.len > 0 && flushBuffer() != 0) {
+                if (!waitWritable(context)) {
+                    return -1;
+                }
+                checkClosed();
             }
-            checkClosed();
+        } finally {
+            if (locked) unlock();
         }
 
         return 0;
     }
 
     // rb_io_wait_writable
     public boolean waitWritable(ThreadContext context, long timeout) {
-        assert lock.isHeldByCurrentThread();
-
-        if (posix.errno == null) return false;
-
-        if (fd == null) throw runtime.newIOError(RubyIO.CLOSED_STREAM_MSG);
-
-        switch (posix.errno) {
-            case EINTR:
-//            case ERESTART: // not defined in jnr-constants
-                runtime.getCurrentContext().pollThreadEvents();
-                return true;
-            case EAGAIN:
-            case EWOULDBLOCK:
-                ready(runtime, context.getThread(), SelectExecutor.WRITE_CONNECT_OPS, timeout);
-                return true;
-            default:
-                return false;
+        boolean locked = lock();
+        try {
+            if (posix.errno == null) return false;
+
+            if (fd == null) throw runtime.newIOError(RubyIO.CLOSED_STREAM_MSG);
+
+            switch (posix.errno) {
+                case EINTR:
+                    //            case ERESTART: // not defined in jnr-constants
+                    runtime.getCurrentContext().pollThreadEvents();
+                    return true;
+                case EAGAIN:
+                case EWOULDBLOCK:
+                    ready(runtime, context.getThread(), SelectExecutor.WRITE_CONNECT_OPS, timeout);
+                    return true;
+                default:
+                    return false;
+            }
+        } finally {
+            if (locked) unlock();
         }
     }
 
@@ -456,23 +468,26 @@ public boolean waitWritable(ThreadContext context) {
 
     // rb_io_wait_readable
     public boolean waitReadable(ThreadContext context, long timeout) {
-        assert lock.isHeldByCurrentThread();
-
-        if (posix.errno == null) return false;
-
-        if (fd == null) throw runtime.newIOError(RubyIO.CLOSED_STREAM_MSG);
-
-        switch (posix.errno) {
-            case EINTR:
-//            case ERESTART: // not defined in jnr-constants
-                runtime.getCurrentContext().pollThreadEvents();
-                return true;
-            case EAGAIN:
-            case EWOULDBLOCK:
-                ready(runtime, context.getThread(), SelectionKey.OP_READ, timeout);
-                return true;
-            default:
-                return false;
+        boolean locked = lock();
+        try {
+            if (posix.errno == null) return false;
+
+            if (fd == null) throw runtime.newIOError(RubyIO.CLOSED_STREAM_MSG);
+
+            switch (posix.errno) {
+                case EINTR:
+                    //            case ERESTART: // not defined in jnr-constants
+                    runtime.getCurrentContext().pollThreadEvents();
+                    return true;
+                case EAGAIN:
+                case EWOULDBLOCK:
+                    ready(runtime, context.getThread(), SelectionKey.OP_READ, timeout);
+                    return true;
+                default:
+                    return false;
+            }
+        } finally {
+            if (locked) unlock();
         }
     }
 
@@ -481,10 +496,6 @@ public boolean waitReadable(ThreadContext context) {
         return waitReadable(context, 0);
     }
 
-    public boolean ready(ThreadContext context, int ops) {
-        return ready(runtime, context.getThread(), ops, 0);
-    }
-
     /**
      * Wait until the primary
      *
@@ -494,8 +505,7 @@ public boolean ready(ThreadContext context, int ops) {
      * @return
      */
     public boolean ready(Ruby runtime, RubyThread thread, int ops, long timeout) {
-        assert lock.isHeldByCurrentThread();
-
+        boolean locked = lock();
         try {
             if (fd.chSelect != null) {
                 int ready_stat = 0;
@@ -511,6 +521,13 @@ public boolean ready(Ruby runtime, RubyThread thread, int ops, long timeout) {
                         } else {
                             addedThread = true;
                             addBlockingThread(thread);
+                            // release lock while selecting
+                            unlock();
+                            try {
+                                sel.select(timeout);
+                            } finally {
+                                lock();
+                            }
                         }
                         sel.close();
                     } finally {
@@ -535,6 +552,8 @@ public boolean ready(Ruby runtime, RubyThread thread, int ops, long timeout) {
             }
         } catch (IOException ioe) {
             throw runtime.newIOErrorFromException(ioe);
+        } finally {
+            if (locked) unlock();
         }
 
     }
@@ -562,16 +581,15 @@ public int flushBufferAsync2() {
     }
 
     // io_flush_buffer_sync2
-    public int flushBufferSync2() {
+    private int flushBufferSync2() {
         int result = flushBufferSync();
 
         return result;
 //        return result == 0 ? 1 : result;
     }
 
     // io_flush_buffer_sync
-    public int flushBufferSync()
-    {
+    private int flushBufferSync() {
         int l = writableLength(wbuf.len);
         int r = posix.write(fd, wbuf.ptr, wbuf.off, l, nonblock);
 
@@ -589,8 +607,7 @@ public int flushBufferSync()
     }
 
     // io_writable_length
-    public int writableLength(int l)
-    {
+    private int writableLength(int l) {
         // We don't use wsplit mode, so we just pass length back directly.
 //        if (PIPE_BUF < l &&
 //                // we should always assume other threads, so we don't use rb_thread_alone
@@ -615,7 +632,7 @@ public int writableLength(int l)
      * @return
      */
     // wsplit_p
-    boolean wsplit()
+    private boolean wsplit()
     {
         int r;
 
@@ -636,29 +653,40 @@ boolean wsplit()
 
     // io_seek
     public long seek(ThreadContext context, long offset, int whence) {
-        assert lock.isHeldByCurrentThread();
-
-        flushBeforeSeek(context);
-        return posix.lseek(fd, offset, whence);
+        boolean locked = lock();
+        try {
+            flushBeforeSeek(context);
+            return posix.lseek(fd, offset, whence);
+        } finally {
+            if (locked) unlock();
+        }
     }
 
     // flush_before_seek
     private void flushBeforeSeek(ThreadContext context) {
-        assert lock.isHeldByCurrentThread();
-
-        if (io_fflush(context) < 0)
-            throw context.runtime.newErrnoFromErrno(posix.errno, "");
-        unread(context);
-        posix.errno = null;
+        boolean locked = lock();
+        try {
+            if (io_fflush(context) < 0)
+                throw context.runtime.newErrnoFromErrno(posix.errno, "");
+            unread(context);
+            posix.errno = null;
+        } finally {
+            if (locked) unlock();
+        }
     }
 
     public void checkWritable(ThreadContext context) {
-        checkClosed();
-        if ((mode & WRITABLE) == 0) {
-            throw context.runtime.newIOError("not opened for writing");
-        }
-        if (rbuf.len != 0) {
-            unread(context);
+        boolean locked = lock();
+        try {
+            checkClosed();
+            if ((mode & WRITABLE) == 0) {
+                throw context.runtime.newIOError("not opened for writing");
+            }
+            if (rbuf.len != 0) {
+                unread(context);
+            }
+        } finally {
+            if (locked) unlock();
         }
     }
 
@@ -2491,7 +2519,7 @@ public void errno(Errno newErrno) {
         posix.errno = newErrno;
     }
 
-    public int cloexecDup2(Ruby runtime, ChannelFD oldfd, ChannelFD newfd) {
+    public static int cloexecDup2(PosixShim posix, ChannelFD oldfd, ChannelFD newfd) {
         int ret;
         /* When oldfd == newfd, dup2 succeeds but dup3 fails with EINVAL.
          * rb_cloexec_dup2 succeeds as dup2.  */