Repository: systemml Updated Branches: refs/heads/master a66126d49 -> 11723cde8
[SYSTEMML-1345] Removed instruction patching rand/seq/sample dims This patch removes unnecessary instruction patching for the rows, cols, and seq parameters of CP and Spark rand, seq, and sample instructions. This avoids unnecessary strings replacement and instruction parsing in case of initial unknowns and forced singlenode execution type (e.g., JMLC), which improves performance for scenarios with many small rand and seq operations. Project: http://git-wip-us.apache.org/repos/asf/systemml/repo Commit: http://git-wip-us.apache.org/repos/asf/systemml/commit/11723cde Tree: http://git-wip-us.apache.org/repos/asf/systemml/tree/11723cde Diff: http://git-wip-us.apache.org/repos/asf/systemml/diff/11723cde Branch: refs/heads/master Commit: 11723cde8f6827e8a130cc07bf94cc75c37bcf4b Parents: a66126d Author: Matthias Boehm <[email protected]> Authored: Wed Nov 8 20:54:25 2017 -0800 Committer: Matthias Boehm <[email protected]> Committed: Wed Nov 8 21:58:51 2017 -0800 ---------------------------------------------------------------------- .../java/org/apache/sysml/lops/DataGen.java | 36 +-- .../runtime/instructions/cp/BooleanObject.java | 2 +- .../instructions/cp/DataGenCPInstruction.java | 150 +++++----- .../runtime/instructions/cp/DoubleObject.java | 2 +- .../runtime/instructions/cp/IntObject.java | 2 +- .../runtime/instructions/cp/StringObject.java | 2 +- .../instructions/spark/RandSPInstruction.java | 272 +++++++------------ 7 files changed, 185 insertions(+), 281 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/systemml/blob/11723cde/src/main/java/org/apache/sysml/lops/DataGen.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/lops/DataGen.java b/src/main/java/org/apache/sysml/lops/DataGen.java index d30efe2..51bfb5a 100644 --- a/src/main/java/org/apache/sysml/lops/DataGen.java +++ b/src/main/java/org/apache/sysml/lops/DataGen.java @@ -123,13 +123,13 @@ public class DataGen extends Lop switch( method ) { case RAND: - return getCPInstruction_Rand(output); + return getRandInstructionCPSpark(output); case SINIT: - return getCPInstruction_SInit(output); + return getSInitInstructionCPSpark(output); case SEQ: - return getCPInstruction_Seq(output); + return getSeqInstructionCPSpark(output); case SAMPLE: - return getCPInstruction_Sample(output); + return getSampleInstructionCPSpark(output); default: throw new LopsException("Unknown data generation method: " + method); @@ -142,9 +142,9 @@ public class DataGen extends Lop { switch(method) { case RAND: - return getMRInstruction_Rand(inputIndex, outputIndex); + return getRandInstructionMR(inputIndex, outputIndex); case SEQ: - return getMRInstruction_Seq(inputIndex, outputIndex); + return getSeqInstructionMR(inputIndex, outputIndex); default: throw new LopsException("Unknown data generation method: " + method); @@ -158,7 +158,7 @@ public class DataGen extends Lop * @return cp instruction for rand * @throws LopsException if LopsException occurs */ - private String getCPInstruction_Rand(String output) + private String getRandInstructionCPSpark(String output) throws LopsException { //sanity checks @@ -178,11 +178,11 @@ public class DataGen extends Lop sb.append(OPERAND_DELIMITOR); Lop iLop = _inputParams.get(DataExpression.RAND_ROWS.toString()); - sb.append(iLop.prepScalarLabel()); + sb.append(iLop.prepScalarInputOperand(getExecType())); sb.append(OPERAND_DELIMITOR); iLop = _inputParams.get(DataExpression.RAND_COLS.toString()); - sb.append(iLop.prepScalarLabel()); + sb.append(iLop.prepScalarInputOperand(getExecType())); sb.append(OPERAND_DELIMITOR); sb.append(String.valueOf(getOutputParameters().getRowsInBlock())); @@ -238,7 +238,7 @@ public class DataGen extends Lop return sb.toString(); } - private String getCPInstruction_SInit(String output) + private String getSInitInstructionCPSpark(String output) throws LopsException { if ( method != DataGenMethod.SINIT ) @@ -286,7 +286,7 @@ public class DataGen extends Lop return sb.toString(); } - private String getCPInstruction_Sample(String output) + private String getSampleInstructionCPSpark(String output) throws LopsException { if ( method != DataGenMethod.SAMPLE ) @@ -294,7 +294,7 @@ public class DataGen extends Lop //prepare instruction parameters Lop lsize = _inputParams.get(DataExpression.RAND_ROWS.toString()); - String size = lsize.prepScalarLabel(); + String size = lsize.prepScalarInputOperand(getExecType()); Lop lrange = _inputParams.get(DataExpression.RAND_MAX.toString()); String range = lrange.prepScalarLabel(); @@ -337,7 +337,7 @@ public class DataGen extends Lop * @return cp instruction for seq * @throws LopsException if LopsException occurs */ - private String getCPInstruction_Seq(String output) throws LopsException { + private String getSeqInstructionCPSpark(String output) throws LopsException { if ( method != DataGenMethod.SEQ ) throw new LopsException("Invalid instruction generation for data generation method " + method); @@ -349,13 +349,13 @@ public class DataGen extends Lop Lop iLop = null; iLop = _inputParams.get(Statement.SEQ_FROM.toString()); - String fromString = iLop.prepScalarLabel(); + String fromString = iLop.prepScalarInputOperand(et); iLop = _inputParams.get(Statement.SEQ_TO.toString()); - String toString = iLop.prepScalarLabel(); + String toString = iLop.prepScalarInputOperand(et); iLop = _inputParams.get(Statement.SEQ_INCR.toString()); - String incrString = iLop.prepScalarLabel(); + String incrString = iLop.prepScalarInputOperand(et); String rowsString = String.valueOf(this.getOutputParameters().getNumRows()); String colsString = String.valueOf(this.getOutputParameters().getNumCols()); @@ -395,7 +395,7 @@ public class DataGen extends Lop * @return mr instruction for rand * @throws LopsException if LopsException occurs */ - private String getMRInstruction_Rand(int inputIndex, int outputIndex) + private String getRandInstructionMR(int inputIndex, int outputIndex) throws LopsException { //sanity checks @@ -474,7 +474,7 @@ public class DataGen extends Lop * @return mr instruction for seq * @throws LopsException if LopsException occurs */ - private String getMRInstruction_Seq(int inputIndex, int outputIndex) throws LopsException { + private String getSeqInstructionMR(int inputIndex, int outputIndex) throws LopsException { StringBuilder sb = new StringBuilder(); sb.append( getExecType() ); http://git-wip-us.apache.org/repos/asf/systemml/blob/11723cde/src/main/java/org/apache/sysml/runtime/instructions/cp/BooleanObject.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/BooleanObject.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/BooleanObject.java index dff1339..757807f 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/BooleanObject.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/BooleanObject.java @@ -29,7 +29,7 @@ public class BooleanObject extends ScalarObject { private static final long serialVersionUID = -4506242165735516984L; - private boolean _value; + private final boolean _value; public BooleanObject(boolean val){ this(null,val); http://git-wip-us.apache.org/repos/asf/systemml/blob/11723cde/src/main/java/org/apache/sysml/runtime/instructions/cp/DataGenCPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/DataGenCPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/DataGenCPInstruction.java index 236f2ae..61a8bb7 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/DataGenCPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/DataGenCPInstruction.java @@ -23,6 +23,8 @@ import org.apache.sysml.hops.DataGenOp; import org.apache.sysml.hops.Hop.DataGenMethod; import org.apache.sysml.lops.DataGen; import org.apache.sysml.lops.Lop; +import org.apache.sysml.parser.Expression.DataType; +import org.apache.sysml.parser.Expression.ValueType; import org.apache.sysml.runtime.DMLRuntimeException; import org.apache.sysml.runtime.controlprogram.context.ExecutionContext; import org.apache.sysml.runtime.instructions.InstructionUtils; @@ -36,23 +38,23 @@ public class DataGenCPInstruction extends UnaryCPInstruction { private DataGenMethod method = DataGenMethod.INVALID; - private final long rows, cols; + private final CPOperand rows, cols; private final int rowsInBlock, colsInBlock; private final double minValue, maxValue, sparsity; private final String pdf, pdfParams; private final long seed; // sequence specific attributes - private final double seq_from, seq_to, seq_incr; + private final CPOperand seq_from, seq_to, seq_incr; // sample specific attributes private final boolean replace; private final int numThreads; private DataGenCPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, - long rows, long cols, int rpb, int cpb, double minValue, double maxValue, double sparsity, long seed, + CPOperand rows, CPOperand cols, int rpb, int cpb, double minValue, double maxValue, double sparsity, long seed, String probabilityDensityFunction, String pdfParams, int k, - double seqFrom, double seqTo, double seqIncr, boolean replace, String opcode, String istr) { + CPOperand seqFrom, CPOperand seqTo, CPOperand seqIncr, boolean replace, String opcode, String istr) { super(CPType.Rand, op, in, out, opcode, istr); this.method = mthd; this.rows = rows; @@ -72,32 +74,33 @@ public class DataGenCPInstruction extends UnaryCPInstruction { this.replace = replace; } - private DataGenCPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, long rows, long cols, + private DataGenCPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, CPOperand rows, CPOperand cols, int rpb, int cpb, double minValue, double maxValue, double sparsity, long seed, String probabilityDensityFunction, String pdfParams, int k, String opcode, String istr) { this(op, mthd, in, out, rows, cols, rpb, cpb, minValue, maxValue, sparsity, seed, - probabilityDensityFunction, pdfParams, k, -1, -1, -1, false, opcode, istr); + probabilityDensityFunction, pdfParams, k, null, null, null, false, opcode, istr); } - private DataGenCPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, long rows, long cols, + private DataGenCPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, CPOperand rows, CPOperand cols, int rpb, int cpb, double maxValue, boolean replace, long seed, String opcode, String istr) { this(op, mthd, in, out, rows, cols, rpb, cpb, 0, maxValue, 1.0, seed, - null, null, 1, -1, -1, -1, replace, opcode, istr); + null, null, 1, null, null, null, replace, opcode, istr); } - private DataGenCPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, long rows, long cols, - int rpb, int cpb, double seqFrom, double seqTo, double seqIncr, String opcode, String istr) { + private DataGenCPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, CPOperand rows, CPOperand cols, + int rpb, int cpb, CPOperand seqFrom, CPOperand seqTo, CPOperand seqIncr, String opcode, String istr) { this(op, mthd, in, out, rows, cols, rpb, cpb, 0, 1, 1.0, -1, null, null, 1, seqFrom, seqTo, seqIncr, false, opcode, istr); } public long getRows() { - return rows; + return rows.isLiteral() ? Long.parseLong(rows.getName()) : -1; } public long getCols() { - return cols; + return cols.isLiteral() ? Long.parseLong(cols.getName()) : -1; } + public int getRowsInBlock() { return rowsInBlock; } @@ -141,87 +144,48 @@ public class DataGenCPInstruction extends UnaryCPInstruction { InstructionUtils.checkNumFields ( s, 7 ); } - CPOperand out = new CPOperand(s[s.length-1]); // ouput is specified by the last operand + CPOperand out = new CPOperand(s[s.length-1]); Operator op = null; if ( method == DataGenMethod.RAND ) { - long rows = -1, cols = -1; - if (!s[1].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - rows = Double.valueOf(s[1]).longValue(); - } - if (!s[2].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - cols = Double.valueOf(s[2]).longValue(); - } - + CPOperand rows = new CPOperand(s[1]); + CPOperand cols = new CPOperand(s[2]); int rpb = Integer.parseInt(s[3]); int cpb = Integer.parseInt(s[4]); - - double minValue = -1, maxValue = -1; - if (!s[5].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - minValue = Double.valueOf(s[5]).doubleValue(); - } - if (!s[6].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - maxValue = Double.valueOf(s[6]).doubleValue(); - } - - double sparsity = -1; - if (!s[7].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - sparsity = Double.valueOf(s[7]); - } - - long seed = DataGenOp.UNSPECIFIED_SEED; - if( !s[8].contains( Lop.VARIABLE_NAME_PLACEHOLDER)){ - seed = Long.parseLong(s[8]); - } - + double minValue = !s[5].contains(Lop.VARIABLE_NAME_PLACEHOLDER) ? + Double.valueOf(s[5]).doubleValue() : -1; + double maxValue = !s[6].contains(Lop.VARIABLE_NAME_PLACEHOLDER) ? + Double.valueOf(s[6]).doubleValue() : -1; + double sparsity = !s[7].contains(Lop.VARIABLE_NAME_PLACEHOLDER) ? + Double.valueOf(s[7]).doubleValue() : -1; + long seed = !s[8].contains(Lop.VARIABLE_NAME_PLACEHOLDER) ? + Long.valueOf(s[8]).longValue() : -1; String pdf = s[9]; - String pdfParams = null; - if (!s[10].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - pdfParams = s[10]; - } - + String pdfParams = !s[10].contains( Lop.VARIABLE_NAME_PLACEHOLDER) ? + s[10] : null; int k = Integer.parseInt(s[11]); return new DataGenCPInstruction(op, method, null, out, rows, cols, rpb, cpb, minValue, maxValue, sparsity, seed, pdf, pdfParams, k, opcode, str); } else if ( method == DataGenMethod.SEQ) { - // Example Instruction: CP:seq:11:1:1000:1000:1:0:-0.1:scratch_space/_p7932_192.168.1.120//_t0/:mVar1 - long rows = Double.valueOf(s[1]).longValue(); - long cols = Double.valueOf(s[2]).longValue(); int rpb = Integer.parseInt(s[3]); int cpb = Integer.parseInt(s[4]); + CPOperand from = new CPOperand(s[5]); + CPOperand to = new CPOperand(s[6]); + CPOperand incr = new CPOperand(s[7]); - double from, to, incr; - from = to = incr = Double.NaN; - if (!s[5].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - from = Double.valueOf(s[5]); - } - if (!s[6].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - to = Double.valueOf(s[6]); - } - if (!s[7].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - incr = Double.valueOf(s[7]); - } - - return new DataGenCPInstruction(op, method, null, out, rows, cols, rpb, cpb, from, to, incr, opcode, str); + return new DataGenCPInstruction(op, method, null, out, null, null, rpb, cpb, from, to, incr, opcode, str); } else if ( method == DataGenMethod.SAMPLE) { - // Example Instruction: CP:sample:10:100:false:1000:1000:_mVar2·MATRIX·DOUBLE - double max = 0; - long rows = 0, cols; - boolean replace = false; - - if (!s[1].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) - max = Double.valueOf(s[1]); - if (!s[2].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) - rows = Double.valueOf(s[2]).longValue(); - cols = 1; - - if (!s[3].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) - replace = Boolean.valueOf(s[3]); + double max = !s[1].contains(Lop.VARIABLE_NAME_PLACEHOLDER) ? + Double.valueOf(s[1]) : 0; + CPOperand rows = new CPOperand(s[2]); + CPOperand cols = new CPOperand("1", ValueType.INT, DataType.SCALAR); + boolean replace = (!s[3].contains(Lop.VARIABLE_NAME_PLACEHOLDER) + && Boolean.valueOf(s[3])); long seed = Long.parseLong(s[4]); int rpb = Integer.parseInt(s[5]); @@ -239,12 +203,12 @@ public class DataGenCPInstruction extends UnaryCPInstruction { { MatrixBlock soresBlock = null; - //check valid for integer dimensions (we cannot even represent empty blocks with larger dimensions) - if( rows > Integer.MAX_VALUE || cols > Integer.MAX_VALUE ) - throw new DMLRuntimeException("RandCPInstruction does not support dimensions larger than integer: rows="+rows+", cols="+cols+"."); - //process specific datagen operator if ( method == DataGenMethod.RAND ) { + long lrows = ec.getScalarInput(rows).getLongValue(); + long lcols = ec.getScalarInput(cols).getLongValue(); + checkValidDimensions(lrows, lcols); + //generate pseudo-random seed (because not specified) long lSeed = seed; //seed per invocation if( lSeed == DataGenOp.UNSPECIFIED_SEED ) @@ -253,33 +217,47 @@ public class DataGenCPInstruction extends UnaryCPInstruction { if( LOG.isTraceEnabled() ) LOG.trace("Process DataGenCPInstruction rand with seed = "+lSeed+"."); - RandomMatrixGenerator rgen = LibMatrixDatagen.createRandomMatrixGenerator(pdf, (int) rows, (int) cols, rowsInBlock, colsInBlock, sparsity, minValue, maxValue, pdfParams); + RandomMatrixGenerator rgen = LibMatrixDatagen.createRandomMatrixGenerator( + pdf, (int) lrows, (int) lcols, rowsInBlock, colsInBlock, sparsity, minValue, maxValue, pdfParams); soresBlock = MatrixBlock.randOperations(rgen, seed, numThreads); } else if ( method == DataGenMethod.SEQ ) { + double lfrom = ec.getScalarInput(seq_from).getDoubleValue(); + double lto = ec.getScalarInput(seq_to).getDoubleValue(); + double lincr = ec.getScalarInput(seq_incr).getDoubleValue(); + //handle default 1 to -1 for special case of from>to - double lSeqIncr = LibMatrixDatagen.updateSeqIncr(seq_from, seq_to, seq_incr); + lincr = LibMatrixDatagen.updateSeqIncr(lfrom, lto, lincr); if( LOG.isTraceEnabled() ) - LOG.trace("Process DataGenCPInstruction seq with seqFrom="+seq_from+", seqTo="+seq_to+", seqIncr"+lSeqIncr); + LOG.trace("Process DataGenCPInstruction seq with seqFrom="+lfrom+", seqTo="+lto+", seqIncr"+lincr); - soresBlock = MatrixBlock.seqOperations(seq_from, seq_to, lSeqIncr); + soresBlock = MatrixBlock.seqOperations(lfrom, lto, lincr); } else if ( method == DataGenMethod.SAMPLE ) { + long lrows = ec.getScalarInput(rows).getLongValue(); long range = UtilFunctions.toLong(maxValue); + checkValidDimensions(lrows, 1); if( LOG.isTraceEnabled() ) - LOG.trace("Process DataGenCPInstruction sample with range="+range+", size="+rows+", replace"+replace + ", seed=" + seed); + LOG.trace("Process DataGenCPInstruction sample with range="+range+", size="+lrows+", replace"+replace + ", seed=" + seed); - if ( range < rows && !replace ) - throw new DMLRuntimeException("Sample (size=" + rows + ") larger than population (size=" + range + ") can only be generated with replacement."); + if ( range < lrows && !replace ) + throw new DMLRuntimeException("Sample (size=" + lrows + ") larger than population (size=" + range + ") can only be generated with replacement."); - soresBlock = MatrixBlock.sampleOperations(range, (int)rows, replace, seed); + soresBlock = MatrixBlock.sampleOperations(range, (int)lrows, replace, seed); } //release created output ec.setMatrixOutput(output.getName(), soresBlock, getExtendedOpcode()); } + + private static void checkValidDimensions(long rows, long cols) throws DMLRuntimeException { + //check valid for integer dimensions (we cannot even represent empty blocks with larger dimensions) + if( rows > Integer.MAX_VALUE || cols > Integer.MAX_VALUE ) + throw new DMLRuntimeException("DataGenCPInstruction does not " + + "support dimensions larger than integer: rows="+rows+", cols="+cols+"."); + } } http://git-wip-us.apache.org/repos/asf/systemml/blob/11723cde/src/main/java/org/apache/sysml/runtime/instructions/cp/DoubleObject.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/DoubleObject.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/DoubleObject.java index 0bad105..4290788 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/DoubleObject.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/DoubleObject.java @@ -26,7 +26,7 @@ public class DoubleObject extends ScalarObject { private static final long serialVersionUID = -8525290101679236360L; - private double _value; + private final double _value; public DoubleObject(double val){ this(null,val); http://git-wip-us.apache.org/repos/asf/systemml/blob/11723cde/src/main/java/org/apache/sysml/runtime/instructions/cp/IntObject.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/IntObject.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/IntObject.java index 06c31d9..d4d44a9 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/IntObject.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/IntObject.java @@ -27,7 +27,7 @@ public class IntObject extends ScalarObject private static final long serialVersionUID = 353170585998999528L; //we use consistently to the compiler long in terms of integer (8 byte) - private long _value; + private final long _value; public IntObject(long val) { this(null,val); http://git-wip-us.apache.org/repos/asf/systemml/blob/11723cde/src/main/java/org/apache/sysml/runtime/instructions/cp/StringObject.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/StringObject.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/StringObject.java index 22fdbcd..844d8a6 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/StringObject.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/StringObject.java @@ -28,7 +28,7 @@ public class StringObject extends ScalarObject private static final int MAX_STRING_SIZE = 1*1024*1024; //1MB - private String _value; + private final String _value; public StringObject (String val){ this(null,val); http://git-wip-us.apache.org/repos/asf/systemml/blob/11723cde/src/main/java/org/apache/sysml/runtime/instructions/spark/RandSPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/spark/RandSPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/spark/RandSPInstruction.java index 2266eeb..ee9fefb 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/spark/RandSPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/spark/RandSPInstruction.java @@ -48,6 +48,8 @@ import org.apache.sysml.hops.Hop.DataGenMethod; import org.apache.sysml.hops.OptimizerUtils; import org.apache.sysml.lops.DataGen; import org.apache.sysml.lops.Lop; +import org.apache.sysml.parser.Expression.DataType; +import org.apache.sysml.parser.Expression.ValueType; import org.apache.sysml.runtime.DMLRuntimeException; import org.apache.sysml.runtime.controlprogram.context.ExecutionContext; import org.apache.sysml.runtime.controlprogram.context.SparkExecutionContext; @@ -71,30 +73,23 @@ public class RandSPInstruction extends UnarySPInstruction { private static final long INMEMORY_NUMBLOCKS_THRESHOLD = 1024 * 1024; private DataGenMethod method = DataGenMethod.INVALID; - - private long rows; - private long cols; - private int rowsInBlock; - private int colsInBlock; - private double minValue; - private double maxValue; - private double sparsity; - private String pdf; - private String pdfParams; + private final CPOperand rows, cols; + private final int rowsInBlock, colsInBlock; + private final double minValue, maxValue; + private final double sparsity; + private final String pdf, pdfParams; private long seed = 0; - private String dir; - private double seq_from; - private double seq_to; - private double seq_incr; + private final String dir; + private final CPOperand seq_from, seq_to, seq_incr; // sample specific attributes - private boolean replace; + private final boolean replace; - private RandSPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, long rows, long cols, - int rpb, int cpb, double minValue, double maxValue, double sparsity, long seed, String dir, - String probabilityDensityFunction, String pdfParams, String opcode, String istr) { + private RandSPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, + CPOperand rows, CPOperand cols, int rpb, int cpb, double minValue, double maxValue, double sparsity, long seed, + String dir, String probabilityDensityFunction, String pdfParams, + CPOperand seqFrom, CPOperand seqTo, CPOperand seqIncr, boolean replace, String opcode, String istr) { super(op, in, out, opcode, istr); - this.method = mthd; this.rows = rows; this.cols = cols; @@ -107,92 +102,59 @@ public class RandSPInstruction extends UnarySPInstruction { this.dir = dir; this.pdf = probabilityDensityFunction; this.pdfParams = pdfParams; - - } - - private RandSPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, long rows, long cols, - int rpb, int cpb, double seqFrom, double seqTo, double seqIncr, String opcode, String istr) { - super(op, in, out, opcode, istr); - this.method = mthd; - this.rows = rows; - this.cols = cols; - this.rowsInBlock = rpb; - this.colsInBlock = cpb; this.seq_from = seqFrom; this.seq_to = seqTo; this.seq_incr = seqIncr; - } - - private RandSPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, long rows, long cols, - int rpb, int cpb, double maxValue, boolean replace, long seed, String opcode, String istr) { - super(op, in, out, opcode, istr); - - this.method = mthd; - this.rows = rows; - this.cols = cols; - this.rowsInBlock = rpb; - this.colsInBlock = cpb; - this.maxValue = maxValue; this.replace = replace; - this.seed = seed; + } + + private RandSPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, CPOperand rows, CPOperand cols, + int rpb, int cpb, double minValue, double maxValue, double sparsity, long seed, String dir, + String probabilityDensityFunction, String pdfParams, String opcode, String istr) { + this(op, mthd, in, out, rows, cols, rpb, cpb, minValue, maxValue, sparsity, + seed, dir, probabilityDensityFunction, pdfParams, null, null, null, false, opcode, istr); } - public long getRows() { - return rows; + private RandSPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, CPOperand rows, CPOperand cols, + int rpb, int cpb, CPOperand seqFrom, CPOperand seqTo, CPOperand seqIncr, String opcode, String istr) { + this(op, mthd, in, out, rows, cols, rpb, cpb, -1, -1, -1, + -1, null, null, null, seqFrom, seqTo, seqIncr, false, opcode, istr); } - public void setRows(long rows) { - this.rows = rows; + private RandSPInstruction(Operator op, DataGenMethod mthd, CPOperand in, CPOperand out, CPOperand rows, CPOperand cols, + int rpb, int cpb, double maxValue, boolean replace, long seed, String opcode, String istr) { + this(op, mthd, in, out, rows, cols, rpb, cpb, -1, maxValue, -1, + seed, null, null, null, null, null, null, replace, opcode, istr); } - public long getCols() { - return cols; + public long getRows() { + return rows.isLiteral() ? Long.parseLong(rows.getName()) : -1; } - public void setCols(long cols) { - this.cols = cols; + public long getCols() { + return cols.isLiteral() ? Long.parseLong(cols.getName()) : -1; } public int getRowsInBlock() { return rowsInBlock; } - public void setRowsInBlock(int rowsInBlock) { - this.rowsInBlock = rowsInBlock; - } - public int getColsInBlock() { return colsInBlock; } - public void setColsInBlock(int colsInBlock) { - this.colsInBlock = colsInBlock; - } - public double getMinValue() { return minValue; } - public void setMinValue(double minValue) { - this.minValue = minValue; - } - public double getMaxValue() { return maxValue; } - public void setMaxValue(double maxValue) { - this.maxValue = maxValue; - } - public double getSparsity() { return sparsity; } - public void setSparsity(double sparsity) { - this.sparsity = sparsity; - } - public static RandSPInstruction parseInstruction(String str) throws DMLRuntimeException { @@ -220,78 +182,43 @@ public class RandSPInstruction extends UnarySPInstruction { CPOperand out = new CPOperand(s[s.length-1]); if ( method == DataGenMethod.RAND ) { - long rows = -1, cols = -1; - if (!s[1].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - rows = Double.valueOf(s[1]).longValue(); - } - if (!s[2].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - cols = Double.valueOf(s[2]).longValue(); - } - + CPOperand rows = new CPOperand(s[1]); + CPOperand cols = new CPOperand(s[2]); int rpb = Integer.parseInt(s[3]); int cpb = Integer.parseInt(s[4]); - - double minValue = -1, maxValue = -1; - if (!s[5].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - minValue = Double.valueOf(s[5]).doubleValue(); - } - if (!s[6].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - maxValue = Double.valueOf(s[6]).doubleValue(); - } - - double sparsity = -1; - if (!s[7].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - sparsity = Double.valueOf(s[7]); - } - - long seed = DataGenOp.UNSPECIFIED_SEED; - if (!s[8].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - seed = Long.parseLong(s[8]); - } - + double minValue = !s[5].contains(Lop.VARIABLE_NAME_PLACEHOLDER) ? + Double.valueOf(s[5]).doubleValue() : -1; + double maxValue = !s[6].contains(Lop.VARIABLE_NAME_PLACEHOLDER) ? + Double.valueOf(s[6]).doubleValue() : -1; + double sparsity = !s[7].contains(Lop.VARIABLE_NAME_PLACEHOLDER) ? + Double.valueOf(s[7]).doubleValue() : -1; + long seed = !s[8].contains(Lop.VARIABLE_NAME_PLACEHOLDER) ? + Long.valueOf(s[8]).longValue() : -1; String dir = s[9]; - String pdf = s[10]; - String pdfParams = s[11]; + String pdf = s[10]; + String pdfParams = !s[11].contains( Lop.VARIABLE_NAME_PLACEHOLDER) ? + s[11] : null; return new RandSPInstruction(op, method, null, out, rows, cols, rpb, cpb, minValue, maxValue, sparsity, seed, dir, pdf, pdfParams, opcode, str); } else if ( method == DataGenMethod.SEQ) { - // Example Instruction: CP:seq:11:1:1000:1000:1:0:-0.1:scratch_space/_p7932_192.168.1.120//_t0/:mVar1 - long rows = Double.valueOf(s[1]).longValue(); - long cols = Double.valueOf(s[2]).longValue(); int rpb = Integer.parseInt(s[3]); int cpb = Integer.parseInt(s[4]); - - double from, to, incr; - from = to = incr = Double.NaN; - if (!s[5].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - from = Double.valueOf(s[5]); - } - if (!s[6].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - to = Double.valueOf(s[6]); - } - if (!s[7].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) { - incr = Double.valueOf(s[7]); - } + CPOperand from = new CPOperand(s[5]); + CPOperand to = new CPOperand(s[6]); + CPOperand incr = new CPOperand(s[7]); CPOperand in = null; - return new RandSPInstruction(op, method, in, out, rows, cols, rpb, cpb, from, to, incr, opcode, str); + return new RandSPInstruction(op, method, in, out, null, null, rpb, cpb, from, to, incr, opcode, str); } else if ( method == DataGenMethod.SAMPLE) { - // Example Instruction: SPARK:sample:10:100:false:1000:1000:_mVar2·MATRIX·DOUBLE - double max = 0; - long rows = 0, cols; - boolean replace = false; - - if (!s[1].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) - max = Double.valueOf(s[1]); - if (!s[2].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) - rows = Double.valueOf(s[2]).longValue(); - cols = 1; - - if (!s[3].contains( Lop.VARIABLE_NAME_PLACEHOLDER)) - replace = Boolean.valueOf(s[3]); + double max = !s[1].contains(Lop.VARIABLE_NAME_PLACEHOLDER) ? + Double.valueOf(s[1]) : 0; + CPOperand rows = new CPOperand(s[2]); + CPOperand cols = new CPOperand("1", ValueType.INT, DataType.SCALAR); + boolean replace = (!s[3].contains(Lop.VARIABLE_NAME_PLACEHOLDER) + && Boolean.valueOf(s[3])); long seed = Long.parseLong(s[4]); int rpb = Integer.parseInt(s[5]); @@ -313,7 +240,7 @@ public class RandSPInstruction extends UnarySPInstruction { switch( method ) { case RAND: generateRandData(sec); break; case SEQ: generateSequence(sec); break; - case SAMPLE: generateSample(sec); break; + case SAMPLE: generateSample(sec); break; default: throw new DMLRuntimeException("Invalid datagen method: "+method); } @@ -322,6 +249,9 @@ public class RandSPInstruction extends UnarySPInstruction { private void generateRandData(SparkExecutionContext sec) throws DMLRuntimeException { + long lrows = sec.getScalarInput(rows).getLongValue(); + long lcols = sec.getScalarInput(cols).getLongValue(); + //step 1: generate pseudo-random seed (because not specified) long lSeed = seed; //seed per invocation if( lSeed == DataGenOp.UNSPECIFIED_SEED ) @@ -331,12 +261,12 @@ public class RandSPInstruction extends UnarySPInstruction { LOG.trace("Process RandSPInstruction rand with seed = "+lSeed+"."); //step 2: potential in-memory rand operations if applicable - if( isMemAvail(rows, cols, sparsity, minValue, maxValue) + if( isMemAvail(lrows, lcols, sparsity, minValue, maxValue) && DMLScript.rtplatform != RUNTIME_PLATFORM.SPARK ) { RandomMatrixGenerator rgen = LibMatrixDatagen.createRandomMatrixGenerator( - pdf, (int)rows, (int)cols, rowsInBlock, colsInBlock, - sparsity, minValue, maxValue, pdfParams); + pdf, (int)lrows, (int)lcols, rowsInBlock, colsInBlock, + sparsity, minValue, maxValue, pdfParams); MatrixBlock mb = MatrixBlock.randOperations(rgen, lSeed); sec.setMatrixOutput(output.getName(), mb, getExtendedOpcode()); @@ -347,13 +277,13 @@ public class RandSPInstruction extends UnarySPInstruction { //step 3: seed generation JavaPairRDD<MatrixIndexes, Tuple2<Long, Long>> seedsRDD = null; Well1024a bigrand = LibMatrixDatagen.setupSeedsForRand(lSeed); - LongStream nnz = LibMatrixDatagen.computeNNZperBlock(rows, cols, rowsInBlock, colsInBlock, sparsity); + LongStream nnz = LibMatrixDatagen.computeNNZperBlock(lrows, lcols, rowsInBlock, colsInBlock, sparsity); PrimitiveIterator.OfLong nnzIter = nnz.iterator(); - double totalSize = OptimizerUtils.estimatePartitionedSizeExactSparsity( rows, cols, rowsInBlock, + double totalSize = OptimizerUtils.estimatePartitionedSizeExactSparsity( lrows, lcols, rowsInBlock, colsInBlock, sparsity); //overestimate for on disk, ensures hdfs block per partition double hdfsBlkSize = InfrastructureAnalyzer.getHDFSBlockSize(); - long numBlocks = new MatrixCharacteristics(rows, cols, rowsInBlock, colsInBlock).getNumBlocks(); - long numColBlocks = (long)Math.ceil((double)cols/(double)colsInBlock); + long numBlocks = new MatrixCharacteristics(lrows, lcols, rowsInBlock, colsInBlock).getNumBlocks(); + long numColBlocks = (long)Math.ceil((double)lcols/(double)colsInBlock); //a) in-memory seed rdd construction if( numBlocks < INMEMORY_NUMBLOCKS_THRESHOLD ) @@ -414,16 +344,16 @@ public class RandSPInstruction extends UnarySPInstruction { //step 4: execute rand instruction over seed input JavaPairRDD<MatrixIndexes, MatrixBlock> out = seedsRDD - .mapToPair(new GenerateRandomBlock(rows, cols, rowsInBlock, colsInBlock, - sparsity, minValue, maxValue, pdf, pdfParams)); + .mapToPair(new GenerateRandomBlock(lrows, lcols, rowsInBlock, colsInBlock, + sparsity, minValue, maxValue, pdf, pdfParams)); //step 5: output handling MatrixCharacteristics mcOut = sec.getMatrixCharacteristics(output.getName()); if(!mcOut.dimsKnown(true)) { //note: we cannot compute the nnz from sparsity because this would not reflect the //actual number of non-zeros, except for extreme values of sparsity equals 0 or 1. - long lnnz = (sparsity==0 || sparsity==1) ? (long) (sparsity*rows*cols) : -1; - mcOut.set(rows, cols, rowsInBlock, colsInBlock, lnnz); + long lnnz = (sparsity==0 || sparsity==1) ? (long) (sparsity*lrows*lcols) : -1; + mcOut.set(lrows, lcols, rowsInBlock, colsInBlock, lnnz); } sec.setRDDHandleForVariable(output.getName(), out); } @@ -431,20 +361,24 @@ public class RandSPInstruction extends UnarySPInstruction { private void generateSequence(SparkExecutionContext sec) throws DMLRuntimeException { + double lfrom = sec.getScalarInput(seq_from).getDoubleValue(); + double lto = sec.getScalarInput(seq_to).getDoubleValue(); + double lincr = sec.getScalarInput(seq_incr).getDoubleValue(); + //sanity check valid increment - if(seq_incr == 0) { - throw new DMLRuntimeException("ERROR: While performing seq(" + seq_from + "," + seq_to + "," + seq_incr + ")"); + if( lincr == 0 ) { + throw new DMLRuntimeException("ERROR: While performing seq(" + lfrom + "," + lto + "," + lincr + ")"); } //handle default 1 to -1 for special case of from>to - seq_incr = LibMatrixDatagen.updateSeqIncr(seq_from, seq_to, seq_incr); + lincr = LibMatrixDatagen.updateSeqIncr(lfrom, lto, lincr); if( LOG.isTraceEnabled() ) - LOG.trace("Process RandSPInstruction seq with seqFrom="+seq_from+", seqTo="+seq_to+", seqIncr"+seq_incr); + LOG.trace("Process RandSPInstruction seq with seqFrom="+lfrom+", seqTo="+lto+", seqIncr"+lincr); //step 1: offset generation JavaRDD<Double> offsetsRDD = null; - long nnz = UtilFunctions.getSeqLength(seq_from, seq_to, seq_incr); + long nnz = UtilFunctions.getSeqLength(lfrom, lto, lincr); double totalSize = OptimizerUtils.estimatePartitionedSizeExactSparsity( nnz, 1, rowsInBlock, colsInBlock, nnz); //overestimate for on disk, ensures hdfs block per partition double hdfsBlkSize = InfrastructureAnalyzer.getHDFSBlockSize(); @@ -455,13 +389,13 @@ public class RandSPInstruction extends UnarySPInstruction { { ArrayList<Double> offsets = new ArrayList<>(); for( long i=0; i<numBlocks; i++ ) { - double off = seq_from + seq_incr*i*rowsInBlock; + double off = lfrom + lincr*i*rowsInBlock; offsets.add(off); } - + //for load balancing: degree of parallelism such that ~128MB per partition int numPartitions = (int) Math.max(Math.min(totalSize/hdfsBlkSize, numBlocks), 1); - + //create offset rdd offsetsRDD = sec.getSparkContext().parallelize(offsets, numPartitions); } @@ -475,7 +409,7 @@ public class RandSPInstruction extends UnarySPInstruction { FileSystem fs = IOUtilFunctions.getFileSystem(path); pw = new PrintWriter(fs.create(path)); for( long i=0; i<numBlocks; i++ ) { - double off = seq_from + seq_incr*i*rowsInBlock; + double off = lfrom + lincr*i*rowsInBlock; pw.println(off); } } @@ -495,14 +429,9 @@ public class RandSPInstruction extends UnarySPInstruction { .map(new ExtractOffsetTuple()); } - //sanity check number of non-zeros - if(nnz != rows && rows != -1) { - throw new DMLRuntimeException("Incorrect number of non-zeros: " + nnz + " != " + rows); - } - //step 2: execute seq instruction over offset input JavaPairRDD<MatrixIndexes, MatrixBlock> out = offsetsRDD - .mapToPair(new GenerateSequenceBlock(rowsInBlock, seq_from, seq_to, seq_incr)); + .mapToPair(new GenerateSequenceBlock(rowsInBlock, lfrom, lto, lincr)); //step 3: output handling MatrixCharacteristics mcOut = sec.getMatrixCharacteristics(output.getName()); @@ -521,20 +450,21 @@ public class RandSPInstruction extends UnarySPInstruction { private void generateSample(SparkExecutionContext sec) throws DMLRuntimeException { - if ( maxValue < rows && !replace ) + long lrows = sec.getScalarInput(rows).getLongValue(); + if ( maxValue < lrows && !replace ) throw new DMLRuntimeException("Sample (size=" + rows + ") larger than population (size=" + maxValue + ") can only be generated with replacement."); if( LOG.isTraceEnabled() ) - LOG.trace("Process RandSPInstruction sample with range="+ maxValue +", size="+ rows +", replace="+ replace + ", seed=" + seed); + LOG.trace("Process RandSPInstruction sample with range="+ maxValue +", size="+ lrows +", replace="+ replace + ", seed=" + seed); // sampling rate that guarantees a sample of size >= sampleSizeLowerBound 99.99% of the time. - double fraction = SamplingUtils.computeFractionForSampleSize((int)rows, UtilFunctions.toLong(maxValue), replace); + double fraction = SamplingUtils.computeFractionForSampleSize((int)lrows, UtilFunctions.toLong(maxValue), replace); Well1024a bigrand = LibMatrixDatagen.setupSeedsForRand(seed); // divide the population range across numPartitions by creating SampleTasks double hdfsBlockSize = InfrastructureAnalyzer.getHDFSBlockSize(); - long outputSize = MatrixBlock.estimateSizeDenseInMemory(rows,1); + long outputSize = MatrixBlock.estimateSizeDenseInMemory(lrows,1); int numPartitions = (int) Math.ceil((double)outputSize/hdfsBlockSize); long partitionSize = (long) Math.ceil(maxValue/numPartitions); @@ -558,18 +488,16 @@ public class RandSPInstruction extends UnarySPInstruction { // Trim the sampled list to required size & attach matrix indexes to randomized elements JavaPairRDD<MatrixIndexes, MatrixCell> miRDD = randomizedRDD .zipWithIndex() - .filter( new TrimSample(rows) ) - .mapToPair( new Double2MatrixCell() ); + .filter( new TrimSample(lrows) ) + .mapToPair( new Double2MatrixCell() ); - MatrixCharacteristics mcOut = new MatrixCharacteristics(rows, 1, rowsInBlock, colsInBlock, rows); + MatrixCharacteristics mcOut = new MatrixCharacteristics(lrows, 1, rowsInBlock, colsInBlock, lrows); // Construct BinaryBlock representation JavaPairRDD<MatrixIndexes, MatrixBlock> mbRDD = RDDConverterUtils.binaryCellToBinaryBlock(sec.getSparkContext(), miRDD, mcOut, true); - MatrixCharacteristics retDims = sec.getMatrixCharacteristics(output.getName()); - retDims.setNonZeros(rows); - + sec.getMatrixCharacteristics(output.getName()).setNonZeros(lrows); sec.setRDDHandleForVariable(output.getName(), mbRDD); } @@ -830,13 +758,11 @@ public class RandSPInstruction extends UnarySPInstruction { * @param max maximum value * @return */ - private boolean isMemAvail(long lRows, long lCols, double sparsity, double min, double max) - { - double size = (min == 0 && max == 0) ? OptimizerUtils.estimateSizeEmptyBlock(rows, cols): - OptimizerUtils.estimateSizeExactSparsity(rows, cols, sparsity); - - return ( OptimizerUtils.isValidCPDimensions(rows, cols) - && OptimizerUtils.isValidCPMatrixSize(rows, cols, sparsity) + private static boolean isMemAvail(long lrows, long lcols, double sparsity, double min, double max) { + double size = (min == 0 && max == 0) ? OptimizerUtils.estimateSizeEmptyBlock(lrows, lcols): + OptimizerUtils.estimateSizeExactSparsity(lrows, lcols, sparsity); + return ( OptimizerUtils.isValidCPDimensions(lrows, lcols) + && OptimizerUtils.isValidCPMatrixSize(lrows, lcols, sparsity) && size < OptimizerUtils.getLocalMemBudget() ); }
