code coverage & multiprocessing bug fix in unittest-parallel

Author: e-strauss

.github/workflows/javaTests.yml

@@ -29,6 +29,7 @@ on:
       - '*.html'
       - 'src/main/python/**'
       - 'dev/**'
+      - '.github/workflows/python.yml'
     branches:
       - main
   pull_request:
@@ -38,6 +39,7 @@ on:
       - '*.html'
       - 'src/main/python/**'
       - 'dev/**'
+      - '.github/workflows/python.yml'
     branches:
       - main
 

.github/workflows/python.yml

@@ -132,15 +132,27 @@ jobs:
         export SYSDS_QUIET=1
         export LOG4JPROP=$SYSTEMDS_ROOT/src/test/resources/log4j.properties
         cd src/main/python
+        
+        # parallel tests
         unittest-parallel -t . -s tests -v
+        
+        # non parallel tests
+        python -m unittest discover -s tests/non_parallel_tests
+        
         # python -m unittest discover -s tests -p 'test_*.py'
         echo "Exit Status: " $?
 
     - name: Run all python tests no environment
       run: |
         export LOG4JPROP=$(pwd)/src/test/resources/log4j.properties
         cd src/main/python
+        
+        # parallel tests
         unittest-parallel -t . -s tests -v
+        
+        # non parallel tests
+        python -m unittest discover -s tests/non_parallel_tests
+        
         # python -m unittest discover -s tests -p 'test_*.py'
         echo "Exit Status: " $?
 

src/main/java/org/apache/sysds/api/PythonDMLScript.java

@@ -126,14 +126,16 @@ public MatrixBlock startReadingMbFromPipe(int id, int rlen, int clen, Types.Valu
 			double[] denseBlock = new double[(int) limit];
 			UnixPipeUtils.readNumpyArrayInBatches(pipe, id, BATCH_SIZE, (int) limit, type, denseBlock, 0);
 			mb.init(denseBlock, rlen, clen);
+		} else {
+			throw new DMLRuntimeException("FIFO Pipes are not initialized.");
 		}
 		mb.recomputeNonZeros();
 		mb.examSparsity();
 		LOG.debug("Reading from Python finished");
 		return mb;
 	}
 
-	public MatrixBlock startReadingMbFromPipes(int[] blockSizes, int rlen, int clen, Types.ValueType type) throws IOException {
+	public MatrixBlock startReadingMbFromPipes(int[] blockSizes, int rlen, int clen, Types.ValueType type) throws ExecutionException, InterruptedException {
 		long limit = (long) rlen * clen;
 		if(limit > Integer.MAX_VALUE)
 			throw new DMLRuntimeException("Dense NumPy array of size " + limit +
@@ -157,21 +159,12 @@ public MatrixBlock startReadingMbFromPipes(int[] blockSizes, int rlen, int clen,
 			}
 			// Wait for all tasks and propagate exceptions
 			for (Future<Void> f : futures) {
-				try {
-					f.get();
-				} catch (ExecutionException e) {
-					Throwable cause = e.getCause();
-					if (cause instanceof IOException)
-						throw (IOException) cause;
-					else
-						throw new RuntimeException("Pipe reader thread failed", cause);
-				} catch (InterruptedException e) {
-					Thread.currentThread().interrupt();
-					throw new RuntimeException("Pipe reader interrupted", e);
-				}
+				f.get();
 			}
 
 			mb.init(denseBlock, rlen, clen);
+		} else {
+			throw new DMLRuntimeException("FIFO Pipes are not initialized.");
 		}
 		mb.recomputeNonZeros();
 		mb.examSparsity();
@@ -186,6 +179,8 @@ public void startWritingMbToPipe(int id, MatrixBlock mb) throws IOException {
 
 			BufferedOutputStream out = toPython.get(id);
 			UnixPipeUtils.writeNumpyArrayInBatches(out, id, BATCH_SIZE, numElem, Types.ValueType.FP64, mb);
+		} else {
+			throw new DMLRuntimeException("FIFO Pipes are not initialized.");
 		}
 	}
 

src/main/java/org/apache/sysds/runtime/util/UnixPipeUtils.java

@@ -204,9 +204,9 @@ public static void writeNumpyArrayInBatches(BufferedOutputStream out, int id, in
 
 			// Fill buffer from MatrixBlock into byte[] (typed)
 			int bytesWritten = fillByteArrayFromDoubleArray(type, mb, offset, buffer, currentBatchSize);
-			if (bytesWritten != currentBatchSize) {
-				throw new IOException("Internal error: mismatched buffer fill size");
-			}
+//			if (bytesWritten != currentBatchSize) {
+//				throw new IOException("Internal error: mismatched buffer fill size");
+//			}
 
 			out.write(buffer, 0, currentBatchSize);
 			offset += currentBatchSize / elemSize;

src/main/python/systemds/context/systemds_context.py

@@ -87,7 +87,7 @@ def __init__(
         capture_stdout: bool = False,
         logging_level: int = 20,
         py4j_logging_level: int = 50,
-        data_transfer_mode: int = 1,
+        data_transfer_mode: int = 0,
         multi_pipe_enabled: bool = False,
     ):
         """Starts a new instance of SystemDSContext, in which the connection to a JVM systemds instance is handled
@@ -120,7 +120,9 @@ def __setup_data_transfer(self, data_transfer_mode=0, multi_pipe_enabled=False):
             self.__make_fifo_named_pipes(num_pipes)
             executor_pool, in_pipes, out_pipes = self.__init_pipes(num_pipes)
 
-            self._log.debug("Handshake done for {} IN / OUT Pipes".format(num_pipes))
+            self._log.info(
+                "Data transfer: Handshake done for {} IN / OUT Pipes".format(num_pipes)
+            )
             self.executor_pool = executor_pool
             self.FIFO_PY2JAVA_PIPES = out_pipes
             self.FIFO_JAVA2PY_PIPES = in_pipes

…matrix/test_block_converter_unix_pipe.py …_tests/test_block_converter_unix_pipe.pysrc/main/python/tests/matrix/test_block_converter_unix_pipe.py renamed to src/main/python/tests/non_parallel_tests/test_block_converter_unix_pipe.py src/main/python/tests/matrix/test_block_converter_unix_pipe.py renamed to src/main/python/tests/non_parallel_tests/test_block_converter_unix_pipe.py

@@ -101,6 +101,9 @@ public static class NonParameterizedTest {
 
 		@Test(expected = FileNotFoundException.class)
 		public void testOpenInputFileNotFound() throws IOException {
+			// instantiate class once for coverage
+			new UnixPipeUtils();
+
 			// Create a path that does not exist
 			File nonExistentFile = new File(folder.getRoot(), "nonexistent.pipe");
 

src/test/java/org/apache/sysds/test/component/utils/UnixPipeUtilsTest.java

@@ -24,6 +24,7 @@
 import org.apache.log4j.spi.LoggingEvent;
 import org.apache.sysds.api.PythonDMLScript;
 import org.apache.sysds.common.Types;
+import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
 import org.apache.sysds.runtime.util.UnixPipeUtils;
 import org.apache.sysds.test.LoggingUtils;
@@ -166,7 +167,7 @@ public void testStartupCorrect() throws Exception {
 
 	@Test
 	public void testDataTransfer() throws Exception {
-		PythonDMLScript.main(new String[]{"-python", "4002"});
+		PythonDMLScript.main(new String[]{"-python", "4003"});
 		Thread.sleep(200);
 		PythonDMLScript script = (PythonDMLScript) PythonDMLScript.GwS.getGateway().getEntryPoint();
 
@@ -200,7 +201,7 @@ public void testDataTransfer() throws Exception {
 
 	@Test
 	public void testDataTransferMultiPipes() throws Exception {
-		PythonDMLScript.main(new String[]{"-python", "4002"});
+		PythonDMLScript.main(new String[]{"-python", "4004"});
 		Thread.sleep(200);
 		PythonDMLScript script = (PythonDMLScript) PythonDMLScript.GwS.getGateway().getEntryPoint();
 
@@ -229,6 +230,30 @@ public void testDataTransferMultiPipes() throws Exception {
 		Thread.sleep(200);
 	}
 
+	@Test(expected = DMLRuntimeException.class)
+	public void testDataTransferNotInit1() throws Exception {
+		PythonDMLScript.main(new String[]{"-python", "4005"});
+		Thread.sleep(200);
+		PythonDMLScript script = (PythonDMLScript) PythonDMLScript.GwS.getGateway().getEntryPoint();
+		script.startReadingMbFromPipe(0, 2, 3, Types.ValueType.FP64);
+	}
+
+	@Test(expected = DMLRuntimeException.class)
+	public void testDataTransferNotInit2() throws Exception {
+		PythonDMLScript.main(new String[]{"-python", "4006"});
+		Thread.sleep(200);
+		PythonDMLScript script = (PythonDMLScript) PythonDMLScript.GwS.getGateway().getEntryPoint();
+		script.startWritingMbToPipe(0, null);
+	}
+
+	@Test(expected = DMLRuntimeException.class)
+	public void testDataTransferNotInit3() throws Exception {
+		PythonDMLScript.main(new String[]{"-python", "4007"});
+		Thread.sleep(200);
+		PythonDMLScript script = (PythonDMLScript) PythonDMLScript.GwS.getGateway().getEntryPoint();
+		script.startReadingMbFromPipes(new int[]{3,3}, 2, 3, Types.ValueType.FP64);
+	}
+
 	@SuppressWarnings("removal")
 	class NoExitSecurityManager extends SecurityManager {
 		@Override
@@ -239,4 +264,20 @@ public void checkExit(int status) {
 			throw new SecurityException("Intercepted exit()");
 		}
 	}
+
+	@Test(expected = DMLRuntimeException.class)
+	public void testDataTransferMaxValue1() throws Exception {
+		PythonDMLScript.main(new String[]{"-python", "4008"});
+		Thread.sleep(200);
+		PythonDMLScript script = (PythonDMLScript) PythonDMLScript.GwS.getGateway().getEntryPoint();
+		script.startReadingMbFromPipe(0, Integer.MAX_VALUE, 3, Types.ValueType.FP64);
+	}
+
+	@Test(expected = DMLRuntimeException.class)
+	public void testDataTransferMaxValue2() throws Exception {
+		PythonDMLScript.main(new String[]{"-python", "4009"});
+		Thread.sleep(200);
+		PythonDMLScript script = (PythonDMLScript) PythonDMLScript.GwS.getGateway().getEntryPoint();
+		script.startReadingMbFromPipes(new int[]{3,3}, Integer.MAX_VALUE, 2, Types.ValueType.FP64);
+	}
 }