[SYSTEMDS-2719] Lineage exploitation in the buffer pool (datagen ops)
For matrices that are results of data generation instructions such as
RAND, we store their lineage alongside the data. Under memory pressure,
these matrices do not go to the buffer pool. We recompute them from
lineage instead of reading evicted buffers from FS. This potentially
results in fewer evictions and IO.
AMLS project SS2020.
Closes #1096.
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheStatistics.java b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheStatistics.java
index 975492a..a001e65 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheStatistics.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheStatistics.java
@@ -40,9 +40,11 @@
CACHE_HITS_FSBUFF,
CACHE_HITS_FS,
CACHE_HITS_HDFS,
+ CACHE_HITS_LIN,
CACHE_WRITES_FSBUFF,
CACHE_WRITES_FS,
CACHE_WRITES_HDFS,
+ CACHE_WRITES_LIN,
CACHE_TIME_ACQR, //acquire read
CACHE_TIME_ACQM, //acquire read
CACHE_TIME_RLS, //release
@@ -54,11 +56,13 @@
private static final LongAdder _numHitsFSBuff = new LongAdder();
private static final LongAdder _numHitsFS = new LongAdder();
private static final LongAdder _numHitsHDFS = new LongAdder();
-
+ private static final LongAdder _numHitsLin = new LongAdder();
+
//write statistics caching
private static final LongAdder _numWritesFSBuff = new LongAdder();
private static final LongAdder _numWritesFS = new LongAdder();
private static final LongAdder _numWritesHDFS = new LongAdder();
+ private static final LongAdder _numWritesLin = new LongAdder();
//time statistics caching
private static final LongAdder _ctimeAcquireR = new LongAdder(); //in nano sec
@@ -68,6 +72,7 @@
public static void reset() {
_numHitsMem.reset();
+ _numHitsLin.reset();
_numHitsFSBuff.reset();
_numHitsFS.reset();
_numHitsHDFS.reset();
@@ -75,6 +80,7 @@
_numWritesFSBuff.reset();
_numWritesFS.reset();
_numWritesHDFS.reset();
+ _numWritesLin.reset();
_ctimeAcquireR.reset();
_ctimeAcquireM.reset();
@@ -130,6 +136,18 @@
return _numHitsHDFS.longValue();
}
+ public static void incrementLinHits() {
+ _numHitsLin.increment();
+ }
+
+ public static void incrementLinHits(int delta) {
+ _numHitsLin.add(delta);
+ }
+
+ public static long getLinHits() {
+ return _numHitsLin.longValue();
+ }
+
public static void incrementFSBuffWrites() {
_numWritesFSBuff.increment();
}
@@ -165,6 +183,18 @@
public static long getHDFSWrites() {
return _numWritesHDFS.longValue();
}
+
+ public static void incrementLinWrites() {
+ _numWritesLin.increment();
+ }
+
+ public static void incrementLinWrites(int delta) {
+ _numWritesLin.add(delta);
+ }
+
+ public static long getLinWrites() {
+ return _numWritesLin.longValue();
+ }
public static void incrementAcquireRTime(long delta) {
_ctimeAcquireR.add(delta);
@@ -198,10 +228,12 @@
return _ctimeExport.longValue();
}
- public static String displayHits() {
+ public static String displayHits() {
StringBuilder sb = new StringBuilder();
sb.append(_numHitsMem.longValue());
sb.append("/");
+ sb.append(_numHitsLin.longValue());
+ sb.append("/");
sb.append(_numHitsFSBuff.longValue());
sb.append("/");
sb.append(_numHitsFS.longValue());
@@ -211,8 +243,10 @@
return sb.toString();
}
- public static String displayWrites() {
+ public static String displayWrites() {
StringBuilder sb = new StringBuilder();
+ sb.append(_numWritesLin.longValue());
+ sb.append("/");
sb.append(_numWritesFSBuff.longValue());
sb.append("/");
sb.append(_numWritesFS.longValue());
@@ -222,7 +256,7 @@
return sb.toString();
}
- public static String displayTime() {
+ public static String displayTime() {
StringBuilder sb = new StringBuilder();
sb.append(String.format("%.3f", ((double)_ctimeAcquireR.longValue())/1000000000)); //in sec
sb.append("/");
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 97f211a..9b465a2 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
@@ -52,6 +52,7 @@
import org.apache.sysds.runtime.io.FileFormatProperties;
import org.apache.sysds.runtime.io.IOUtilFunctions;
import org.apache.sysds.runtime.io.ReaderWriterFederated;
+import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.meta.DataCharacteristics;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
import org.apache.sysds.runtime.meta.MetaData;
@@ -212,6 +213,8 @@
private RDDObject _rddHandle = null; //RDD handle
private BroadcastObject<T> _bcHandle = null; //Broadcast handle
protected HashMap<GPUContext, GPUObject> _gpuObjects = null; //Per GPUContext object allocated on GPU
+
+ private LineageItem _lineage = null;
/**
* Basic constructor for any cacheable data.
@@ -341,6 +344,18 @@
public abstract void refreshMetaData();
+ public LineageItem getCacheLineage() {
+ return _lineage;
+ }
+
+ public void setCacheLineage(LineageItem li) {
+ _lineage = li;
+ }
+
+ public boolean hasValidLineage() {
+ return (_lineage != null);
+ }
+
/**
* Check if object is federated.
* @return true if federated else false
@@ -490,11 +505,17 @@
//read data from HDFS/RDD if required
//(probe data for cache_nowrite / jvm_reuse)
- if( _data==null && isEmpty(true) ) {
+ if( _data==null && ( isEmpty(true) || hasValidLineage() )) {
try {
- if( isFederated() ) {
- _data = readBlobFromFederated( _fedMapping );
-
+ if( hasValidLineage() ) {
+ _data = reconstructByLineage(getCacheLineage());
+ _requiresLocalWrite = false;
+ if( DMLScript.STATISTICS )
+ CacheStatistics.incrementLinHits();
+ }
+ else if( isFederated() ) {
+ _data = readBlobFromFederated(_fedMapping);
+
//mark for initial local write despite read operation
_requiresLocalWrite = CACHING_WRITE_CACHE_ON_READ;
}
@@ -638,7 +659,7 @@
&& isCached(true) //not empty and not read/modify
&& !isBelowCachingThreshold() ) //min size for caching
{
- if( write || _requiresLocalWrite ) {
+ if( ( write && !hasValidLineage() ) || _requiresLocalWrite ) {
String filePath = getCacheFilePathAndName();
try {
LazyWriteBuffer.writeBlock(filePath, _data);
@@ -648,6 +669,9 @@
}
_requiresLocalWrite = false;
}
+
+ if( DMLScript.STATISTICS && write && hasValidLineage() )
+ CacheStatistics.incrementLinWrites();
//create cache
createCache();
@@ -956,6 +980,10 @@
return (_data.getInMemorySize() <= CACHING_THRESHOLD);
}
+ public static boolean isBelowCachingThreshold(CacheBlock data) {
+ return LazyWriteBuffer.getCacheBlockSize(data) <= CACHING_THRESHOLD;
+ }
+
public long getDataSize() {
return (_data != null) ?_data.getInMemorySize() : 0;
}
@@ -1004,6 +1032,10 @@
protected abstract void writeBlobFromRDDtoHDFS(RDDObject rdd, String fname, String ofmt)
throws IOException;
+ protected abstract T reconstructByLineage(LineageItem li)
+ throws IOException;
+
+
protected void writeMetaData (String filePathAndName, String outputFormat, FileFormatProperties formatProperties)
throws IOException
{
@@ -1011,7 +1043,7 @@
if (iimd == null)
throw new DMLRuntimeException("Unexpected error while writing mtd file (" + filePathAndName + ") -- metadata is null.");
-
+
// Write the matrix to HDFS in requested format
FileFormat fmt = (outputFormat != null) ? FileFormat.safeValueOf(outputFormat) : iimd.getFileFormat();
if ( fmt != FileFormat.MM ) {
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/FrameObject.java b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/FrameObject.java
index 0418ce7..36cd0f0 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/FrameObject.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/FrameObject.java
@@ -37,6 +37,8 @@
import org.apache.sysds.runtime.io.FrameReaderFactory;
import org.apache.sysds.runtime.io.FrameWriter;
import org.apache.sysds.runtime.io.FrameWriterFactory;
+import org.apache.sysds.runtime.lineage.LineageItem;
+import org.apache.sysds.runtime.lineage.LineageRecomputeUtils;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.meta.DataCharacteristics;
import org.apache.sysds.runtime.meta.MetaData;
@@ -286,4 +288,11 @@
//lazy evaluation of pending transformations.
SparkExecutionContext.writeFrameRDDtoHDFS(rdd, fname, iimd.getFileFormat());
}
+
+ @Override
+ protected FrameBlock reconstructByLineage(LineageItem li) throws IOException {
+ return ((FrameObject) LineageRecomputeUtils
+ .parseNComputeLineageTrace(li.getData(), null))
+ .acquireReadAndRelease();
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/MatrixObject.java b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/MatrixObject.java
index 61bc5b8..d0bce6e 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/MatrixObject.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/MatrixObject.java
@@ -44,6 +44,8 @@
import org.apache.sysds.runtime.instructions.spark.data.RDDObject;
import org.apache.sysds.runtime.io.FileFormatProperties;
import org.apache.sysds.runtime.io.ReaderWriterFederated;
+import org.apache.sysds.runtime.lineage.LineageItem;
+import org.apache.sysds.runtime.lineage.LineageRecomputeUtils;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.meta.DataCharacteristics;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
@@ -629,4 +631,11 @@
long newnnz = SparkExecutionContext.writeMatrixRDDtoHDFS(rdd, fname, fmt);
_metaData.getDataCharacteristics().setNonZeros(newnnz);
}
+
+ @Override
+ protected MatrixBlock reconstructByLineage(LineageItem li) throws IOException {
+ return ((MatrixObject) LineageRecomputeUtils
+ .parseNComputeLineageTrace(li.getData(), null))
+ .acquireReadAndRelease();
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/TensorObject.java b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/TensorObject.java
index 32fb72b..6b29d41 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/TensorObject.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/TensorObject.java
@@ -33,6 +33,8 @@
import org.apache.sysds.runtime.data.TensorIndexes;
import org.apache.sysds.runtime.instructions.spark.data.RDDObject;
import org.apache.sysds.runtime.io.FileFormatProperties;
+import org.apache.sysds.runtime.lineage.LineageItem;
+import org.apache.sysds.runtime.lineage.LineageRecomputeUtils;
import org.apache.sysds.runtime.meta.DataCharacteristics;
import org.apache.sysds.runtime.meta.MetaData;
import org.apache.sysds.runtime.meta.MetaDataFormat;
@@ -190,4 +192,11 @@
throws DMLRuntimeException {
//TODO rdd write
}
+
+ @Override
+ protected TensorBlock reconstructByLineage(LineageItem li) throws IOException {
+ return ((TensorObject) LineageRecomputeUtils
+ .parseNComputeLineageTrace(li.getData(), null))
+ .acquireReadAndRelease();
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/context/ExecutionContext.java b/src/main/java/org/apache/sysds/runtime/controlprogram/context/ExecutionContext.java
index fceaea4..9587b95 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/context/ExecutionContext.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/context/ExecutionContext.java
@@ -534,12 +534,16 @@
}
public void setMatrixOutput(String varName, MatrixBlock outputData) {
+ setMatrixOutputAndLineage(varName, outputData, null);
+ }
+
+ public void setMatrixOutputAndLineage(String varName, MatrixBlock outputData, LineageItem li) {
if( isAutoCreateVars() && !containsVariable(varName) )
setVariable(varName, createMatrixObject(outputData));
MatrixObject mo = getMatrixObject(varName);
mo.acquireModify(outputData);
+ mo.setCacheLineage(li);
mo.release();
- setVariable(varName, mo);
}
public void setMatrixOutput(String varName, MatrixBlock outputData, UpdateType flag) {
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/DataGenCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/DataGenCPInstruction.java
index af74498..86b95a3 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/DataGenCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/DataGenCPInstruction.java
@@ -30,6 +30,7 @@
import org.apache.sysds.lops.DataGen;
import org.apache.sysds.lops.Lop;
import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.controlprogram.caching.CacheableData;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysds.runtime.data.TensorBlock;
import org.apache.sysds.runtime.instructions.InstructionUtils;
@@ -375,11 +376,15 @@
soresBlock.examSparsity();
//release created output
- ec.setMatrixOutput(output.getName(), soresBlock);
- } else if(output.isTensor()) {
+ LineageItem lin = CacheableData.isBelowCachingThreshold(soresBlock) ?
+ null : getLineageItem(ec).getValue();
+ ec.setMatrixOutputAndLineage(output.getName(), soresBlock, lin);
+ }
+ else if(output.isTensor()) {
// TODO memory optimization
ec.setTensorOutput(output.getName(), tensorBlock);
- } else if( output.isScalar() )
+ }
+ else if( output.isScalar() )
ec.setScalarOutput(output.getName(), soresScalar);
}
diff --git a/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java b/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
index 032b79e..e5334fb 100644
--- a/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
@@ -144,7 +144,7 @@
protected SparseBlock sparseBlock = null;
//sparse-block-specific attributes (allocation only)
- protected int estimatedNNzsPerRow = -1;
+ protected int estimatedNNzsPerRow = -1;
////////
// Matrix Constructors
@@ -470,7 +470,7 @@
public int getNumColumns() {
return clen;
}
-
+
public void setNumColumns(int c) {
clen = c;
}
@@ -1644,7 +1644,6 @@
merge((MatrixBlock)that, appendOnly);
}
-
/**
* Merge disjoint: merges all non-zero values of the given input into the current
* matrix block. Note that this method does NOT check for overlapping entries;
diff --git a/src/main/java/org/apache/sysds/utils/Statistics.java b/src/main/java/org/apache/sysds/utils/Statistics.java
index 6f6b26c..f196968 100644
--- a/src/main/java/org/apache/sysds/utils/Statistics.java
+++ b/src/main/java/org/apache/sysds/utils/Statistics.java
@@ -969,8 +969,8 @@
String.format("%.3f", examSparsityTime*1e-9) + "/" + String.format("%.3f", allocateDoubleArrTime*1e-9) + ".\n");
}
- sb.append("Cache hits (Mem, WB, FS, HDFS):\t" + CacheStatistics.displayHits() + ".\n");
- sb.append("Cache writes (WB, FS, HDFS):\t" + CacheStatistics.displayWrites() + ".\n");
+ sb.append("Cache hits (Mem/Li/WB/FS/HDFS):\t" + CacheStatistics.displayHits() + ".\n");
+ sb.append("Cache writes (Li/WB/FS/HDFS):\t" + CacheStatistics.displayWrites() + ".\n");
sb.append("Cache times (ACQr/m, RLS, EXP):\t" + CacheStatistics.displayTime() + " sec.\n");
if (DMLScript.JMLC_MEM_STATISTICS)
sb.append("Max size of live objects:\t" + byteCountToDisplaySize(getSizeofPinnedObjects()) + " (" + getNumPinnedObjects() + " total objects)" + "\n");
diff --git a/src/test/java/org/apache/sysds/test/functions/lineage/LineageExploitationBufferPoolTest.java b/src/test/java/org/apache/sysds/test/functions/lineage/LineageExploitationBufferPoolTest.java
new file mode 100644
index 0000000..3e6ed72
--- /dev/null
+++ b/src/test/java/org/apache/sysds/test/functions/lineage/LineageExploitationBufferPoolTest.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.test.functions.lineage;
+
+import org.apache.sysds.hops.OptimizerUtils;
+import org.apache.sysds.hops.recompile.Recompiler;
+import org.apache.sysds.runtime.lineage.Lineage;
+import org.apache.sysds.test.TestConfiguration;
+import org.apache.sysds.test.TestUtils;
+import org.junit.Test;
+import org.junit.Assert;
+import org.apache.sysds.runtime.controlprogram.caching.CacheStatistics;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class LineageExploitationBufferPoolTest extends LineageBase
+{
+ protected static final String TEST_DIR = "functions/lineage/";
+ protected static final String TEST_NAME1 = "LineageExploitationBufferPool1";
+ protected String TEST_CLASS_DIR = TEST_DIR + LineageExploitationBufferPoolTest.class.getSimpleName() + "/";
+
+ @Override
+ public void setUp() {
+ TestUtils.clearAssertionInformation();
+ addTestConfiguration(TEST_NAME1, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME1));
+ }
+
+ @Test
+ public void testLineageReusePerf1() { testLineageExploitationBufferPool(TEST_NAME1); }
+
+ public void testLineageExploitationBufferPool(String testname) {
+ boolean old_simplification = OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION;
+ boolean old_sum_product = OptimizerUtils.ALLOW_SUM_PRODUCT_REWRITES;
+
+ try {
+
+ LOG.debug("------------ BEGIN " + testname + "------------");
+
+ /* Test description
+ */
+
+ OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION = false;
+ OptimizerUtils.ALLOW_SUM_PRODUCT_REWRITES = false;
+
+ getAndLoadTestConfiguration(testname);
+ fullDMLScriptName = getScript();
+ // System.out.println(LineageCacheEviction.getCacheLimit());
+
+ // costnsize scheme (computationTime/Size)
+ List<String> proArgs = new ArrayList<>();
+ // proArgs.clear();
+ proArgs.add("-stats");
+ proArgs.add("-lineage");
+ proArgs.add("-args");
+ proArgs.add(output("Z"));
+ programArgs = proArgs.toArray(new String[proArgs.size()]);
+ Lineage.resetInternalState();
+
+ runTest(true, EXCEPTION_NOT_EXPECTED, null, -1);
+
+ Assert.assertEquals(6, CacheStatistics.getLinWrites());
+ String[] writes = CacheStatistics.displayWrites().split("/");
+ Assert.assertEquals(6, Long.parseLong(writes[0])); // writes WB
+ Assert.assertEquals(5, Long.parseLong(writes[1])); // writes WB
+
+ } finally {
+ OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION = old_simplification;
+ OptimizerUtils.ALLOW_SUM_PRODUCT_REWRITES = old_sum_product;
+ Recompiler.reinitRecompiler();
+ }
+
+ }
+}
diff --git a/src/test/scripts/functions/lineage/LineageExploitationBufferPool1.dml b/src/test/scripts/functions/lineage/LineageExploitationBufferPool1.dml
new file mode 100644
index 0000000..619caae
--- /dev/null
+++ b/src/test/scripts/functions/lineage/LineageExploitationBufferPool1.dml
@@ -0,0 +1,34 @@
+#-------------------------------------------------------------
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+#-------------------------------------------------------------
+
+size = 1000
+
+X = rand(rows=size, cols=size, sparsity=1.0, seed=1)
+ %*% rand(rows=size, cols=size, sparsity=1.0, seed=2)
+ %*% rand(rows=size, cols=size, sparsity=1.0, seed=3);
+
+Y = rand(rows=size, cols=size, sparsity=1.0, seed=4)
+ %*% rand(rows=size, cols=size, sparsity=1.0, seed=5)
+ %*% rand(rows=size, cols=size, sparsity=1.0, seed=6);
+
+Z = X %*% Y;
+
+write(Z, $1, format="text");
