You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemds.apache.org by mb...@apache.org on 2020/11/08 20:53:11 UTC
[systemds] branch master updated: [SYSTEMDS-2719] Lineage
exploitation in the buffer pool (datagen ops)
This is an automated email from the ASF dual-hosted git repository.
mboehm7 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/systemds.git
The following commit(s) were added to refs/heads/master by this push:
new f2820a1 [SYSTEMDS-2719] Lineage exploitation in the buffer pool (datagen ops)
f2820a1 is described below
commit f2820a1f8dca214a82b66c34819b7eeb6dba7ba6
Author: Sergey Redyuk <se...@gmail.com>
AuthorDate: Sun Nov 8 20:05:44 2020 +0100
[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.
---
.../controlprogram/caching/CacheStatistics.java | 42 +++++++++-
.../controlprogram/caching/CacheableData.java | 44 +++++++++--
.../controlprogram/caching/FrameObject.java | 9 +++
.../controlprogram/caching/MatrixObject.java | 9 +++
.../controlprogram/caching/TensorObject.java | 9 +++
.../controlprogram/context/ExecutionContext.java | 6 +-
.../instructions/cp/DataGenCPInstruction.java | 11 ++-
.../sysds/runtime/matrix/data/MatrixBlock.java | 5 +-
.../java/org/apache/sysds/utils/Statistics.java | 4 +-
.../lineage/LineageExploitationBufferPoolTest.java | 91 ++++++++++++++++++++++
.../lineage/LineageExploitationBufferPool1.dml | 34 ++++++++
11 files changed, 245 insertions(+), 19 deletions(-)
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 @@ public class CacheStatistics
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 @@ public class CacheStatistics
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 class CacheStatistics
public static void reset() {
_numHitsMem.reset();
+ _numHitsLin.reset();
_numHitsFSBuff.reset();
_numHitsFS.reset();
_numHitsHDFS.reset();
@@ -75,6 +80,7 @@ public class CacheStatistics
_numWritesFSBuff.reset();
_numWritesFS.reset();
_numWritesHDFS.reset();
+ _numWritesLin.reset();
_ctimeAcquireR.reset();
_ctimeAcquireM.reset();
@@ -130,6 +136,18 @@ public class CacheStatistics
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 class CacheStatistics
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 @@ public class CacheStatistics
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 @@ public class CacheStatistics
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 @@ public class CacheStatistics
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.instructions.spark.data.RDDObject;
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 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
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 class CacheableData<T extends CacheBlock> extends Data
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 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
//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 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
&& 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 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
}
_requiresLocalWrite = false;
}
+
+ if( DMLScript.STATISTICS && write && hasValidLineage() )
+ CacheStatistics.incrementLinWrites();
//create cache
createCache();
@@ -956,6 +980,10 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
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 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
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 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
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.FileFormatProperties;
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 @@ public class FrameObject extends CacheableData<FrameBlock>
//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.fed.InitFEDInstruction;
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 @@ public class MatrixObject extends CacheableData<MatrixBlock>
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.TensorBlock;
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 @@ public class TensorObject extends CacheableData<TensorBlock> {
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 class ExecutionContext {
}
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.hops.OptimizerUtils;
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 @@ public class DataGenCPInstruction extends UnaryCPInstruction {
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 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
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 class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
public int getNumColumns() {
return clen;
}
-
+
public void setNumColumns(int c) {
clen = c;
}
@@ -1644,7 +1644,6 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
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 @@ public class Statistics
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");