[systemds] branch main updated: [SYSTEMDS-3429] Federated Multithreaded transformencode

[baunsgaard]

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

baunsgaard 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 652b4077be [SYSTEMDS-3429] Federated Multithreaded transformencode
652b4077be is described below

commit 652b4077be6d5dd8b453b775bfe230a805a3f390
Author: Andreas Botzner <andr...@botzner.com>
AuthorDate: Fri Aug 26 14:13:12 2022 +0200

    [SYSTEMDS-3429] Federated Multithreaded transformencode
    
    This commit adds a patch to transfer transform encode as multithreaded
    to the federated site.
    
    Closes #1689
---
 .../fed/MultiReturnParameterizedBuiltinFEDInstruction.java         | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git 
a/src/main/java/org/apache/sysds/runtime/instructions/fed/MultiReturnParameterizedBuiltinFEDInstruction.java
 
b/src/main/java/org/apache/sysds/runtime/instructions/fed/MultiReturnParameterizedBuiltinFEDInstruction.java
index c9135eb013..93a7e41291 100644
--- 
a/src/main/java/org/apache/sysds/runtime/instructions/fed/MultiReturnParameterizedBuiltinFEDInstruction.java
+++ 
b/src/main/java/org/apache/sysds/runtime/instructions/fed/MultiReturnParameterizedBuiltinFEDInstruction.java
@@ -36,6 +36,7 @@ import org.apache.sysds.common.Types.DataType;
 import org.apache.sysds.common.Types.ValueType;
 import org.apache.sysds.hops.fedplanner.FTypes;
 import org.apache.sysds.hops.fedplanner.FTypes.FType;
+import org.apache.sysds.hops.OptimizerUtils;
 import org.apache.sysds.lops.PickByCount;
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.controlprogram.caching.CacheableData;
@@ -84,7 +85,7 @@ public class MultiReturnParameterizedBuiltinFEDInstruction 
extends ComputationFE
                MultiReturnParameterizedBuiltinCPInstruction inst, 
ExecutionContext ec) {
                if(inst.getOpcode().equals("transformencode") && 
inst.input1.isFrame()) {
                        CacheableData<?> fo = ec.getCacheableData(inst.input1);
-                       if(fo.isFederatedExcept(FType.BROADCAST)) 
+                       if(fo.isFederatedExcept(FType.BROADCAST))
                                return 
MultiReturnParameterizedBuiltinFEDInstruction.parseInstruction(inst);
                }
                return null;
@@ -339,7 +340,7 @@ public class MultiReturnParameterizedBuiltinFEDInstruction 
extends ComputationFE
                                .createEncoder(_spec, colNames, 
fb.getNumColumns(), null, _offset, _offset + fb.getNumColumns());
 
                        // build necessary structures for encoding
-                       encoder.build(fb); // FIXME skip equi-height sorting
+                       encoder.build(fb, 
OptimizerUtils.getTransformNumThreads()); // FIXME skip equi-height sorting
                        fo.release();
 
                        // create federated response
@@ -370,7 +371,7 @@ public class MultiReturnParameterizedBuiltinFEDInstruction 
extends ComputationFE
                        // offset is applied on the Worker to shift the local 
encoders to their respective column
                        _encoder.applyColumnOffset();
                        // apply transformation
-                       MatrixBlock mbout = _encoder.apply(fb);
+                       MatrixBlock mbout = _encoder.apply(fb, 
OptimizerUtils.getTransformNumThreads());
 
                        // create output matrix object
                        MatrixObject mo = 
ExecutionContext.createMatrixObject(mbout);