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 2017/03/18 21:33:47 UTC
[1/2] incubator-systemml git commit: [SYSTEMML-1315] Robustness spark
mapmm (handle large output partitions)
Repository: incubator-systemml
Updated Branches:
refs/heads/master d127dfa2d -> 30f72e83f
[SYSTEMML-1315] Robustness spark mapmm (handle large output partitions)
Our spark mapmm (broadcast-based matrix multiply) instruction already
repartitions the input RDD for shapes that potentially create large
outputs in order to control the size of output partitions and with that
avoid failures due to Spark's 2GB limitation per partition. However,
with out perftest-stratstats script we encountered special cases of
dense-sparse matrix multiplications, where this was not enough and we
ran unnecessarily into OOMs. In detail, the sparse matrix was the
broadcast (because it was smaller) but the dense input had a small
number of blocks which limited the maximum number of partitions.
This patch extends the mapmm instruction to consider these cases and
potentially flip broadcast and rdd at runtime if this is beneficial for
controlling the output sizes.
Project: http://git-wip-us.apache.org/repos/asf/incubator-systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-systemml/commit/073b7024
Tree: http://git-wip-us.apache.org/repos/asf/incubator-systemml/tree/073b7024
Diff: http://git-wip-us.apache.org/repos/asf/incubator-systemml/diff/073b7024
Branch: refs/heads/master
Commit: 073b70241547a840518a785054cd319711d2d2a3
Parents: d127dfa
Author: Matthias Boehm <mb...@gmail.com>
Authored: Fri Mar 17 17:38:50 2017 -0700
Committer: Matthias Boehm <mb...@gmail.com>
Committed: Sat Mar 18 14:34:39 2017 -0700
----------------------------------------------------------------------
.../java/org/apache/sysml/lops/MapMult.java | 12 +-
.../mr/AggregateBinaryInstruction.java | 8 +-
.../instructions/spark/MapmmSPInstruction.java | 113 +++++++++++--------
3 files changed, 80 insertions(+), 53 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/073b7024/src/main/java/org/apache/sysml/lops/MapMult.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/lops/MapMult.java b/src/main/java/org/apache/sysml/lops/MapMult.java
index 1cb8e46..6597136 100644
--- a/src/main/java/org/apache/sysml/lops/MapMult.java
+++ b/src/main/java/org/apache/sysml/lops/MapMult.java
@@ -38,9 +38,19 @@ public class MapMult extends Lop
LEFT,
LEFT_PART;
- public boolean isRightCache(){
+ public boolean isRight() {
return (this == RIGHT || this == RIGHT_PART);
}
+
+ public CacheType getFlipped() {
+ switch( this ) {
+ case RIGHT: return LEFT;
+ case RIGHT_PART: return LEFT_PART;
+ case LEFT: return RIGHT;
+ case LEFT_PART: return RIGHT_PART;
+ default: return null;
+ }
+ }
}
private CacheType _cacheType = null;
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/073b7024/src/main/java/org/apache/sysml/runtime/instructions/mr/AggregateBinaryInstruction.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/instructions/mr/AggregateBinaryInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/mr/AggregateBinaryInstruction.java
index 63f72bd..e9a023a 100644
--- a/src/main/java/org/apache/sysml/runtime/instructions/mr/AggregateBinaryInstruction.java
+++ b/src/main/java/org/apache/sysml/runtime/instructions/mr/AggregateBinaryInstruction.java
@@ -119,7 +119,7 @@ public class AggregateBinaryInstruction extends BinaryMRInstructionBase implemen
@Override //IDistributedCacheConsumer
public boolean isDistCacheOnlyIndex( String inst, byte index )
{
- return _cacheType.isRightCache() ?
+ return _cacheType.isRight() ?
(index==input2 && index!=input1) :
(index==input1 && index!=input2);
}
@@ -127,7 +127,7 @@ public class AggregateBinaryInstruction extends BinaryMRInstructionBase implemen
@Override //IDistributedCacheConsumer
public void addDistCacheIndex( String inst, ArrayList<Byte> indexes )
{
- indexes.add( _cacheType.isRightCache() ? input2 : input1 );
+ indexes.add( _cacheType.isRight() ? input2 : input1 );
}
@Override
@@ -142,7 +142,7 @@ public class AggregateBinaryInstruction extends BinaryMRInstructionBase implemen
if ( _opcode.equals(MapMult.OPCODE) )
{
//check empty inputs (data for different instructions)
- if( _cacheType.isRightCache() ? in1==null : in2==null )
+ if( _cacheType.isRight() ? in1==null : in2==null )
return;
// one of the input is from distributed cache.
@@ -190,7 +190,7 @@ public class AggregateBinaryInstruction extends BinaryMRInstructionBase implemen
{
boolean removeOutput = true;
- if( _cacheType.isRightCache() )
+ if( _cacheType.isRight() )
{
DistributedCacheInput dcInput = MRBaseForCommonInstructions.dcValues.get(input2);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/073b7024/src/main/java/org/apache/sysml/runtime/instructions/spark/MapmmSPInstruction.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/instructions/spark/MapmmSPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/spark/MapmmSPInstruction.java
index 5baffb0..c1fdea6 100644
--- a/src/main/java/org/apache/sysml/runtime/instructions/spark/MapmmSPInstruction.java
+++ b/src/main/java/org/apache/sysml/runtime/instructions/spark/MapmmSPInstruction.java
@@ -55,19 +55,14 @@ import org.apache.sysml.runtime.matrix.operators.AggregateBinaryOperator;
import org.apache.sysml.runtime.matrix.operators.AggregateOperator;
import org.apache.sysml.runtime.matrix.operators.Operator;
-/**
- * TODO: we need to reason about multiple broadcast variables for chains of mapmults (sum of operations until cleanup)
- *
- */
public class MapmmSPInstruction extends BinarySPInstruction
-{
-
+{
private CacheType _type = null;
private boolean _outputEmpty = true;
private SparkAggType _aggtype;
public MapmmSPInstruction(Operator op, CPOperand in1, CPOperand in2, CPOperand out, CacheType type,
- boolean outputEmpty, SparkAggType aggtype, String opcode, String istr )
+ boolean outputEmpty, SparkAggType aggtype, String opcode, String istr )
{
super(op, in1, in2, out, opcode, istr);
_sptype = SPINSTRUCTION_TYPE.MAPMM;
@@ -83,23 +78,19 @@ public class MapmmSPInstruction extends BinarySPInstruction
String parts[] = InstructionUtils.getInstructionPartsWithValueType(str);
String opcode = parts[0];
- if ( opcode.equalsIgnoreCase(MapMult.OPCODE))
- {
- CPOperand in1 = new CPOperand(parts[1]);
- CPOperand in2 = new CPOperand(parts[2]);
- CPOperand out = new CPOperand(parts[3]);
- CacheType type = CacheType.valueOf(parts[4]);
- boolean outputEmpty = Boolean.parseBoolean(parts[5]);
- SparkAggType aggtype = SparkAggType.valueOf(parts[6]);
-
- AggregateOperator agg = new AggregateOperator(0, Plus.getPlusFnObject());
- AggregateBinaryOperator aggbin = new AggregateBinaryOperator(Multiply.getMultiplyFnObject(), agg);
- return new MapmmSPInstruction(aggbin, in1, in2, out, type, outputEmpty, aggtype, opcode, str);
- }
- else {
+ if(!opcode.equalsIgnoreCase(MapMult.OPCODE))
throw new DMLRuntimeException("MapmmSPInstruction.parseInstruction():: Unknown opcode " + opcode);
- }
+
+ CPOperand in1 = new CPOperand(parts[1]);
+ CPOperand in2 = new CPOperand(parts[2]);
+ CPOperand out = new CPOperand(parts[3]);
+ CacheType type = CacheType.valueOf(parts[4]);
+ boolean outputEmpty = Boolean.parseBoolean(parts[5]);
+ SparkAggType aggtype = SparkAggType.valueOf(parts[6]);
+ AggregateOperator agg = new AggregateOperator(0, Plus.getPlusFnObject());
+ AggregateBinaryOperator aggbin = new AggregateBinaryOperator(Multiply.getMultiplyFnObject(), agg);
+ return new MapmmSPInstruction(aggbin, in1, in2, out, type, outputEmpty, aggtype, opcode, str);
}
@Override
@@ -108,14 +99,37 @@ public class MapmmSPInstruction extends BinarySPInstruction
{
SparkExecutionContext sec = (SparkExecutionContext)ec;
- String rddVar = (_type==CacheType.LEFT) ? input2.getName() : input1.getName();
- String bcastVar = (_type==CacheType.LEFT) ? input1.getName() : input2.getName();
+ CacheType type = _type;
+ String rddVar = type.isRight() ? input1.getName() : input2.getName();
+ String bcastVar = type.isRight() ? input2.getName() : input1.getName();
MatrixCharacteristics mcRdd = sec.getMatrixCharacteristics(rddVar);
MatrixCharacteristics mcBc = sec.getMatrixCharacteristics(bcastVar);
+ //get input rdd
+ JavaPairRDD<MatrixIndexes,MatrixBlock> in1 = sec.getBinaryBlockRDDHandleForVariable(rddVar);
+
+ //investigate if a repartitioning - including a potential flip of broadcast and rdd
+ //inputs - is required to ensure moderately sized output partitions (2GB limitation)
+ if( requiresFlatMapFunction(type, mcBc) &&
+ requiresRepartitioning(type, mcRdd, mcBc, in1.getNumPartitions()) )
+ {
+ int numParts = getNumRepartitioning(type, mcRdd, mcBc);
+ int numParts2 = getNumRepartitioning(type.getFlipped(), mcBc, mcRdd);
+ if( numParts2 > numParts ) { //flip required
+ type = type.getFlipped();
+ rddVar = type.isRight() ? input1.getName() : input2.getName();
+ bcastVar = type.isRight() ? input2.getName() : input1.getName();
+ mcRdd = sec.getMatrixCharacteristics(rddVar);
+ mcBc = sec.getMatrixCharacteristics(bcastVar);
+ in1 = sec.getBinaryBlockRDDHandleForVariable(rddVar);
+ LOG.warn("Mapmm: Switching rdd ('"+bcastVar+"') and broadcast ('"+rddVar+"') inputs "
+ + "for repartitioning because this allows better control of output partition "
+ + "sizes ("+numParts+" < "+numParts2+").");
+ }
+ }
+
//get inputs
- JavaPairRDD<MatrixIndexes,MatrixBlock> in1 = sec.getBinaryBlockRDDHandleForVariable( rddVar );
- PartitionedBroadcast<MatrixBlock> in2 = sec.getBroadcastForVariable( bcastVar );
+ PartitionedBroadcast<MatrixBlock> in2 = sec.getBroadcastForVariable(bcastVar);
//empty input block filter
if( !_outputEmpty )
@@ -124,7 +138,7 @@ public class MapmmSPInstruction extends BinarySPInstruction
//execute mapmm and aggregation if necessary and put output into symbol table
if( _aggtype == SparkAggType.SINGLE_BLOCK )
{
- JavaRDD<MatrixBlock> out = in1.map(new RDDMapMMFunction2(_type, in2));
+ JavaRDD<MatrixBlock> out = in1.map(new RDDMapMMFunction2(type, in2));
MatrixBlock out2 = RDDAggregateUtils.sumStable(out);
//put output block into symbol table (no lineage because single block)
@@ -134,15 +148,19 @@ public class MapmmSPInstruction extends BinarySPInstruction
else //MULTI_BLOCK or NONE
{
JavaPairRDD<MatrixIndexes,MatrixBlock> out = null;
- if( requiresFlatMapFunction(_type, mcBc) ) {
- if( requiresRepartitioning(_type, mcRdd, mcBc, in1.getNumPartitions()) )
- in1 = in1.repartition(getNumRepartitioning(_type, mcRdd, mcBc, in1.getNumPartitions()));
- out = in1.flatMapToPair( new RDDFlatMapMMFunction(_type, in2) );
+ if( requiresFlatMapFunction(type, mcBc) ) {
+ if( requiresRepartitioning(type, mcRdd, mcBc, in1.getNumPartitions()) ) {
+ int numParts = getNumRepartitioning(type, mcRdd, mcBc);
+ LOG.warn("Mapmm: Repartition input rdd '"+rddVar+"' from "+in1.getNumPartitions()+" to "
+ +numParts+" partitions to satisfy size restrictions of output partitions.");
+ in1 = in1.repartition(numParts);
+ }
+ out = in1.flatMapToPair( new RDDFlatMapMMFunction(type, in2) );
}
- else if( preservesPartitioning(mcRdd, _type) )
- out = in1.mapPartitionsToPair(new RDDMapMMPartitionFunction(_type, in2), true);
+ else if( preservesPartitioning(mcRdd, type) )
+ out = in1.mapPartitionsToPair(new RDDMapMMPartitionFunction(type, in2), true);
else
- out = in1.mapToPair( new RDDMapMMFunction(_type, in2) );
+ out = in1.mapToPair( new RDDMapMMFunction(type, in2) );
//empty output block filter
if( !_outputEmpty )
@@ -216,10 +234,9 @@ public class MapmmSPInstruction extends BinarySPInstruction
* @param type cache type
* @param mcRdd rdd matrix characteristics
* @param mcBc ?
- * @param numPartitions number of partitions
* @return number of target partitions for repartitioning
*/
- private static int getNumRepartitioning( CacheType type, MatrixCharacteristics mcRdd, MatrixCharacteristics mcBc, int numPartitions ) {
+ private static int getNumRepartitioning( CacheType type, MatrixCharacteristics mcRdd, MatrixCharacteristics mcBc ) {
boolean isLeft = (type == CacheType.LEFT);
long sizeOutput = (OptimizerUtils.estimatePartitionedSizeExactSparsity(isLeft?mcBc.getRows():mcRdd.getRows(),
isLeft?mcRdd.getCols():mcBc.getCols(), isLeft?mcBc.getRowsPerBlock():mcRdd.getRowsPerBlock(),
@@ -232,9 +249,9 @@ public class MapmmSPInstruction extends BinarySPInstruction
{
private static final long serialVersionUID = 8197406787010296291L;
- private CacheType _type = null;
- private AggregateBinaryOperator _op = null;
- private PartitionedBroadcast<MatrixBlock> _pbc = null;
+ private final CacheType _type;
+ private final AggregateBinaryOperator _op;
+ private final PartitionedBroadcast<MatrixBlock> _pbc;
public RDDMapMMFunction( CacheType type, PartitionedBroadcast<MatrixBlock> binput )
{
@@ -287,9 +304,9 @@ public class MapmmSPInstruction extends BinarySPInstruction
{
private static final long serialVersionUID = -2753453898072910182L;
- private CacheType _type = null;
- private AggregateBinaryOperator _op = null;
- private PartitionedBroadcast<MatrixBlock> _pbc = null;
+ private final CacheType _type;
+ private final AggregateBinaryOperator _op;
+ private final PartitionedBroadcast<MatrixBlock> _pbc;
public RDDMapMMFunction2( CacheType type, PartitionedBroadcast<MatrixBlock> binput )
{
@@ -333,9 +350,9 @@ public class MapmmSPInstruction extends BinarySPInstruction
{
private static final long serialVersionUID = 1886318890063064287L;
- private CacheType _type = null;
- private AggregateBinaryOperator _op = null;
- private PartitionedBroadcast<MatrixBlock> _pbc = null;
+ private final CacheType _type;
+ private final AggregateBinaryOperator _op;
+ private final PartitionedBroadcast<MatrixBlock> _pbc;
public RDDMapMMPartitionFunction( CacheType type, PartitionedBroadcast<MatrixBlock> binput )
{
@@ -399,9 +416,9 @@ public class MapmmSPInstruction extends BinarySPInstruction
{
private static final long serialVersionUID = -6076256569118957281L;
- private CacheType _type = null;
- private AggregateBinaryOperator _op = null;
- private PartitionedBroadcast<MatrixBlock> _pbc = null;
+ private final CacheType _type;
+ private final AggregateBinaryOperator _op;
+ private final PartitionedBroadcast<MatrixBlock> _pbc;
public RDDFlatMapMMFunction( CacheType type, PartitionedBroadcast<MatrixBlock> binput )
{
[2/2] incubator-systemml git commit: [SYSTEMML-1420] Extended code
generator (sparsity-exploiting celltmpl)
Posted by mb...@apache.org.
[SYSTEMML-1420] Extended code generator (sparsity-exploiting celltmpl)
Project: http://git-wip-us.apache.org/repos/asf/incubator-systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-systemml/commit/30f72e83
Tree: http://git-wip-us.apache.org/repos/asf/incubator-systemml/tree/30f72e83
Diff: http://git-wip-us.apache.org/repos/asf/incubator-systemml/diff/30f72e83
Branch: refs/heads/master
Commit: 30f72e83fd334042eec1be5c42588e127a9f3ee0
Parents: 073b702
Author: Matthias Boehm <mb...@gmail.com>
Authored: Fri Mar 17 20:05:40 2017 -0700
Committer: Matthias Boehm <mb...@gmail.com>
Committed: Sat Mar 18 14:34:41 2017 -0700
----------------------------------------------------------------------
.../sysml/hops/codegen/cplan/CNodeCell.java | 25 +++-
.../sysml/hops/codegen/template/CellTpl.java | 2 +
.../sysml/runtime/codegen/SpoofCellwise.java | 127 ++++++++++---------
.../functions/codegen/CellwiseTmplTest.java | 42 ++++--
.../scripts/functions/codegen/cellwisetmpl8.R | 33 +++++
.../scripts/functions/codegen/cellwisetmpl8.dml | 29 +++++
6 files changed, 181 insertions(+), 77 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/30f72e83/src/main/java/org/apache/sysml/hops/codegen/cplan/CNodeCell.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/hops/codegen/cplan/CNodeCell.java b/src/main/java/org/apache/sysml/hops/codegen/cplan/CNodeCell.java
index 8122078..caf7b6a 100644
--- a/src/main/java/org/apache/sysml/hops/codegen/cplan/CNodeCell.java
+++ b/src/main/java/org/apache/sysml/hops/codegen/cplan/CNodeCell.java
@@ -39,15 +39,16 @@ public class CNodeCell extends CNodeTpl
+ "\n"
+ "public final class %TMP% extends SpoofCellwise {\n"
+ " public %TMP%() {\n"
- + " _type = CellType.%TYPE%;\n"
+ + " super(CellType.%TYPE%, %SPARSE_SAFE%);\n"
+ " }\n"
- + " protected double genexec( double a, double[][] b, double[] scalars, int n, int m, int rowIndex, int colIndex) { \n"
+ + " protected double genexec( double a, double[][] b, double[] scalars, int m, int n, int rowIndex, int colIndex) { \n"
+ "%BODY_dense%"
+ " return %OUT%;\n"
+ " } \n"
+ "}";
private CellType _type = null;
+ private boolean _sparseSafe = false;
private boolean _requiresCastdtm = false;
private boolean _multipleConsumers = false;
@@ -72,6 +73,14 @@ public class CNodeCell extends CNodeTpl
return _type;
}
+ public void setSparseSafe(boolean flag) {
+ _sparseSafe = flag;
+ }
+
+ public boolean isSparseSafe() {
+ return _sparseSafe;
+ }
+
public void setRequiresCastDtm(boolean flag) {
_requiresCastdtm = flag;
_hash = 0;
@@ -99,8 +108,9 @@ public class CNodeCell extends CNodeTpl
//return last TMP
tmp = tmp.replaceAll("%OUT%", getCurrentVarName());
- //replace aggregate information
- tmp = tmp.replaceAll("%TYPE%", getCellType().toString());
+ //replace meta data information
+ tmp = tmp.replaceAll("%TYPE%", getCellType().name());
+ tmp = tmp.replaceAll("%SPARSE_SAFE%", String.valueOf(isSparseSafe()));
return tmp;
}
@@ -136,9 +146,10 @@ public class CNodeCell extends CNodeTpl
if( _hash == 0 ) {
int h1 = super.hashCode();
int h2 = _type.hashCode();
- int h3 = Boolean.valueOf(_requiresCastdtm).hashCode();
+ int h3 = Boolean.valueOf(_sparseSafe).hashCode();
+ int h4 = Boolean.valueOf(_requiresCastdtm).hashCode();
//note: _multipleConsumers irrelevant for plan comparison
- _hash = Arrays.hashCode(new int[]{h1,h2,h3});
+ _hash = Arrays.hashCode(new int[]{h1,h2,h3,h4});
}
return _hash;
}
@@ -151,6 +162,7 @@ public class CNodeCell extends CNodeTpl
CNodeCell that = (CNodeCell)o;
return super.equals(that)
&& _type == that._type
+ && _sparseSafe == that._sparseSafe
&& _requiresCastdtm == that._requiresCastdtm
&& equalInputReferences(
_output, that._output, _inputs, that._inputs);
@@ -161,6 +173,7 @@ public class CNodeCell extends CNodeTpl
StringBuilder sb = new StringBuilder();
sb.append("SPOOF CELLWISE [type=");
sb.append(_type.name());
+ sb.append(", spafeSafe="+_sparseSafe);
sb.append(", castdtm="+_requiresCastdtm);
sb.append(", mc="+_multipleConsumers);
sb.append("]");
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/30f72e83/src/main/java/org/apache/sysml/hops/codegen/template/CellTpl.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/hops/codegen/template/CellTpl.java b/src/main/java/org/apache/sysml/hops/codegen/template/CellTpl.java
index 1aa1909..f5d11d1 100644
--- a/src/main/java/org/apache/sysml/hops/codegen/template/CellTpl.java
+++ b/src/main/java/org/apache/sysml/hops/codegen/template/CellTpl.java
@@ -112,6 +112,8 @@ public class CellTpl extends BaseTpl
CNode output = tmp.get(hop.getHopID());
CNodeCell tpl = new CNodeCell(inputs, output);
tpl.setCellType(TemplateUtils.getCellType(hop));
+ tpl.setSparseSafe((HopRewriteUtils.isBinary(hop, OpOp2.MULT) && hop.getInput().contains(sinHops.getFirst()))
+ || (HopRewriteUtils.isBinary(hop, OpOp2.DIV) && hop.getInput().get(0) == sinHops.getFirst()));
tpl.setRequiresCastDtm(hop instanceof AggBinaryOp);
// return cplan instance
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/30f72e83/src/main/java/org/apache/sysml/runtime/codegen/SpoofCellwise.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/codegen/SpoofCellwise.java b/src/main/java/org/apache/sysml/runtime/codegen/SpoofCellwise.java
index ea23e4d..f75aed6 100644
--- a/src/main/java/org/apache/sysml/runtime/codegen/SpoofCellwise.java
+++ b/src/main/java/org/apache/sysml/runtime/codegen/SpoofCellwise.java
@@ -48,16 +48,22 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
ROW_AGG,
}
- protected CellType _type = CellType.NO_AGG;
+ private final CellType _type;
+ private final boolean _sparseSafe;
- public SpoofCellwise() {
-
+ public SpoofCellwise(CellType type, boolean sparseSafe) {
+ _type = type;
+ _sparseSafe = sparseSafe;
}
public CellType getCellType() {
return _type;
}
+ public boolean isSparseSafe() {
+ return _sparseSafe;
+ }
+
@Override
public ScalarObject execute(ArrayList<MatrixBlock> inputs, ArrayList<ScalarObject> scalarObjects, int k)
throws DMLRuntimeException
@@ -73,15 +79,19 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
//input preparation
double[][] b = prepInputMatrices(inputs);
double[] scalars = prepInputScalars(scalarObjects);
-
final int m = inputs.get(0).getNumRows();
- final int n = inputs.get(0).getNumColumns();
+ final int n = inputs.get(0).getNumColumns();
+
+ //sparse safe check
+ boolean sparseSafe = isSparseSafe() || (b.length == 0
+ && genexec( 0, b, scalars, m, n, 0, 0 ) == 0);
+
double sum = 0;
if( k <= 1 ) //SINGLE-THREADED
{
sum = ( !inputs.get(0).isInSparseFormat() ) ?
- executeDenseAndAgg(inputs.get(0).getDenseBlock(), b, scalars, n, m, 0, m) :
- executeSparseAndAgg(inputs.get(0).getSparseBlock(), b, scalars, n, m, 0, m);
+ executeDenseAndAgg(inputs.get(0).getDenseBlock(), b, scalars, m, n, sparseSafe, 0, m) :
+ executeSparseAndAgg(inputs.get(0).getSparseBlock(), b, scalars, m, n, sparseSafe, 0, m);
}
else //MULTI-THREADED
{
@@ -91,7 +101,7 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
int nk = UtilFunctions.roundToNext(Math.min(8*k,m/32), k);
int blklen = (int)(Math.ceil((double)m/nk));
for( int i=0; i<nk & i*blklen<m; i++ )
- tasks.add(new ParAggTask(inputs.get(0), b, scalars, n, m,i*blklen, Math.min((i+1)*blklen, m)));
+ tasks.add(new ParAggTask(inputs.get(0), b, scalars, m, n, sparseSafe, i*blklen, Math.min((i+1)*blklen, m)));
//execute tasks
List<Future<Double>> taskret = pool.invokeAll(tasks);
pool.shutdown();
@@ -138,17 +148,19 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
//input preparation
double[][] b = prepInputMatrices(inputs);
double[] scalars = prepInputScalars(scalarObjects);
-
- //core sequential execute
final int m = inputs.get(0).getNumRows();
final int n = inputs.get(0).getNumColumns();
+ //sparse safe check
+ boolean sparseSafe = isSparseSafe() || (b.length == 0
+ && genexec( 0, b, scalars, m, n, 0, 0 ) == 0);
+
long lnnz = 0;
if( k <= 1 ) //SINGLE-THREADED
{
lnnz = (!inputs.get(0).isInSparseFormat()) ?
- executeDense(inputs.get(0).getDenseBlock(), b, scalars, c, n, m, 0, m) :
- executeSparse(inputs.get(0).getSparseBlock(), b, scalars, c, n, m, 0, m);
+ executeDense(inputs.get(0).getDenseBlock(), b, scalars, c, m, n, sparseSafe, 0, m) :
+ executeSparse(inputs.get(0).getSparseBlock(), b, scalars, c, m, n, sparseSafe, 0, m);
}
else //MULTI-THREADED
{
@@ -159,7 +171,7 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
int blklen = (int)(Math.ceil((double)m/nk));
for( int i=0; i<nk & i*blklen<m; i++ )
tasks.add(new ParExecTask(inputs.get(0), b, scalars, c,
- n, m, i*blklen, Math.min((i+1)*blklen, m)));
+ m, n, sparseSafe, i*blklen, Math.min((i+1)*blklen, m)));
//execute tasks
List<Future<Long>> taskret = pool.invokeAll(tasks);
pool.shutdown();
@@ -187,48 +199,50 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
* @param rl
* @param ru
*/
- private double executeDenseAndAgg(double[] a, double[][] b, double[] scalars, int n, int m, int rl, int ru)
+ private double executeDenseAndAgg(double[] a, double[][] b, double[] scalars, int m, int n, boolean sparseSafe, int rl, int ru)
{
KahanObject kbuff = new KahanObject(0, 0);
KahanPlus kplus = KahanPlus.getKahanPlusFnObject();
- if( a == null ) { //empty
+ if( a == null && !sparseSafe ) { //empty
//note: we can't determine sparse-safeness by executing the operator once
//as the output might change with different row indices
for( int i=rl; i<ru; i++ )
for( int j=0; j<n; j++ )
- kplus.execute2(kbuff, genexec( 0, b, scalars, n, m, i, j ));
+ kplus.execute2(kbuff, genexec( 0, b, scalars, m, n, i, j ));
}
- else { //general case
+ else if( a != null ) { //general case
for( int i=rl, ix=rl*n; i<ru; i++ )
for( int j=0; j<n; j++, ix++ )
- kplus.execute2(kbuff, genexec( a[ix], b, scalars, n, m, i, j ));
+ if( a[ix] != 0 || !sparseSafe)
+ kplus.execute2(kbuff, genexec( a[ix], b, scalars, m, n, i, j ));
}
return kbuff._sum;
}
- private long executeDense(double[] a, double[][] b, double[] scalars, double[] c, int n, int m, int rl, int ru)
+ private long executeDense(double[] a, double[][] b, double[] scalars, double[] c, int m, int n, boolean sparseSafe, int rl, int ru)
{
long lnnz = 0;
if( _type == CellType.NO_AGG )
{
- if( a == null ) { //empty
+ if( a == null && !sparseSafe ) { //empty
//note: we can't determine sparse-safeness by executing the operator once
//as the output might change with different row indices
for( int i=rl, ix=rl*n; i<ru; i++ )
for( int j=0; j<n; j++, ix++ ) {
- c[ix] = genexec( 0, b, scalars, n, m, i, j );
+ c[ix] = genexec( 0, b, scalars, m, n, i, j );
lnnz += (c[ix]!=0) ? 1 : 0;
}
}
- else { //general case
+ else if( a != null ) { //general case
for( int i=rl, ix=rl*n; i<ru; i++ )
- for( int j=0; j<n; j++, ix++ ) {
- c[ix] = genexec( a[ix], b, scalars, n, m, i, j);
- lnnz += (c[ix]!=0) ? 1 : 0;
- }
+ for( int j=0; j<n; j++, ix++ )
+ if( a[ix] != 0 || !sparseSafe) {
+ c[ix] = genexec( a[ix], b, scalars, m, n, i, j);
+ lnnz += (c[ix]!=0) ? 1 : 0;
+ }
}
}
else if( _type == CellType.ROW_AGG )
@@ -236,22 +250,23 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
KahanObject kbuff = new KahanObject(0, 0);
KahanPlus kplus = KahanPlus.getKahanPlusFnObject();
- if( a == null ) { //empty
+ if( a == null && !sparseSafe ) { //empty
//note: we can't determine sparse-safeness by executing the operator once
//as the output might change with different row indices
for( int i=rl; i<ru; i++ ) {
kbuff.set(0, 0);
for( int j=0; j<n; j++ )
- kplus.execute2(kbuff, genexec( 0, b, scalars, n, m, i, j ));
+ kplus.execute2(kbuff, genexec( 0, b, scalars, m, n, i, j ));
c[i] = kbuff._sum;
lnnz += (c[i]!=0) ? 1 : 0;
}
}
- else { //general case
+ else if( a != null ) { //general case
for( int i=rl, ix=rl*n; i<ru; i++ ) {
kbuff.set(0, 0);
for( int j=0; j<n; j++, ix++ )
- kplus.execute2(kbuff, genexec( a[ix], b, scalars, n, m, i, j ));
+ if( a[ix] != 0 || !sparseSafe)
+ kplus.execute2(kbuff, genexec( a[ix], b, scalars, m, n, i, j ));
c[i] = kbuff._sum;
lnnz += (c[i]!=0) ? 1 : 0;
}
@@ -261,15 +276,11 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
return lnnz;
}
- private double executeSparseAndAgg(SparseBlock sblock, double[][] b, double[] scalars, int n, int m, int rl, int ru)
+ private double executeSparseAndAgg(SparseBlock sblock, double[][] b, double[] scalars, int m, int n, boolean sparseSafe, int rl, int ru)
{
KahanObject kbuff = new KahanObject(0, 0);
KahanPlus kplus = KahanPlus.getKahanPlusFnObject();
- //sparse safe check
- boolean sparseSafe = (b.length == 0 //no sideways inputs
- && genexec( 0, b, scalars, n, m, 0, 0 ) == 0); //0 input results in 0
-
if( sparseSafe ) {
if( sblock != null ) {
for( int i=rl; i<ru; i++ )
@@ -278,7 +289,7 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
int alen = sblock.size(i);
double[] avals = sblock.values(i);
for( int j=apos; j<apos+alen; j++ ) {
- kplus.execute2( kbuff, genexec(avals[j], b, scalars, n, m, i, j));
+ kplus.execute2( kbuff, genexec(avals[j], b, scalars, m, n, i, j));
}
}
}
@@ -287,19 +298,15 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
for(int i=rl; i<ru; i++)
for(int j=0; j<n; j++) {
double valij = (sblock != null) ? sblock.get(i, j) : 0;
- kplus.execute2( kbuff, genexec(valij, b, scalars, n, m, i, j));
+ kplus.execute2( kbuff, genexec(valij, b, scalars, m, n, i, j));
}
}
return kbuff._sum;
}
- private long executeSparse(SparseBlock sblock, double[][] b, double[] scalars, double[] c, int n, int m, int rl, int ru)
+ private long executeSparse(SparseBlock sblock, double[][] b, double[] scalars, double[] c, int m, int n, boolean sparseSafe, int rl, int ru)
{
- //sparse safe check
- boolean sparseSafe = (b.length == 0 //no sideways inputs
- && genexec( 0, b, scalars, n, m, 0, 0 ) == 0); //0 input results in 0
-
long lnnz = 0;
if( _type == CellType.NO_AGG )
{
@@ -311,7 +318,7 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
int alen = sblock.size(i);
double[] avals = sblock.values(i);
for( int j=apos; j<apos+alen; j++ ) {
- double val = genexec(avals[j], b, scalars, n, m, i, j);
+ double val = genexec(avals[j], b, scalars, m, n, i, j);
c[i*n+sblock.indexes(i)[j]] = val;
lnnz += (val!=0) ? 1 : 0;
}
@@ -322,7 +329,7 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
for(int i=rl, cix=rl*n; i<ru; i++, cix+=n)
for(int j=0; j<n; j++) {
double valij = (sblock != null) ? sblock.get(i, j) : 0;
- c[cix+j] = genexec(valij, b, scalars, n, m, i, j);
+ c[cix+j] = genexec(valij, b, scalars, m, n, i, j);
lnnz += (c[cix+j]!=0) ? 1 : 0;
}
}
@@ -341,7 +348,7 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
int alen = sblock.size(i);
double[] avals = sblock.values(i);
for( int j=apos; j<apos+alen; j++ ) {
- kplus.execute2(kbuff, genexec(avals[j], b, scalars, n, m, i, j));
+ kplus.execute2(kbuff, genexec(avals[j], b, scalars, m, n, i, j));
}
c[i] = kbuff._sum;
lnnz += (c[i]!=0) ? 1 : 0;
@@ -353,7 +360,7 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
kbuff.set(0, 0);
for(int j=0; j<n; j++) {
double valij = (sblock != null) ? sblock.get(i, j) : 0;
- kplus.execute2( kbuff, genexec(valij, b, scalars, n, m, i, j));
+ kplus.execute2( kbuff, genexec(valij, b, scalars, m, n, i, j));
}
c[i] = kbuff._sum;
lnnz += (c[i]!=0) ? 1 : 0;
@@ -364,24 +371,27 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
return lnnz;
}
- protected abstract double genexec( double a, double[][] b, double[] scalars, int n, int m, int rowIndex, int colIndex);
+ protected abstract double genexec( double a, double[][] b, double[] scalars, int m, int n, int rowIndex, int colIndex);
private class ParAggTask implements Callable<Double>
{
private final MatrixBlock _a;
private final double[][] _b;
private final double[] _scalars;
- private final int _clen;
private final int _rlen;
+ private final int _clen;
+ private final boolean _safe;
private final int _rl;
private final int _ru;
- protected ParAggTask( MatrixBlock a, double[][] b, double[] scalars, int clen, int rlen, int rl, int ru ) {
+ protected ParAggTask( MatrixBlock a, double[][] b, double[] scalars,
+ int rlen, int clen, boolean sparseSafe, int rl, int ru ) {
_a = a;
_b = b;
_scalars = scalars;
- _clen = clen;
_rlen = rlen;
+ _clen = clen;
+ _safe = sparseSafe;
_rl = rl;
_ru = ru;
}
@@ -389,8 +399,8 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
@Override
public Double call() throws DMLRuntimeException {
return ( !_a.isInSparseFormat()) ?
- executeDenseAndAgg(_a.getDenseBlock(), _b, _scalars, _clen, _rlen, _rl, _ru) :
- executeSparseAndAgg(_a.getSparseBlock(), _b, _scalars, _clen, _rlen, _rl, _ru);
+ executeDenseAndAgg(_a.getDenseBlock(), _b, _scalars, _rlen, _clen, _safe, _rl, _ru) :
+ executeSparseAndAgg(_a.getSparseBlock(), _b, _scalars, _rlen, _clen, _safe, _rl, _ru);
}
}
@@ -400,18 +410,21 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
private final double[][] _b;
private final double[] _scalars;
private final double[] _c;
- private final int _clen;
private final int _rlen;
+ private final int _clen;
+ private final boolean _safe;
private final int _rl;
private final int _ru;
- protected ParExecTask( MatrixBlock a, double[][] b, double[] scalars, double[] c, int clen, int rlen, int rl, int ru ) {
+ protected ParExecTask( MatrixBlock a, double[][] b, double[] scalars, double[] c,
+ int rlen, int clen, boolean sparseSafe, int rl, int ru ) {
_a = a;
_b = b;
_scalars = scalars;
_c = c;
- _clen = clen;
_rlen = rlen;
+ _clen = clen;
+ _safe = sparseSafe;
_rl = rl;
_ru = ru;
}
@@ -419,8 +432,8 @@ public abstract class SpoofCellwise extends SpoofOperator implements Serializabl
@Override
public Long call() throws DMLRuntimeException {
return (!_a.isInSparseFormat()) ?
- executeDense(_a.getDenseBlock(), _b, _scalars, _c, _clen, _rlen, _rl, _ru) :
- executeSparse(_a.getSparseBlock(), _b, _scalars, _c, _clen, _rlen, _rl, _ru);
+ executeDense(_a.getDenseBlock(), _b, _scalars, _c, _rlen, _clen, _safe, _rl, _ru) :
+ executeSparse(_a.getSparseBlock(), _b, _scalars, _c, _rlen, _clen, _safe, _rl, _ru);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/30f72e83/src/test/java/org/apache/sysml/test/integration/functions/codegen/CellwiseTmplTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/sysml/test/integration/functions/codegen/CellwiseTmplTest.java b/src/test/java/org/apache/sysml/test/integration/functions/codegen/CellwiseTmplTest.java
index 08edd03..32105fd 100644
--- a/src/test/java/org/apache/sysml/test/integration/functions/codegen/CellwiseTmplTest.java
+++ b/src/test/java/org/apache/sysml/test/integration/functions/codegen/CellwiseTmplTest.java
@@ -35,13 +35,15 @@ import org.apache.sysml.test.utils.TestUtils;
public class CellwiseTmplTest extends AutomatedTestBase
{
- private static final String TEST_NAME1 = "cellwisetmpl1";
- private static final String TEST_NAME2 = "cellwisetmpl2";
- private static final String TEST_NAME3 = "cellwisetmpl3";
- private static final String TEST_NAME4 = "cellwisetmpl4";
- private static final String TEST_NAME5 = "cellwisetmpl5";
- private static final String TEST_NAME6 = "cellwisetmpl6";
- private static final String TEST_NAME7 = "cellwisetmpl7";
+ private static final String TEST_NAME = "cellwisetmpl";
+ private static final String TEST_NAME1 = TEST_NAME+1;
+ private static final String TEST_NAME2 = TEST_NAME+2;
+ private static final String TEST_NAME3 = TEST_NAME+3;
+ private static final String TEST_NAME4 = TEST_NAME+4;
+ private static final String TEST_NAME5 = TEST_NAME+5;
+ private static final String TEST_NAME6 = TEST_NAME+6;
+ private static final String TEST_NAME7 = TEST_NAME+7;
+ private static final String TEST_NAME8 = TEST_NAME+8;
private static final String TEST_DIR = "functions/codegen/";
private static final String TEST_CLASS_DIR = TEST_DIR + CellwiseTmplTest.class.getSimpleName() + "/";
@@ -53,13 +55,10 @@ public class CellwiseTmplTest extends AutomatedTestBase
@Override
public void setUp() {
TestUtils.clearAssertionInformation();
- addTestConfiguration( TEST_NAME1, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME1, new String[] { "1" }) );
- addTestConfiguration( TEST_NAME2, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME2, new String[] { "2" }) );
- addTestConfiguration( TEST_NAME3, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME3, new String[] { "3" }) );
- addTestConfiguration( TEST_NAME4, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME4, new String[] { "4" }) );
- addTestConfiguration( TEST_NAME5, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME5, new String[] { "5" }) );
- addTestConfiguration( TEST_NAME6, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME6, new String[] { "6" }) );
- addTestConfiguration( TEST_NAME7, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME7, new String[] { "7" }) );
+ for( int i=1; i<=8; i++ ) {
+ addTestConfiguration( TEST_NAME+i, new TestConfiguration(
+ TEST_CLASS_DIR, TEST_NAME+i, new String[] {String.valueOf(i)}) );
+ }
}
@Test
@@ -97,6 +96,11 @@ public class CellwiseTmplTest extends AutomatedTestBase
public void testCodegenCellwiseRewrite7() {
testCodegenIntegration( TEST_NAME7, true, ExecType.CP );
}
+
+ @Test
+ public void testCodegenCellwiseRewrite8() {
+ testCodegenIntegration( TEST_NAME8, true, ExecType.CP );
+ }
@Test
public void testCodegenCellwise1() {
@@ -133,6 +137,11 @@ public class CellwiseTmplTest extends AutomatedTestBase
public void testCodegenCellwise7() {
testCodegenIntegration( TEST_NAME7, false, ExecType.CP );
}
+
+ @Test
+ public void testCodegenCellwise8() {
+ testCodegenIntegration( TEST_NAME8, false, ExecType.CP );
+ }
@Test
public void testCodegenCellwiseRewrite1_sp() {
@@ -144,6 +153,11 @@ public class CellwiseTmplTest extends AutomatedTestBase
testCodegenIntegration( TEST_NAME7, true, ExecType.SPARK );
}
+ @Test
+ public void testCodegenCellwiseRewrite8_sp() {
+ testCodegenIntegration( TEST_NAME8, true, ExecType.SPARK );
+ }
+
private void testCodegenIntegration( String testname, boolean rewrites, ExecType instType )
{
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/30f72e83/src/test/scripts/functions/codegen/cellwisetmpl8.R
----------------------------------------------------------------------
diff --git a/src/test/scripts/functions/codegen/cellwisetmpl8.R b/src/test/scripts/functions/codegen/cellwisetmpl8.R
new file mode 100644
index 0000000..9ad506b
--- /dev/null
+++ b/src/test/scripts/functions/codegen/cellwisetmpl8.R
@@ -0,0 +1,33 @@
+#-------------------------------------------------------------
+#
+# 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.
+#
+#-------------------------------------------------------------
+
+args<-commandArgs(TRUE)
+options(digits=22)
+library("Matrix")
+
+X = matrix(1, 1002, 23);
+Y = seq(1, 1002);
+X[100:900,] = matrix(0, 801, 23);
+if(1==1){}
+
+R = X * ((X + 7.7) * (Y%*%matrix(1,1,23)));
+
+writeMM(as(R,"CsparseMatrix"), paste(args[2], "S", sep=""));
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/30f72e83/src/test/scripts/functions/codegen/cellwisetmpl8.dml
----------------------------------------------------------------------
diff --git a/src/test/scripts/functions/codegen/cellwisetmpl8.dml b/src/test/scripts/functions/codegen/cellwisetmpl8.dml
new file mode 100644
index 0000000..d330d70
--- /dev/null
+++ b/src/test/scripts/functions/codegen/cellwisetmpl8.dml
@@ -0,0 +1,29 @@
+#-------------------------------------------------------------
+#
+# 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.
+#
+#-------------------------------------------------------------
+
+X = matrix(1, rows=1002, cols=23);
+Y = seq(1, 1002);
+X[100:900,] = matrix(0, rows=801, cols=23);
+if(1==1){}
+
+R = X * ((X + 7.7) * Y);
+
+write(R, $1)