formatting

Author: e-strauss

.github/workflows/python.yml

@@ -45,6 +45,7 @@ on:
 jobs:
   test:
     runs-on: ${{ matrix.os }}
+    timeout-minutes: 60
     strategy:
       fail-fast: false
       matrix:

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

@@ -255,8 +255,15 @@ def exception_and_close(self, exception, trace_back_limit: int = None):
 
         # print error in case we encounter another exception during closing
         # also print stacktrace of the original exception
-        tb_str = exception if isinstance(exception, str) \
-            else ''.join(traceback.format_exception(type(exception), exception, exception.__traceback__))
+        tb_str = (
+            exception
+            if isinstance(exception, str)
+            else "".join(
+                traceback.format_exception(
+                    type(exception), exception, exception.__traceback__
+                )
+            )
+        )
         print("Encountered exception & shutting down: \n\n{}".format(tb_str.strip()))
 
         self.close()

src/main/python/systemds/operator/nodes/multi_return.py

@@ -81,7 +81,9 @@ def _parse_output_result_variables(self, result_variables):
             output = self._outputs[idx]
             if str(output) == "MatrixNode":
                 result_var.append(
-                    matrix_block_to_numpy(jvmV, result_variables.getMatrixBlock(v))
+                    matrix_block_to_numpy(
+                        self.sds_context, result_variables.getMatrixBlock(v)
+                    )
                 )
             elif str(output) == "FrameNode":
                 result_var.append(

src/main/python/systemds/utils/converters.py

@@ -32,7 +32,7 @@
 
 
 def format_bytes(size):
-    for unit in ['Bytes', 'KB', 'MB', 'GB', 'TB', 'PB']:
+    for unit in ["Bytes", "KB", "MB", "GB", "TB", "PB"]:
         if size < 1024.0:
             return f"{size:.2f} {unit}"
         size /= 1024.0
@@ -123,7 +123,9 @@ def numpy_to_matrix_block(sds, np_arr: np.array):
             not sds.multi_pipe_enabled or total_bytes < 2 * min_bytes_per_pipe
         )
         if use_single_pipe:
-            sds._log.debug("Using single FIFO pipe for {}".format(format_bytes(total_bytes)))
+            sds._log.debug(
+                "Using single FIFO pipe for {}".format(format_bytes(total_bytes))
+            )
             pipe_id = 0
             pipe = sds.FIFO_PY2JAVA_PIPES[pipe_id]
             fut = sds.executor_pool.submit(
@@ -193,19 +195,18 @@ def matrix_block_to_numpy(sds, mb: JavaObject):
 
     rows = mb.getNumRows()
     cols = mb.getNumColumns()
+    try:
+        if sds.data_transfer_mode == 1:
+            dtype = np.float64
 
-    if sds.data_transfer_mode == 1:
-        dtype = np.float64
+            elem_size = np.dtype(dtype).itemsize
+            num_elements = rows * cols
+            total_bytes = num_elements * elem_size
+            batch_size_bytes = 32 * 1024  # 32 KB
 
-        elem_size = np.dtype(dtype).itemsize
-        num_elements = rows * cols
-        total_bytes = num_elements * elem_size
-        batch_size_bytes = 32 * 1024  # 32 KB
+            arr = np.empty(num_elements, dtype=dtype)
+            mv = memoryview(arr).cast("B")
 
-        arr = np.empty(num_elements, dtype=dtype)
-        mv = memoryview(arr).cast("B")
-
-        try:
             pipe_id = 0
             pipe = sds.FIFO_JAVA2PY_PIPES[pipe_id]
 
@@ -219,15 +220,16 @@ def matrix_block_to_numpy(sds, mb: JavaObject):
             fut.result()  # wait for Java to finish writing
             return arr.reshape((rows, cols))
 
-        except Exception as e:
-            sds.exception_and_close(e)
-    else:
-        buf = jvm.org.apache.sysds.runtime.util.Py4jConverterUtils.convertMBtoPy4JDenseArr(
-            mb
-        )
-        return np.frombuffer(buf, count=rows * cols, dtype=np.float64).reshape(
-            (rows, cols)
-        )
+        else:
+            buf = jvm.org.apache.sysds.runtime.util.Py4jConverterUtils.convertMBtoPy4JDenseArr(
+                mb
+            )
+            return np.frombuffer(buf, count=rows * cols, dtype=np.float64).reshape(
+                (rows, cols)
+            )
+    except Exception as e:
+        sds.exception_and_close(e)
+        return None
 
 
 def convert(jvm, fb, idx, num_elements, value_type, pd_series, conversion="column"):