[Truffle] Poll during select.

jruby · Oct 14, 2015 · 14a7a31 · 14a7a31 · chrisseaton · Oct 14, 2015
1 parent 2c5017c
commit 14a7a31
Showing 3 changed files with 80 additions and 17 deletions.
diff --git a/truffle/src/main/java/org/jruby/truffle/nodes/rubinius/IOPrimitiveNodes.java b/truffle/src/main/java/org/jruby/truffle/nodes/rubinius/IOPrimitiveNodes.java
@@ -45,7 +45,9 @@
 import com.oracle.truffle.api.source.SourceSection;
 import jnr.constants.platform.Errno;
 import jnr.constants.platform.Fcntl;
-import jnr.ffi.Pointer;
+import jnr.ffi.*;
+import jnr.posix.DefaultNativeTimeval;
+import jnr.posix.Timeval;
 import org.jruby.truffle.nodes.RubyGuards;
 import org.jruby.truffle.nodes.dispatch.CallDispatchHeadNode;
 import org.jruby.truffle.nodes.dispatch.DispatchHeadNodeFactory;
@@ -100,7 +102,7 @@ public IOConnectPipeNode(RubyContext context, SourceSection sourceSection) {
         }
 
         @Specialization
-        public boolean connectPipe(VirtualFrame frame, DynamicObject lhs, DynamicObject rhs) {
+        public boolean connectPipe(DynamicObject lhs, DynamicObject rhs) {
             final int[] fds = new int[2];
 
             if (posix().pipe(fds) == -1) {
@@ -269,13 +271,11 @@ public Object readIfAvailable(VirtualFrame frame, DynamicObject file, int number
             final FDSet fdSet = fdSetFactory.create();
             fdSet.set(fd);
 
-            // TODO CS 2-Sep-15 why are longs 8 bytes? Is that always the case?
-            final Pointer timeout = getContext().getMemoryManager().allocateDirect(8 * 2); // Needs to be two longs.
-            timeout.putLong(0, 0);
-            timeout.putLong(8, 0);
+            final Timeval timeoutObject = new DefaultNativeTimeval(jnr.ffi.Runtime.getSystemRuntime());
+            timeoutObject.setTime(new long[]{0, 0});
 
             final int res = nativeSockets().select(fd + 1, fdSet.getPointer(),
-                    PointerNodes.NULL_POINTER, PointerNodes.NULL_POINTER, timeout);
+                    PointerNodes.NULL_POINTER, PointerNodes.NULL_POINTER, timeoutObject);
 
             if (res == 0) {
                 CompilerDirectives.transferToInterpreter();
@@ -594,19 +594,25 @@ public Object select(DynamicObject readables, DynamicObject writables, DynamicOb
                 readableSet.set(fd);
             }
 
-            final int result = getContext().getThreadManager().runUntilResult(this, new ThreadManager.BlockingAction<Integer>() {
+            final int result = getContext().getThreadManager().runUntilTimeout(this, timeout, new ThreadManager.BlockingTimeoutAction<Integer>() {
                 @Override
-                public Integer block() throws InterruptedException {
-                    return nativeSockets().select(
+                public Integer block(Timeval timeoutToUse) throws InterruptedException {
+                    final int result = nativeSockets().select(
                             max(readableFds) + 1,
                             readableSet.getPointer(),
                             PointerNodes.NULL_POINTER,
                             PointerNodes.NULL_POINTER,
-                            PointerNodes.NULL_POINTER);
+                            timeoutToUse);
+
+                    if (result == 0) {
+                        return null;
+                    }
+
+                    return result;
                 }
             });
 
-            if (result == -1) {
+            if (result == -1 || result == 0) {
                 return nil();
             }
 
@@ -637,7 +643,7 @@ public Integer block() throws InterruptedException {
                             PointerNodes.NULL_POINTER,
                             writableSet.getPointer(),
                             PointerNodes.NULL_POINTER,
-                            PointerNodes.NULL_POINTER);
+                            null);
                 }
             });
 

diff --git a/truffle/src/main/java/org/jruby/truffle/runtime/sockets/NativeSockets.java b/truffle/src/main/java/org/jruby/truffle/runtime/sockets/NativeSockets.java
@@ -10,6 +10,7 @@
 package org.jruby.truffle.runtime.sockets;
 
 import jnr.ffi.Pointer;
+import jnr.posix.Timeval;
 
 public interface NativeSockets {
 
@@ -86,7 +87,7 @@ public interface NativeSockets {
      *        fd_set *restrict errorfds, struct timeval *restrict timeout);
      */
 
-    int select(int nfds, Pointer readfds, Pointer writefds, Pointer errorfds, Pointer timeout);
+    int select(int nfds, Pointer readfds, Pointer writefds, Pointer errorfds, Timeval timeout);
 
     /*
      * int

diff --git a/truffle/src/main/java/org/jruby/truffle/runtime/subsystems/ThreadManager.java b/truffle/src/main/java/org/jruby/truffle/runtime/subsystems/ThreadManager.java
@@ -12,6 +12,8 @@
 import com.oracle.truffle.api.CompilerDirectives.TruffleBoundary;
 import com.oracle.truffle.api.nodes.Node;
 import com.oracle.truffle.api.object.DynamicObject;
+import jnr.posix.DefaultNativeTimeval;
+import jnr.posix.Timeval;
 import org.jruby.RubyThread.Status;
 import org.jruby.truffle.nodes.RubyGuards;
 import org.jruby.truffle.nodes.core.FiberNodes;
@@ -52,13 +54,16 @@ public DynamicObject getRootThread() {
         return rootThread;
     }
 
-
-    public static interface BlockingAction<T> {
-        public static boolean SUCCESS = true;
+    public interface BlockingAction<T> {
+        boolean SUCCESS = true;
 
         T block() throws InterruptedException;
     }
 
+    public interface BlockingTimeoutAction<T> {
+        T block(Timeval timeoutToUse) throws InterruptedException;
+    }
+
     /**
      * Runs {@code action} until it returns a non-null value.
      * The action might be {@link Thread#interrupted()}, for instance by
@@ -90,6 +95,57 @@ public <T> T runUntilResult(Node currentNode, BlockingAction<T> action) {
         return result;
     }
 
+    public <T> T runUntilTimeout(Node currentNode, int timeout, final BlockingTimeoutAction<T> action) {
+        final Timeval timeoutToUse = new DefaultNativeTimeval(jnr.ffi.Runtime.getSystemRuntime());
+
+        if (timeout == 0) {
+            timeoutToUse.setTime(new long[]{0, 0});
+
+            return runUntilResult(currentNode, new BlockingAction<T>() {
+
+                @Override
+                public T block() throws InterruptedException {
+                    return action.block(timeoutToUse);
+                }
+
+            });
+        } else {
+            final int pollTime = 500_000_000;
+            final long requestedTimeoutAt = System.nanoTime() + timeout * 1_000;
+
+            return runUntilResult(currentNode, new BlockingAction<T>() {
+
+                @Override
+                public T block() throws InterruptedException {
+                    final long timeUntilRequestedTimeout = requestedTimeoutAt - System.nanoTime();
+
+                    final boolean timeoutForPoll;
+                    final long effectiveTimeout;
+
+                    if (timeUntilRequestedTimeout < pollTime) {
+                        timeoutForPoll = false;
+                        effectiveTimeout = timeUntilRequestedTimeout;
+                    } else {
+                        timeoutForPoll = true;
+                        effectiveTimeout = pollTime;
+                    }
+
+                    final long effectiveTimeoutMicros = effectiveTimeout / 1_000;
+                    timeoutToUse.setTime(new long[]{effectiveTimeoutMicros / 1_000_000, effectiveTimeoutMicros % 1_000_000});
+
+                    final T result = action.block(timeoutToUse);
+
+                    if (result == null && timeoutForPoll && requestedTimeoutAt - System.nanoTime() > 0) {
+                        throw new InterruptedException();
+                    }
+
+                    return result;
+                }
+
+            });
+        }
+    }
+
     public void initializeCurrentThread(DynamicObject thread) {
         assert RubyGuards.isRubyThread(thread);
         currentThread.set(thread);