Adding additional wait for normal output end of the pipe shutdown

dlyubimov · Nov 22, 2012 · a80fa13 · a80fa13
1 parent 96d78c8
commit a80fa13
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 6 deletions.
diff --git a/README.md b/README.md
@@ -8,9 +8,9 @@ For starters, you want to be running Linux-- we're not quite ready for OS X.
 
 To use R Crunch, you need to have the following R packages installed locally:
 * rJava
-* doxygen
+* roxygen2
 * bitops
-* RProtoBuf
+* RProtoBuf (optional)
 
 You will also need the protocol buffer compiler, `protoc`, version 2.4.1 installed on your path.
 

diff --git a/crunchR/src/main/Rpkg/R/crunchR.R b/crunchR/src/main/Rpkg/R/crunchR.R
@@ -4,6 +4,7 @@
 #' @name crunchR
 #' @exportPattern "^crunchR\\."
 #' @import rJava
+#' @import bitops
 #' 
 #' @include zzzClasses.R
 #' @include Pipeline.R

diff --git a/crunchR/src/main/java/org/crunchr/io/TwoWayRPipe.java b/crunchR/src/main/java/org/crunchr/io/TwoWayRPipe.java
@@ -294,16 +294,27 @@ public boolean checkOutputQueue(boolean wait) throws IOException {
 
                 int secondsWaited = 0;
                 int waitTime = 20;
+                boolean threadExited = false;
                 while (null == (emitBuff = wait ? outQueue.poll(waitTime, TimeUnit.SECONDS) : outQueue.poll())) {
                     if (!wait)
                         return true;
                     secondsWaited += waitTime;
                     if (secondsWaited >= outputQueueMaxWait)
                         throw new IOException("R side is not responding in the maximum amount of time allowed.");
                     if (!rThread.isAlive()) {
-
-                        throw lastErr != null ? new IOException("R side exited prematurely", lastErr)
-                            : new IOException("R side exited prematurely (no errors were captured)");
+                        /*
+                         * if we see thread having exited it is still possible
+                         * that a normal output closure message is pending. so
+                         * we make sure we check again for the graceful message
+                         * before stating it was unexpected R-side breakage.
+                         */
+                        if (threadExited) {
+                            throw lastErr != null ? new IOException("R side exited prematurely", lastErr)
+                                : new IOException("R side exited prematurely (no errors were captured)");
+                        } else {
+                            threadExited = true;
+                            /* and try to wait again for the normal shutdown. */
+                        }
                     }
                 }
 

diff --git a/examples/src/main/R/Example1.R b/examples/src/main/R/Example1.R
@@ -32,4 +32,5 @@ wordCountExample()
 #doFn$callProcess(c("A","B","C"))	
 #
 #line <- "this is a line"
-#class(strsplit(tolower(line),"[^[:alnum:]]")[[1]])
+#class(strsplit(tolower(line),"[^[:alnum:]]")[[1]])
+