You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemds.apache.org by ar...@apache.org on 2022/11/28 23:05:19 UTC
[systemds] branch main updated: [SYSTEMDS-3470] Generalize selecting reusable Spark instructions
This is an automated email from the ASF dual-hosted git repository.
arnabp20 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 03c546fe33 [SYSTEMDS-3470] Generalize selecting reusable Spark instructions
03c546fe33 is described below
commit 03c546fe3392759b18529db0ccb3d26328b02e0f
Author: Arnab Phani <ph...@gmail.com>
AuthorDate: Tue Nov 29 00:05:01 2022 +0100
[SYSTEMDS-3470] Generalize selecting reusable Spark instructions
This patch allows putting any Matrix Object in the lineage cache
which does not have a valid RDD. This change makes it easier to
separate Spark instructions which returns intermediate back to local.
Closes #1742
---
.../sysds/runtime/controlprogram/caching/CacheableData.java | 4 ++++
.../java/org/apache/sysds/runtime/lineage/LineageCache.java | 6 ++++++
.../apache/sysds/runtime/lineage/LineageCacheConfig.java | 2 +-
.../sysds/test/functions/async/LineageReuseSparkTest.java | 13 ++++++++++---
.../test/functions/builtin/part2/BuiltinNaLocfTest.java | 2 ++
5 files changed, 23 insertions(+), 4 deletions(-)
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheableData.java b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheableData.java
index 57cd49791b..9d195ea7fd 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheableData.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheableData.java
@@ -442,6 +442,10 @@ public abstract class CacheableData<T extends CacheBlock<?>> extends Data
if( _rddHandle != null )
rdd.setBackReference(this);
}
+
+ public boolean hasRDDHandle() {
+ return _rddHandle != null && _rddHandle.hasBackReference();
+ }
public BroadcastObject<T> getBroadcastHandle() {
return _bcHandle;
diff --git a/src/main/java/org/apache/sysds/runtime/lineage/LineageCache.java b/src/main/java/org/apache/sysds/runtime/lineage/LineageCache.java
index 4c79b885b7..ecd734c588 100644
--- a/src/main/java/org/apache/sysds/runtime/lineage/LineageCache.java
+++ b/src/main/java/org/apache/sysds/runtime/lineage/LineageCache.java
@@ -579,6 +579,12 @@ public class LineageCache
continue;
}
+ if (data instanceof MatrixObject && ((MatrixObject) data).hasRDDHandle()) {
+ // Avoid triggering pre-matured Spark instruction chains
+ removePlaceholder(item);
+ continue;
+ }
+
if (LineageCacheConfig.isOutputFederated(inst, data)) {
// Do not cache federated outputs (in the coordinator)
// Cannot skip putting the placeholder as the above is only known after execution
diff --git a/src/main/java/org/apache/sysds/runtime/lineage/LineageCacheConfig.java b/src/main/java/org/apache/sysds/runtime/lineage/LineageCacheConfig.java
index 4259481444..5a7c46dfe7 100644
--- a/src/main/java/org/apache/sysds/runtime/lineage/LineageCacheConfig.java
+++ b/src/main/java/org/apache/sysds/runtime/lineage/LineageCacheConfig.java
@@ -206,7 +206,7 @@ public class LineageCacheConfig
boolean insttype = (inst instanceof ComputationCPInstruction
|| inst instanceof ComputationFEDInstruction
|| inst instanceof GPUInstruction
- || (inst instanceof ComputationSPInstruction && isRightSparkOp(inst)))
+ || inst instanceof ComputationSPInstruction)
&& !(inst instanceof ListIndexingCPInstruction);
boolean rightop = (ArrayUtils.contains(REUSE_OPCODES, inst.getOpcode())
|| (inst.getOpcode().equals("append") && isVectorAppend(inst, ec))
diff --git a/src/test/java/org/apache/sysds/test/functions/async/LineageReuseSparkTest.java b/src/test/java/org/apache/sysds/test/functions/async/LineageReuseSparkTest.java
index 83992c5772..e958093122 100644
--- a/src/test/java/org/apache/sysds/test/functions/async/LineageReuseSparkTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/async/LineageReuseSparkTest.java
@@ -51,15 +51,22 @@ public class LineageReuseSparkTest extends AutomatedTestBase {
}
@Test
- public void testlmds() {
- runTest(TEST_NAME+"1");
+ public void testlmdsHB() {
+ runTest(TEST_NAME+"1", ExecMode.HYBRID);
}
- public void runTest(String testname) {
+ @Test
+ public void testlmdsSP() {
+ // Only reuse the actions
+ runTest(TEST_NAME+"1", ExecMode.SPARK);
+ }
+
+ public void runTest(String testname, ExecMode execMode) {
boolean old_simplification = OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION;
boolean old_sum_product = OptimizerUtils.ALLOW_SUM_PRODUCT_REWRITES;
boolean old_trans_exec_type = OptimizerUtils.ALLOW_TRANSITIVE_SPARK_EXEC_TYPE;
ExecMode oldPlatform = setExecMode(ExecMode.HYBRID);
+ rtplatform = execMode;
long oldmem = InfrastructureAnalyzer.getLocalMaxMemory();
long mem = 1024*1024*8;
diff --git a/src/test/java/org/apache/sysds/test/functions/builtin/part2/BuiltinNaLocfTest.java b/src/test/java/org/apache/sysds/test/functions/builtin/part2/BuiltinNaLocfTest.java
index 4d9ddac658..64999e46ae 100644
--- a/src/test/java/org/apache/sysds/test/functions/builtin/part2/BuiltinNaLocfTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/builtin/part2/BuiltinNaLocfTest.java
@@ -22,6 +22,7 @@ package org.apache.sysds.test.functions.builtin.part2;
import org.apache.commons.lang.ArrayUtils;
import org.apache.sysds.common.Types;
import org.apache.sysds.common.Types.ExecType;
+import org.apache.sysds.runtime.lineage.Lineage;
import org.apache.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType;
import org.apache.sysds.runtime.matrix.data.MatrixValue;
import org.apache.sysds.test.AutomatedTestBase;
@@ -105,6 +106,7 @@ public class BuiltinNaLocfTest extends AutomatedTestBase {
double[][] A = getRandomMatrix(rows, cols, -10, 10, 0.6, 7);
writeInputMatrixWithMTD("A", A, true);
+ Lineage.resetInternalState();
runTest(true, false, null, -1);
runRScript(true);
//compare matrices