(systemds) branch main updated: [SYSTEMDS-3675] Fix correctness multi-threaded central-moment/covariance

[mboehm7]

This is an automated email from the ASF dual-hosted git repository.

mboehm7 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/systemds.git


The following commit(s) were added to refs/heads/main by this push:
     new 6e79f819de [SYSTEMDS-3675] Fix correctness multi-threaded 
central-moment/covariance
6e79f819de is described below

commit 6e79f819de6e7ef241110ccaffce6d38c1f3720e
Author: Matthias Boehm <mboe...@gmail.com>
AuthorDate: Sat Mar 16 13:25:56 2024 +0100

    [SYSTEMDS-3675] Fix correctness multi-threaded central-moment/covariance
    
    After the recent change of the multi-threading threshold, the existing
    tests revealed a long-standing incorrectness of multi-threaded
    central-moment and covariance operations. During the aggregation of
    partial results, the output of the first task was not properly taken
    into account.
---
 src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixAgg.java | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git 
a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixAgg.java 
b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixAgg.java
index f8ac37c443..64a7bf3aa9 100644
--- a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixAgg.java
+++ b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixAgg.java
@@ -489,7 +489,7 @@ public class LibMatrixAgg {
                if( in1.isEmptyBlock(false) || 
!satisfiesMultiThreadingConstraints(in1, k) )
                        return aggregateCmCov(in1, in2, in3, fn);
                
-               CM_COV_Object ret = new CM_COV_Object();
+               CM_COV_Object ret = null;
                
                try {
                        ExecutorService pool = CommonThreadPool.get(k);
@@ -500,6 +500,7 @@ public class LibMatrixAgg {
                        List<Future<CM_COV_Object>> rtasks = 
pool.invokeAll(tasks);
                        pool.shutdown();
                        //aggregate partial results and error handling
+                       ret = rtasks.get(0).get();
                        for( int i=1; i<rtasks.size(); i++ )
                                fn.execute(ret, rtasks.get(i).get());
                }