You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemml.apache.org by mb...@apache.org on 2018/03/30 05:02:14 UTC
[1/2] systemml git commit: [SYSTEMML-2016] Performance multi-threaded
matrix multiply thresholds
Repository: systemml
Updated Branches:
refs/heads/master 5ef4f5ac2 -> 094f5551d
[SYSTEMML-2016] Performance multi-threaded matrix multiply thresholds
Recently, we've introduced a common thread pool for all multi-threaded
operations, which eliminated the costs for thread pool creation per
operation. Therefore, multi-threading is now beneficial for smaller data
sizes. This patch accordingly updates the size and FLOP thresholds used
for matrix multiply (mm) and mmchain operations. In addition, we now
also use a slightly more efficient approach for aggregating partial
results for mmchain.
For example, on a scenario of 10,000 operations, this patch improved the
cumulative runtime for [1Kx1K] mm [1Kx1] from 8.1s to 1.7s, and for
[500x1K] mmchain [1Kx1] from 6.5s to 4s.
Project: http://git-wip-us.apache.org/repos/asf/systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/systemml/commit/665d037e
Tree: http://git-wip-us.apache.org/repos/asf/systemml/tree/665d037e
Diff: http://git-wip-us.apache.org/repos/asf/systemml/diff/665d037e
Branch: refs/heads/master
Commit: 665d037e68d1ac1f359c64ee36d5395ad704a63e
Parents: 5ef4f5a
Author: Matthias Boehm <mb...@gmail.com>
Authored: Thu Mar 29 16:53:08 2018 -0700
Committer: Matthias Boehm <mb...@gmail.com>
Committed: Thu Mar 29 16:53:08 2018 -0700
----------------------------------------------------------------------
.../sysml/runtime/codegen/SpoofRowwise.java | 2 +-
.../runtime/matrix/data/LibMatrixMult.java | 76 +++++++++++++-------
2 files changed, 53 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/systemml/blob/665d037e/src/main/java/org/apache/sysml/runtime/codegen/SpoofRowwise.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/codegen/SpoofRowwise.java b/src/main/java/org/apache/sysml/runtime/codegen/SpoofRowwise.java
index a4e2e7a..7454f78 100644
--- a/src/main/java/org/apache/sysml/runtime/codegen/SpoofRowwise.java
+++ b/src/main/java/org/apache/sysml/runtime/codegen/SpoofRowwise.java
@@ -174,7 +174,7 @@ public abstract class SpoofRowwise extends SpoofOperator
public MatrixBlock execute(ArrayList<MatrixBlock> inputs, ArrayList<ScalarObject> scalarObjects, MatrixBlock out, int k)
{
//redirect to serial execution
- if( k <= 1 || (_type.isColumnAgg() && !LibMatrixMult.checkParColumnAgg(inputs.get(0), k, false))
+ if( k <= 1 || (_type.isColumnAgg() && !LibMatrixMult.satisfiesMultiThreadingConstraints(inputs.get(0), k))
|| getTotalInputSize(inputs) < PAR_NUMCELL_THRESHOLD ) {
return execute(inputs, scalarObjects, out);
}
http://git-wip-us.apache.org/repos/asf/systemml/blob/665d037e/src/main/java/org/apache/sysml/runtime/matrix/data/LibMatrixMult.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/matrix/data/LibMatrixMult.java b/src/main/java/org/apache/sysml/runtime/matrix/data/LibMatrixMult.java
index 1eef244..a98cd79 100644
--- a/src/main/java/org/apache/sysml/runtime/matrix/data/LibMatrixMult.java
+++ b/src/main/java/org/apache/sysml/runtime/matrix/data/LibMatrixMult.java
@@ -36,6 +36,7 @@ import org.apache.sysml.lops.WeightedSigmoid.WSigmoidType;
import org.apache.sysml.lops.WeightedSquaredLoss.WeightsType;
import org.apache.sysml.lops.WeightedUnaryMM.WUMMType;
import org.apache.sysml.runtime.DMLRuntimeException;
+import org.apache.sysml.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
import org.apache.sysml.runtime.functionobjects.SwapIndex;
import org.apache.sysml.runtime.functionobjects.ValueFunction;
import org.apache.sysml.runtime.matrix.operators.ReorgOperator;
@@ -56,7 +57,8 @@ public class LibMatrixMult
//internal configuration
private static final boolean LOW_LEVEL_OPTIMIZATION = true;
private static final long MEM_OVERHEAD_THRESHOLD = 2L*1024*1024; //MAX 2 MB
- private static final long PAR_MINFLOP_THRESHOLD = 2L*1024*1024; //MIN 2 MFLOP
+ private static final long PAR_MINFLOP_THRESHOLD1 = 2L*1024*1024; //MIN 2 MFLOP
+ private static final long PAR_MINFLOP_THRESHOLD2 = 128L*1024; //MIN 2 MFLOP
private static final int L2_CACHESIZE = 256 *1024; //256KB (common size)
private LibMatrixMult() {
@@ -164,11 +166,8 @@ public class LibMatrixMult
return;
}
- //check too high additional vector-matrix memory requirements (fallback to sequential)
- //check too small workload in terms of flops (fallback to sequential too)
- if( m1.rlen == 1 && (8L * m2.clen * k > MEM_OVERHEAD_THRESHOLD || !LOW_LEVEL_OPTIMIZATION || m2.clen==1 || m1.isUltraSparse() || m2.isUltraSparse())
- || 2L * m1.rlen * m1.clen * m2.clen < PAR_MINFLOP_THRESHOLD )
- {
+ //check too small workload and fallback to sequential if needed
+ if( !satisfiesMultiThreadingConstraints(m1, m2, m1.rlen==1, true, 2, k) ) {
matrixMult(m1, m2, ret);
return;
}
@@ -286,9 +285,8 @@ public class LibMatrixMult
return;
}
- //check too high additional memory requirements (fallback to sequential)
- //check too small workload in terms of flops (fallback to sequential too)
- if( !checkParColumnAgg(mX, k, true) ) {
+ //check temporary memory and too small workload for multi-threading
+ if( !satisfiesMultiThreadingConstraints(mX, true, true, mX.sparse?2:4, k) ) {
matrixMultChain(mX, mV, mW, ret, ct);
return;
}
@@ -307,12 +305,13 @@ public class LibMatrixMult
ArrayList<MatrixMultChainTask> tasks = new ArrayList<>();
for( int i=0, lb=0; i<blklens.size(); lb+=blklens.get(i), i++ )
tasks.add(new MatrixMultChainTask(mX, mV, mW, ct, lb, lb+blklens.get(i)));
- //execute tasks
List<Future<double[]>> taskret = pool.invokeAll(tasks);
pool.shutdown();
- //aggregate partial results
- for( Future<double[]> task : taskret )
- vectAdd(task.get(), ret.getDenseBlockValues(), 0, 0, mX.clen);
+ //aggregate partial results and error handling
+ double[][] a = new double[taskret.size()][];
+ for(int i=0; i<taskret.size(); i++)
+ a[i] = taskret.get(i).get();
+ vectAddAll(a, ret.getDenseBlockValues(), 0, 0, mX.clen);
}
catch(Exception ex) {
throw new DMLRuntimeException(ex);
@@ -360,12 +359,8 @@ public class LibMatrixMult
return;
}
- //check no parallelization benefit (fallback to sequential)
- //check too small workload in terms of flops (fallback to sequential too)
- if( ret.rlen == 1 || k <= 1
- || leftTranspose && 1L * m1.rlen * m1.clen * m1.clen < PAR_MINFLOP_THRESHOLD
- || !leftTranspose && 1L * m1.clen * m1.rlen * m1.rlen < PAR_MINFLOP_THRESHOLD)
- {
+ //check too small workload and fallback to sequential if necessary
+ if( !satisfiesMultiThreadingConstraintsTSMM(m1, leftTranspose, 1, k) ) {
matrixMultTransposeSelf(m1, ret, leftTranspose);
return;
}
@@ -3354,6 +3349,16 @@ public class LibMatrixMult
}
}
+ private static void vectAddAll(double[][] a, double[] c, int ai, int ci, final int len) {
+ int bi = a.length % 4;
+ //process stride for remaining blocks
+ for(int i=0; i<bi; i++)
+ vectAdd(a[i], c, ai, ci, len);
+ //process stride in 4 blocks at a time
+ for(int i=bi; i<a.length; i+=4)
+ vectAdd4(a[i], a[i+1], a[i+2], a[i+3], c, ai, ci, len);
+ }
+
public static void vectAddInPlace(double aval, double[] c, final int ci, final int len) {
final int bn = len%8;
//rest, not aligned to 8-blocks
@@ -3609,11 +3614,6 @@ public class LibMatrixMult
&& m2clen < 64 && (!inclCacheSize || 8*m2rlen*m2clen < L2_CACHESIZE);
}
- public static boolean checkParColumnAgg(MatrixBlock m1, int k, boolean inclFLOPs) {
- return (8L * m1.clen * k <= MEM_OVERHEAD_THRESHOLD
- && (!inclFLOPs || 4L * m1.rlen * m1.clen / (m1.sparse?2:1) >= PAR_MINFLOP_THRESHOLD));
- }
-
private static boolean checkParMatrixMultRightInputRows( MatrixBlock m1, MatrixBlock m2, int k ) {
//parallelize over rows in rhs matrix if number of rows in lhs/output is very small
return (m1.rlen==1 && LOW_LEVEL_OPTIMIZATION && m2.clen>1 && !(m1.isUltraSparse()||m2.isUltraSparse()))
@@ -3629,6 +3629,34 @@ public class LibMatrixMult
&& 8*m1.rlen*m1.clen < 256*1024 ); //lhs fits in L2 cache
}
+ public static boolean satisfiesMultiThreadingConstraints(MatrixBlock m1, int k) {
+ return satisfiesMultiThreadingConstraints(m1, true, false, -1, k);
+ }
+
+ public static boolean satisfiesMultiThreadingConstraints(MatrixBlock m1, boolean checkMem, boolean checkFLOPs, long FPfactor, int k) {
+ boolean sharedTP = (InfrastructureAnalyzer.getLocalParallelism() == k);
+ return k > 1 && LOW_LEVEL_OPTIMIZATION
+ && (!checkMem || 8L * m1.clen * k < MEM_OVERHEAD_THRESHOLD)
+ && (!checkFLOPs || FPfactor * m1.rlen * m1.clen >
+ (sharedTP ? PAR_MINFLOP_THRESHOLD2 : PAR_MINFLOP_THRESHOLD1));
+ }
+
+ public static boolean satisfiesMultiThreadingConstraints(MatrixBlock m1, MatrixBlock m2, boolean checkMem, boolean checkFLOPs, long FPfactor, int k) {
+ boolean sharedTP = (InfrastructureAnalyzer.getLocalParallelism() == k);
+ return k > 1 && LOW_LEVEL_OPTIMIZATION
+ && (!checkMem || 8L * m2.clen * k < MEM_OVERHEAD_THRESHOLD)
+ && (!checkFLOPs || FPfactor * m1.rlen * m1.clen * m2.clen >
+ (sharedTP ? PAR_MINFLOP_THRESHOLD2 : PAR_MINFLOP_THRESHOLD1));
+ }
+
+ private static boolean satisfiesMultiThreadingConstraintsTSMM(MatrixBlock m1, boolean leftTranspose, long FPfactor, int k) {
+ boolean sharedTP = (InfrastructureAnalyzer.getLocalParallelism() == k);
+ double threshold = sharedTP ? PAR_MINFLOP_THRESHOLD2 : PAR_MINFLOP_THRESHOLD1;
+ return k > 1 && LOW_LEVEL_OPTIMIZATION && (leftTranspose?m1.clen:m1.rlen)!=1
+ && ((leftTranspose && FPfactor * m1.rlen * m1.clen * m1.clen > threshold)
+ ||(!leftTranspose && FPfactor * m1.clen * m1.rlen * m1.rlen > threshold));
+ }
+
public static boolean isUltraSparseMatrixMult(MatrixBlock m1, MatrixBlock m2) {
//note: ultra-sparse matrix mult implies also sparse outputs, hence we need
//to be conservative an cannot use this for all ultra-sparse matrices.
[2/2] systemml git commit: [SYSTEMML-2217] Fix OOM read sparse
textcell matrix w/ unknown nnz
Posted by mb...@apache.org.
[SYSTEMML-2217] Fix OOM read sparse textcell matrix w/ unknown nnz
In case of forced single-node execution, the read of ultra-sparse
textcell matrices with unknown nnzs can cause unnecessary out-of-memory
errors because we always allocate the output in dense format for
unknowns. This patch fixes this robustness issue by obtaining a rough
estimate of the number of non-zeros based on the ratio of filesize on
hdfs to expected size in dense format.
Project: http://git-wip-us.apache.org/repos/asf/systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/systemml/commit/094f5551
Tree: http://git-wip-us.apache.org/repos/asf/systemml/tree/094f5551
Diff: http://git-wip-us.apache.org/repos/asf/systemml/diff/094f5551
Branch: refs/heads/master
Commit: 094f5551dd08f1d1747d70bbf23cc246d503d95e
Parents: 665d037
Author: Matthias Boehm <mb...@gmail.com>
Authored: Thu Mar 29 21:34:46 2018 -0700
Committer: Matthias Boehm <mb...@gmail.com>
Committed: Thu Mar 29 21:34:46 2018 -0700
----------------------------------------------------------------------
.../runtime/controlprogram/caching/MatrixObject.java | 3 ++-
.../java/org/apache/sysml/runtime/io/ReaderTextCell.java | 11 +++++++----
.../apache/sysml/runtime/io/ReaderTextCellParallel.java | 2 ++
.../org/apache/sysml/runtime/matrix/data/InputInfo.java | 9 +++++++--
.../org/apache/sysml/runtime/util/MapReduceTool.java | 8 ++++++++
5 files changed, 26 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/systemml/blob/094f5551/src/main/java/org/apache/sysml/runtime/controlprogram/caching/MatrixObject.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/MatrixObject.java b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/MatrixObject.java
index f258ea8..de49222 100644
--- a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/MatrixObject.java
+++ b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/MatrixObject.java
@@ -421,7 +421,8 @@ public class MatrixObject extends CacheableData<MatrixBlock>
}
//read matrix and maintain meta data
- double sparsity = (mc.getNonZeros() >= 0 ? ((double)mc.getNonZeros())/(mc.getRows()*mc.getCols()) : 1.0d);
+ double sparsity = (mc.getNonZeros() < 0) ? (iimd.getInputInfo().isTextIJV()?-1:1) :
+ OptimizerUtils.getSparsity(mc.getNonZeros(),mc.getRows(),mc.getCols());
MatrixBlock newData = DataConverter.readMatrixFromHDFS(fname, iimd.getInputInfo(), rlen, clen,
mc.getRowsPerBlock(), mc.getColsPerBlock(), sparsity, getFileFormatProperties());
setHDFSFileExists(true);
http://git-wip-us.apache.org/repos/asf/systemml/blob/094f5551/src/main/java/org/apache/sysml/runtime/io/ReaderTextCell.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/io/ReaderTextCell.java b/src/main/java/org/apache/sysml/runtime/io/ReaderTextCell.java
index 918ab6b..1fa8940 100644
--- a/src/main/java/org/apache/sysml/runtime/io/ReaderTextCell.java
+++ b/src/main/java/org/apache/sysml/runtime/io/ReaderTextCell.java
@@ -41,6 +41,7 @@ import org.apache.sysml.runtime.matrix.data.DenseBlock;
import org.apache.sysml.runtime.matrix.data.InputInfo;
import org.apache.sysml.runtime.matrix.data.MatrixBlock;
import org.apache.sysml.runtime.util.FastStringTokenizer;
+import org.apache.sysml.runtime.util.MapReduceTool;
public class ReaderTextCell extends MatrixReader
{
@@ -55,14 +56,16 @@ public class ReaderTextCell extends MatrixReader
public MatrixBlock readMatrixFromHDFS(String fname, long rlen, long clen, int brlen, int bclen, long estnnz)
throws IOException, DMLRuntimeException
{
- //allocate output matrix block
- MatrixBlock ret = createOutputMatrixBlock(rlen, clen, (int)rlen, (int)clen, estnnz, true, false);
-
//prepare file access
- JobConf job = new JobConf(ConfigurationManager.getCachedJobConf());
+ JobConf job = new JobConf(ConfigurationManager.getCachedJobConf());
Path path = new Path( fname );
FileSystem fs = IOUtilFunctions.getFileSystem(path, job);
+ //allocate output matrix block
+ if( estnnz < 0 )
+ estnnz = MapReduceTool.estimateNnzBasedOnFileSize(path, rlen, clen, brlen, bclen, 3);
+ MatrixBlock ret = createOutputMatrixBlock(rlen, clen, (int)rlen, (int)clen, estnnz, true, false);
+
//check existence and non-empty file
checkValidInputFile(fs, path);
http://git-wip-us.apache.org/repos/asf/systemml/blob/094f5551/src/main/java/org/apache/sysml/runtime/io/ReaderTextCellParallel.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/io/ReaderTextCellParallel.java b/src/main/java/org/apache/sysml/runtime/io/ReaderTextCellParallel.java
index 040b43c..72e0cde 100644
--- a/src/main/java/org/apache/sysml/runtime/io/ReaderTextCellParallel.java
+++ b/src/main/java/org/apache/sysml/runtime/io/ReaderTextCellParallel.java
@@ -91,6 +91,8 @@ public class ReaderTextCellParallel extends MatrixReader
checkValidInputFile(fs, path);
//allocate output matrix block
+ if( estnnz < 0 )
+ estnnz = MapReduceTool.estimateNnzBasedOnFileSize(path, rlen, clen, brlen, bclen, 3);
MatrixBlock ret = createOutputMatrixBlock(rlen, clen, (int)rlen, (int)clen, estnnz, true, false);
//core read
http://git-wip-us.apache.org/repos/asf/systemml/blob/094f5551/src/main/java/org/apache/sysml/runtime/matrix/data/InputInfo.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/matrix/data/InputInfo.java b/src/main/java/org/apache/sysml/runtime/matrix/data/InputInfo.java
index 2bbad62..02df7d1 100644
--- a/src/main/java/org/apache/sysml/runtime/matrix/data/InputInfo.java
+++ b/src/main/java/org/apache/sysml/runtime/matrix/data/InputInfo.java
@@ -54,10 +54,15 @@ public class InputInfo implements Serializable
inputValueClass=valueCls;
}
- public void setMetaData(MetaData md)
- {
+ public void setMetaData(MetaData md) {
metadata=md;
}
+
+ public boolean isTextIJV() {
+ return this == InputInfo.TextCellInputInfo
+ || this == InputInfo.MatrixMarketInputInfo;
+ }
+
public static final InputInfo TextCellInputInfo=new InputInfo(TextInputFormat.class,
LongWritable.class, Text.class);
public static final InputInfo MatrixMarketInputInfo = new InputInfo (TextInputFormat.class,
http://git-wip-us.apache.org/repos/asf/systemml/blob/094f5551/src/main/java/org/apache/sysml/runtime/util/MapReduceTool.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/util/MapReduceTool.java b/src/main/java/org/apache/sysml/runtime/util/MapReduceTool.java
index 52ad388..c6d9604 100644
--- a/src/main/java/org/apache/sysml/runtime/util/MapReduceTool.java
+++ b/src/main/java/org/apache/sysml/runtime/util/MapReduceTool.java
@@ -43,6 +43,7 @@ import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.sysml.conf.ConfigurationManager;
+import org.apache.sysml.hops.OptimizerUtils;
import org.apache.sysml.parser.DataExpression;
import org.apache.sysml.parser.Expression.DataType;
import org.apache.sysml.parser.Expression.ValueType;
@@ -238,6 +239,13 @@ public class MapReduceTool
FileUtil.copy(fs, originalPath, fs, newPath, deleteSource, overwrite, job);
}
}
+
+ public static long estimateNnzBasedOnFileSize(Path path,
+ long rlen, long clen, int brlen, int bclen, double factor) throws IOException
+ {
+ return (long) Math.min(rlen*clen, rlen*clen*(getFilesizeOnHDFS(path)/factor/
+ OptimizerUtils.estimatePartitionedSizeExactSparsity(rlen, clen, brlen, bclen, 1.0)));
+ }
/**
* Returns the size of a file or directory on hdfs in bytes.