input stream not yet reaching the threads

Author: j143

src/main/java/org/apache/sysds/runtime/instructions/ooc/UnaryOOCInstruction.java

@@ -59,11 +59,14 @@ public void processInstruction( ExecutionContext ec ) {
                                 tmp.getValue().unaryOperations(uop, new MatrixBlock()));
                         qOut.enqueueTask(tmpOut);
                     }
-                    qOut.dequeueTask();
                 }
                 catch(Exception ex) {
                     throw new DMLRuntimeException(ex);
                 }
+                finally {
+                    qOut.closeInput();
+                }
+
             });
         }
         finally {

src/test/java/org/apache/sysds/test/functions/ooc/UnaryTest.java

@@ -77,7 +77,7 @@ public void testUnaryOperation(boolean rewrite)
 			programArgs = new String[] {"-explain", "-stats", "-ooc", 
 				"-args", input(INPUT_NAME), output(OUTPUT_NAME)};
 
-			int rows = 3500, cols = 1000;
+			int rows = 1000, cols = 1000;
 			MatrixBlock mb = MatrixBlock.randOperations(rows, cols, 1.0, -1, 1, "uniform", 7);
 			MatrixWriter writer = MatrixWriterFactory.createMatrixWriter(FileFormat.BINARY);
 			writer.writeMatrixToHDFS(mb, input(INPUT_NAME), rows, cols, 1000, rows*cols);
@@ -86,7 +86,7 @@ public void testUnaryOperation(boolean rewrite)
 
 			runTest(true, false, null, -1);
 
-//			HashMap<MatrixValue.CellIndex, Double> dmlfile = readDMLMatrixFromOutputDir(OUTPUT_NAME);
+			HashMap<MatrixValue.CellIndex, Double> dmlfile = readDMLMatrixFromOutputDir(OUTPUT_NAME);
 //			Double result = dmlfile.get(new MatrixValue.CellIndex(1, 1));
 //			double expected = 0.0;
 //			for(int i = 0; i < rows; i++) {

src/test/scripts/functions/ooc/Unary.dml

@@ -25,4 +25,4 @@ print(toString(X))
 Y = ceil(X);
 print(toString(Y))
 # Write the final matrix result
-write(Y, $2);
+write(Y, $2, format="binary");