You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemml.apache.org by lr...@apache.org on 2015/12/03 19:46:10 UTC
[39/78] [abbrv] [partial] incubator-systemml git commit: Move files
to new package folder structure
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/276d9257/src/main/java/com/ibm/bi/dml/hops/QuaternaryOp.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/hops/QuaternaryOp.java b/src/main/java/com/ibm/bi/dml/hops/QuaternaryOp.java
deleted file mode 100644
index fd329f0..0000000
--- a/src/main/java/com/ibm/bi/dml/hops/QuaternaryOp.java
+++ /dev/null
@@ -1,1684 +0,0 @@
-/**
- * (C) Copyright IBM Corp. 2010, 2015
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package com.ibm.bi.dml.hops;
-
-import com.ibm.bi.dml.hops.Hop.MultiThreadedHop;
-import com.ibm.bi.dml.lops.Aggregate;
-import com.ibm.bi.dml.lops.DataPartition;
-import com.ibm.bi.dml.lops.Group;
-import com.ibm.bi.dml.lops.Lop;
-import com.ibm.bi.dml.lops.LopsException;
-import com.ibm.bi.dml.lops.RepMat;
-import com.ibm.bi.dml.lops.Transform;
-import com.ibm.bi.dml.lops.Unary;
-import com.ibm.bi.dml.lops.UnaryCP;
-import com.ibm.bi.dml.lops.LopProperties.ExecType;
-import com.ibm.bi.dml.lops.PartialAggregate.CorrectionLocationType;
-import com.ibm.bi.dml.lops.WeightedCrossEntropy;
-import com.ibm.bi.dml.lops.WeightedCrossEntropyR;
-import com.ibm.bi.dml.lops.WeightedDivMM;
-import com.ibm.bi.dml.lops.WeightedCrossEntropy.WCeMMType;
-import com.ibm.bi.dml.lops.WeightedDivMM.WDivMMType;
-import com.ibm.bi.dml.lops.WeightedDivMMR;
-import com.ibm.bi.dml.lops.WeightedSigmoid;
-import com.ibm.bi.dml.lops.WeightedSigmoid.WSigmoidType;
-import com.ibm.bi.dml.lops.WeightedSigmoidR;
-import com.ibm.bi.dml.lops.WeightedSquaredLoss;
-import com.ibm.bi.dml.lops.WeightedSquaredLoss.WeightsType;
-import com.ibm.bi.dml.lops.WeightedSquaredLossR;
-import com.ibm.bi.dml.lops.WeightedUnaryMM;
-import com.ibm.bi.dml.lops.WeightedUnaryMM.WUMMType;
-import com.ibm.bi.dml.lops.WeightedUnaryMMR;
-import com.ibm.bi.dml.parser.Expression.DataType;
-import com.ibm.bi.dml.parser.Expression.ValueType;
-import com.ibm.bi.dml.runtime.controlprogram.ParForProgramBlock.PDataPartitionFormat;
-import com.ibm.bi.dml.runtime.controlprogram.context.SparkExecutionContext;
-import com.ibm.bi.dml.runtime.matrix.MatrixCharacteristics;
-import com.ibm.bi.dml.runtime.matrix.mapred.DistributedCacheInput;
-
-/**
- * Note: this hop should be called AggQuaternaryOp in consistency with AggUnaryOp and AggBinaryOp;
- * however, since there does not exist a real QuaternaryOp yet - we can leave it as is for now.
- */
-public class QuaternaryOp extends Hop implements MultiThreadedHop
-{
-
- //config influencing mr operator selection (for testing purposes only)
- public static boolean FORCE_REPLICATION = false;
-
- private OpOp4 _op = null;
- private int _maxNumThreads = -1; //-1 for unlimited
-
- //wsloss-specific attributes
- private boolean _postWeights = false;
-
- //wsigmoid-specific attributes
- private boolean _logout = false;
- private boolean _minusin = false;
-
- //wdivmm-specific attributes
- private int _baseType = -1;
- private boolean _mult = false;
- private boolean _minus = false;
-
- //wumm-specific attributes
- private boolean _umult = false;
- private OpOp1 _uop = null;
- private OpOp2 _sop = null;
-
- private QuaternaryOp() {
- //default constructor for clone
- }
-
- /**
- * Constructor for wsloss.
- *
- * @param l
- * @param dt
- * @param vt
- * @param o
- * @param inX
- * @param inU
- * @param inV
- * @param inW
- * @param post
- */
- public QuaternaryOp(String l, DataType dt, ValueType vt, Hop.OpOp4 o,
- Hop inX, Hop inU, Hop inV, Hop inW, boolean post)
- {
- this(l, dt, vt, o, inX, inU, inV);
- getInput().add(3, inW);
- inW.getParent().add(this);
-
- _postWeights = post;
- }
-
- /**
- * Constructor for wsigmoid.
- *
- * @param l
- * @param dt
- * @param vt
- * @param o
- * @param inX
- * @param inU
- * @param inV
- * @param logout
- * @param minusin
- */
- public QuaternaryOp(String l, DataType dt, ValueType vt, Hop.OpOp4 o,
- Hop inX, Hop inU, Hop inV, boolean flag1, boolean flag2)
- {
- this(l, dt, vt, o, inX, inU, inV);
-
- _logout = flag1;
- _minusin = flag2;
- }
-
- public QuaternaryOp(String l, DataType dt, ValueType vt, Hop.OpOp4 o,
- Hop inW, Hop inU, Hop inV, int baseType, boolean flag1, boolean flag2)
- {
- this(l, dt, vt, o, inW, inU, inV);
-
- _baseType = baseType;
- _mult = flag1;
- _minus = flag2;
- }
-
- public QuaternaryOp(String l, DataType dt, ValueType vt, Hop.OpOp4 o,
- Hop inW, Hop inU, Hop inV, boolean umult, OpOp1 uop, OpOp2 sop)
- {
- this(l, dt, vt, o, inW, inU, inV);
-
- _umult = umult;
- _uop = uop;
- _sop = sop;
- }
-
- /**
- *
- * @param l
- * @param dt
- * @param vt
- * @param o
- * @param inX
- * @param inU
- * @param inV
- */
- public QuaternaryOp(String l, DataType dt, ValueType vt, Hop.OpOp4 o, Hop inX, Hop inU, Hop inV)
- {
- super(l, dt, vt);
- _op = o;
- getInput().add(0, inX);
- getInput().add(1, inU);
- getInput().add(2, inV);
- inX.getParent().add(this);
- inU.getParent().add(this);
- inV.getParent().add(this);
- }
-
- public OpOp4 getOp(){
- return _op;
- }
-
- @Override
- public void setMaxNumThreads( int k ) {
- _maxNumThreads = k;
- }
-
- @Override
- public int getMaxNumThreads() {
- return _maxNumThreads;
- }
-
- @Override
- public Lop constructLops()
- throws HopsException, LopsException
- {
- //return already created lops
- if( getLops() != null )
- return getLops();
-
- try
- {
- ExecType et = optFindExecType();
-
- switch( _op ) {
- case WSLOSS: {
- WeightsType wtype = checkWeightsType();
-
- if( et == ExecType.CP )
- constructCPLopsWeightedSquaredLoss(wtype);
- else if( et == ExecType.MR )
- constructMRLopsWeightedSquaredLoss(wtype);
- else if( et == ExecType.SPARK )
- constructSparkLopsWeightedSquaredLoss(wtype);
- else
- throw new HopsException("Unsupported quaternaryop-wsloss exec type: "+et);
- break;
- }
-
- case WSIGMOID:{
- WSigmoidType wtype = checkWSigmoidType();
-
- if( et == ExecType.CP )
- constructCPLopsWeightedSigmoid(wtype);
- else if( et == ExecType.MR )
- constructMRLopsWeightedSigmoid(wtype);
- else if( et == ExecType.SPARK )
- constructSparkLopsWeightedSigmoid(wtype);
- else
- throw new HopsException("Unsupported quaternaryop-wsigmoid exec type: "+et);
- break;
- }
-
- case WDIVMM:{
- WDivMMType wtype = checkWDivMMType();
- if( et == ExecType.CP )
- constructCPLopsWeightedDivMM(wtype);
- else if( et == ExecType.MR )
- constructMRLopsWeightedDivMM(wtype);
- else if( et == ExecType.SPARK )
- constructSparkLopsWeightedDivMM(wtype);
- else
- throw new HopsException("Unsupported quaternaryop-wdivmm exec type: "+et);
- break;
- }
-
- case WCEMM:{
- WCeMMType wtype = WCeMMType.BASIC;
-
- if( et == ExecType.CP )
- constructCPLopsWeightedCeMM(wtype);
- else if( et == ExecType.MR )
- constructMRLopsWeightedCeMM(wtype);
- else if( et == ExecType.SPARK )
- constructSparkLopsWeightedCeMM(wtype);
- else
- throw new HopsException("Unsupported quaternaryop-wcemm exec type: "+et);
- break;
- }
-
- case WUMM:{
- WUMMType wtype = _umult ? WUMMType.MULT : WUMMType.DIV;
-
- if( et == ExecType.CP )
- constructCPLopsWeightedUMM(wtype);
- else if( et == ExecType.MR )
- constructMRLopsWeightedUMM(wtype);
- else if( et == ExecType.SPARK )
- constructSparkLopsWeightedUMM(wtype);
- else
- throw new HopsException("Unsupported quaternaryop-wumm exec type: "+et);
- break;
- }
-
- default:
- throw new HopsException(this.printErrorLocation() + "Unknown QuaternaryOp (" + _op + ") while constructing Lops");
- }
- }
- catch(LopsException e) {
- throw new HopsException(this.printErrorLocation() + "error constructing lops for QuaternaryOp." , e);
- }
-
- //add reblock/checkpoint lops if necessary
- constructAndSetLopsDataFlowProperties();
-
- return getLops();
- }
-
- @Override
- public String getOpString() {
- String s = new String("");
- s += "q(" + HopsOpOp4String.get(_op) + ")";
- return s;
- }
-
- public void printMe() throws HopsException {
- if (LOG.isDebugEnabled()){
- if (getVisited() != VisitStatus.DONE) {
- super.printMe();
- LOG.debug(" Operation: " + _op);
- for (Hop h : getInput()) {
- h.printMe();
- }
- }
- setVisited(VisitStatus.DONE);
- }
- }
-
- @Override
- public boolean allowsAllExecTypes()
- {
- return true;
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructCPLopsWeightedSquaredLoss(WeightsType wtype)
- throws HopsException, LopsException
- {
- WeightedSquaredLoss wsloss = new WeightedSquaredLoss(
- getInput().get(0).constructLops(),
- getInput().get(1).constructLops(),
- getInput().get(2).constructLops(),
- getInput().get(3).constructLops(),
- getDataType(), getValueType(), wtype, ExecType.CP);
-
- //set degree of parallelism
- int k = OptimizerUtils.getConstrainedNumThreads(_maxNumThreads);
- wsloss.setNumThreads(k);
-
- setOutputDimensions( wsloss );
- setLineNumbers( wsloss );
- setLops( wsloss );
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructMRLopsWeightedSquaredLoss(WeightsType wtype)
- throws HopsException, LopsException
- {
- //NOTE: the common case for wsloss are factors U/V with a rank of 10s to 100s; the current runtime only
- //supports single block outer products (U/V rank <= blocksize, i.e., 1000 by default); we enforce this
- //by applying the hop rewrite for Weighted Squared Loss only if this constraint holds.
-
- Hop X = getInput().get(0);
- Hop U = getInput().get(1);
- Hop V = getInput().get(2);
- Hop W = getInput().get(3);
-
- //MR operator selection, part1
- double m1Size = OptimizerUtils.estimateSize(U.getDim1(), U.getDim2()); //size U
- double m2Size = OptimizerUtils.estimateSize(V.getDim1(), V.getDim2()); //size V
- boolean isMapWsloss = (!wtype.hasFourInputs() && m1Size+m2Size < OptimizerUtils.getRemoteMemBudgetMap(true));
-
- if( !FORCE_REPLICATION && isMapWsloss ) //broadcast
- {
- //partitioning of U
- boolean needPartU = !U.dimsKnown() || U.getDim1() * U.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- Lop lU = U.constructLops();
- if( needPartU ){ //requires partitioning
- lU = new DataPartition(lU, DataType.MATRIX, ValueType.DOUBLE, (m1Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), getRowsInBlock(), getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
- }
-
- //partitioning of V
- boolean needPartV = !V.dimsKnown() || V.getDim1() * V.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- Lop lV = V.constructLops();
- if( needPartV ){ //requires partitioning
- lV = new DataPartition(lV, DataType.MATRIX, ValueType.DOUBLE, (m2Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lV.getOutputParameters().setDimensions(V.getDim1(), V.getDim2(), getRowsInBlock(), getColsInBlock(), V.getNnz());
- setLineNumbers(lV);
- }
-
- //map-side wsloss always with broadcast
- Lop wsloss = new WeightedSquaredLoss( X.constructLops(), lU, lV, W.constructLops(),
- DataType.MATRIX, ValueType.DOUBLE, wtype, ExecType.MR);
- wsloss.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(wsloss);
-
- Group grp = new Group(wsloss, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grp.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(grp);
-
- Aggregate agg1 = new Aggregate(grp, HopsAgg2Lops.get(AggOp.SUM), DataType.MATRIX, ValueType.DOUBLE, ExecType.MR);
- agg1.setupCorrectionLocation(CorrectionLocationType.NONE); // aggregation uses kahanSum
- agg1.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(agg1);
-
- UnaryCP unary1 = new UnaryCP(agg1, HopsOpOp1LopsUS.get(OpOp1.CAST_AS_SCALAR), getDataType(), getValueType());
- unary1.getOutputParameters().setDimensions(0, 0, 0, 0, -1);
- setLineNumbers(unary1);
- setLops(unary1);
- }
- else //general case
- {
- //MR operator selection part 2
- boolean cacheU = !FORCE_REPLICATION && (m1Size < OptimizerUtils.getRemoteMemBudgetReduce());
- boolean cacheV = !FORCE_REPLICATION && ((!cacheU && m2Size < OptimizerUtils.getRemoteMemBudgetReduce())
- || (cacheU && m1Size+m2Size < OptimizerUtils.getRemoteMemBudgetReduce()));
-
- Group grpX = new Group(X.constructLops(), Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpX.getOutputParameters().setDimensions(X.getDim1(), X.getDim2(), X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(grpX);
-
- Lop grpW = W.constructLops();
- if( grpW.getDataType()==DataType.MATRIX ) {
- grpW = new Group(W.constructLops(), Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpW.getOutputParameters().setDimensions(W.getDim1(), W.getDim2(), W.getRowsInBlock(), W.getColsInBlock(), -1);
- setLineNumbers(grpW);
- }
-
- Lop lU = null;
- if( cacheU ) {
- //partitioning of U for read through distributed cache
- boolean needPartU = !U.dimsKnown() || U.getDim1() * U.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- lU = U.constructLops();
- if( needPartU ){ //requires partitioning
- lU = new DataPartition(lU, DataType.MATRIX, ValueType.DOUBLE, (m1Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), getRowsInBlock(), getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
- }
- }
- else {
- //replication of U for shuffle to target block
- Lop offset = createOffsetLop(V, false); //ncol of t(V) -> nrow of V determines num replicates
- lU = new RepMat(U.constructLops(), offset, true, V.getDataType(), V.getValueType());
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(),
- U.getRowsInBlock(), U.getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
-
- Group grpU = new Group(lU, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), U.getRowsInBlock(), U.getColsInBlock(), -1);
- setLineNumbers(grpU);
- lU = grpU;
- }
-
- Lop lV = null;
- if( cacheV ) {
- //partitioning of V for read through distributed cache
- boolean needPartV = !V.dimsKnown() || V.getDim1() * V.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- lV = V.constructLops();
- if( needPartV ){ //requires partitioning
- lV = new DataPartition(lV, DataType.MATRIX, ValueType.DOUBLE, (m2Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lV.getOutputParameters().setDimensions(V.getDim1(), V.getDim2(), getRowsInBlock(), getColsInBlock(), V.getNnz());
- setLineNumbers(lV);
- }
- }
- else {
- //replication of t(V) for shuffle to target block
- Transform ltV = new Transform( V.constructLops(), HopsTransf2Lops.get(ReOrgOp.TRANSPOSE), getDataType(), getValueType(), ExecType.MR);
- ltV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(),
- V.getColsInBlock(), V.getRowsInBlock(), V.getNnz());
- setLineNumbers(ltV);
-
- Lop offset = createOffsetLop(U, false); //nrow of U determines num replicates
- lV = new RepMat(ltV, offset, false, V.getDataType(), V.getValueType());
- lV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(),
- V.getColsInBlock(), V.getRowsInBlock(), V.getNnz());
- setLineNumbers(lV);
-
- Group grpV = new Group(lV, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(), V.getColsInBlock(), V.getRowsInBlock(), -1);
- setLineNumbers(grpV);
- lV = grpV;
- }
-
- //reduce-side wsloss w/ or without broadcast
- Lop wsloss = new WeightedSquaredLossR(
- grpX, lU, lV, grpW, DataType.MATRIX, ValueType.DOUBLE, wtype, cacheU, cacheV, ExecType.MR);
- wsloss.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(wsloss);
-
- Group grp = new Group(wsloss, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grp.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(grp);
-
- Aggregate agg1 = new Aggregate(grp, HopsAgg2Lops.get(AggOp.SUM), DataType.MATRIX, ValueType.DOUBLE, ExecType.MR);
- agg1.setupCorrectionLocation(CorrectionLocationType.NONE); // aggregation uses kahanSum
- agg1.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(agg1);
-
- UnaryCP unary1 = new UnaryCP(agg1, HopsOpOp1LopsUS.get(OpOp1.CAST_AS_SCALAR), getDataType(), getValueType());
- unary1.getOutputParameters().setDimensions(0, 0, 0, 0, -1);
- setLineNumbers(unary1);
- setLops(unary1);
- }
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructSparkLopsWeightedSquaredLoss(WeightsType wtype)
- throws HopsException, LopsException
- {
- //NOTE: the common case for wsloss are factors U/V with a rank of 10s to 100s; the current runtime only
- //supports single block outer products (U/V rank <= blocksize, i.e., 1000 by default); we enforce this
- //by applying the hop rewrite for Weighted Squared Loss only if this constraint holds.
-
- //Notes: Any broadcast needs to fit twice in local memory because we partition the input in cp,
- //and needs to fit once in executor broadcast memory. The 2GB broadcast constraint is no longer
- //required because the max_int byte buffer constraint has been fixed in Spark 1.4
- double memBudgetExec = SparkExecutionContext.getBroadcastMemoryBudget();
- double memBudgetLocal = OptimizerUtils.getLocalMemBudget();
-
- Hop X = getInput().get(0);
- Hop U = getInput().get(1);
- Hop V = getInput().get(2);
- Hop W = getInput().get(3);
-
- //MR operator selection, part1
- double m1Size = OptimizerUtils.estimateSize(U.getDim1(), U.getDim2()); //size U
- double m2Size = OptimizerUtils.estimateSize(V.getDim1(), V.getDim2()); //size V
- boolean isMapWsloss = (!wtype.hasFourInputs() && m1Size+m2Size < memBudgetExec
- && 2*m1Size < memBudgetLocal && 2*m2Size < memBudgetLocal);
-
- if( !FORCE_REPLICATION && isMapWsloss ) //broadcast
- {
- //map-side wsloss always with broadcast
- Lop wsloss = new WeightedSquaredLoss( X.constructLops(), U.constructLops(), V.constructLops(), W.constructLops(),
- DataType.SCALAR, ValueType.DOUBLE, wtype, ExecType.SPARK);
- setOutputDimensions(wsloss);
- setLineNumbers(wsloss);
- setLops(wsloss);
- }
- else //general case
- {
- //MR operator selection part 2
- boolean cacheU = !FORCE_REPLICATION && (m1Size < memBudgetExec && 2*m1Size < memBudgetLocal);
- boolean cacheV = !FORCE_REPLICATION && ((!cacheU && m2Size < memBudgetExec )
- || (cacheU && m1Size+m2Size < memBudgetExec)) && 2*m2Size < memBudgetLocal;
-
- //reduce-side wsloss w/ or without broadcast
- Lop wsloss = new WeightedSquaredLossR(
- X.constructLops(), U.constructLops(), V.constructLops(), W.constructLops(),
- DataType.SCALAR, ValueType.DOUBLE, wtype, cacheU, cacheV, ExecType.SPARK);
- setOutputDimensions(wsloss);
- setLineNumbers(wsloss);
- setLops(wsloss);
- }
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructCPLopsWeightedSigmoid(WSigmoidType wtype)
- throws HopsException, LopsException
- {
- WeightedSigmoid wsig = new WeightedSigmoid(
- getInput().get(0).constructLops(),
- getInput().get(1).constructLops(),
- getInput().get(2).constructLops(),
- getDataType(), getValueType(), wtype, ExecType.CP);
-
- //set degree of parallelism
- int k = OptimizerUtils.getConstrainedNumThreads(_maxNumThreads);
- wsig.setNumThreads(k);
-
- setOutputDimensions( wsig );
- setLineNumbers( wsig );
- setLops( wsig );
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructMRLopsWeightedSigmoid( WSigmoidType wtype )
- throws HopsException, LopsException
- {
- //NOTE: the common case for wsigmoid are factors U/V with a rank of 10s to 100s; the current runtime only
- //supports single block outer products (U/V rank <= blocksize, i.e., 1000 by default); we enforce this
- //by applying the hop rewrite for Weighted Squared Loss only if this constraint holds.
-
- Hop X = getInput().get(0);
- Hop U = getInput().get(1);
- Hop V = getInput().get(2);
-
- //MR operator selection, part1
- double m1Size = OptimizerUtils.estimateSize(U.getDim1(), U.getDim2()); //size U
- double m2Size = OptimizerUtils.estimateSize(V.getDim1(), V.getDim2()); //size V
- boolean isMapWsloss = (m1Size+m2Size < OptimizerUtils.getRemoteMemBudgetMap(true));
-
- if( !FORCE_REPLICATION && isMapWsloss ) //broadcast
- {
- //partitioning of U
- boolean needPartU = !U.dimsKnown() || U.getDim1() * U.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- Lop lU = U.constructLops();
- if( needPartU ){ //requires partitioning
- lU = new DataPartition(lU, DataType.MATRIX, ValueType.DOUBLE, (m1Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), getRowsInBlock(), getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
- }
-
- //partitioning of V
- boolean needPartV = !V.dimsKnown() || V.getDim1() * V.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- Lop lV = V.constructLops();
- if( needPartV ){ //requires partitioning
- lV = new DataPartition(lV, DataType.MATRIX, ValueType.DOUBLE, (m2Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lV.getOutputParameters().setDimensions(V.getDim1(), V.getDim2(), getRowsInBlock(), getColsInBlock(), V.getNnz());
- setLineNumbers(lV);
- }
-
- //map-side wsloss always with broadcast
- Lop wsigmoid = new WeightedSigmoid( X.constructLops(), lU, lV,
- DataType.MATRIX, ValueType.DOUBLE, wtype, ExecType.MR);
- setOutputDimensions(wsigmoid);
- setLineNumbers(wsigmoid);
- setLops( wsigmoid );
-
- //in contrast to wsloss no aggregation required
- }
- else //general case
- {
- //MR operator selection part 2
- boolean cacheU = !FORCE_REPLICATION && (m1Size < OptimizerUtils.getRemoteMemBudgetReduce());
- boolean cacheV = !FORCE_REPLICATION && ((!cacheU && m2Size < OptimizerUtils.getRemoteMemBudgetReduce())
- || (cacheU && m1Size+m2Size < OptimizerUtils.getRemoteMemBudgetReduce()));
-
- Group grpX = new Group(X.constructLops(), Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpX.getOutputParameters().setDimensions(X.getDim1(), X.getDim2(), X.getRowsInBlock(), X.getColsInBlock(), X.getNnz());
- setLineNumbers(grpX);
-
- Lop lU = null;
- if( cacheU ) {
- //partitioning of U for read through distributed cache
- boolean needPartU = !U.dimsKnown() || U.getDim1() * U.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- lU = U.constructLops();
- if( needPartU ){ //requires partitioning
- lU = new DataPartition(lU, DataType.MATRIX, ValueType.DOUBLE, (m1Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), getRowsInBlock(), getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
- }
- }
- else {
- //replication of U for shuffle to target block
- Lop offset = createOffsetLop(V, false); //ncol of t(V) -> nrow of V determines num replicates
- lU = new RepMat(U.constructLops(), offset, true, V.getDataType(), V.getValueType());
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(),
- U.getRowsInBlock(), U.getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
-
- Group grpU = new Group(lU, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), U.getRowsInBlock(), U.getColsInBlock(), -1);
- setLineNumbers(grpU);
- lU = grpU;
- }
-
- Lop lV = null;
- if( cacheV ) {
- //partitioning of V for read through distributed cache
- boolean needPartV = !V.dimsKnown() || V.getDim1() * V.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- lV = V.constructLops();
- if( needPartV ){ //requires partitioning
- lV = new DataPartition(lV, DataType.MATRIX, ValueType.DOUBLE, (m2Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lV.getOutputParameters().setDimensions(V.getDim1(), V.getDim2(), getRowsInBlock(), getColsInBlock(), V.getNnz());
- setLineNumbers(lV);
- }
- }
- else {
- //replication of t(V) for shuffle to target block
- Transform ltV = new Transform( V.constructLops(), HopsTransf2Lops.get(ReOrgOp.TRANSPOSE), getDataType(), getValueType(), ExecType.MR);
- ltV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(),
- V.getColsInBlock(), V.getRowsInBlock(), V.getNnz());
- setLineNumbers(ltV);
-
- Lop offset = createOffsetLop(U, false); //nrow of U determines num replicates
- lV = new RepMat(ltV, offset, false, V.getDataType(), V.getValueType());
- lV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(),
- V.getColsInBlock(), V.getRowsInBlock(), V.getNnz());
- setLineNumbers(lV);
-
- Group grpV = new Group(lV, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(), V.getColsInBlock(), V.getRowsInBlock(), -1);
- setLineNumbers(grpV);
- lV = grpV;
- }
-
- //reduce-side wsloss w/ or without broadcast
- Lop wsigmoid = new WeightedSigmoidR(
- grpX, lU, lV, DataType.MATRIX, ValueType.DOUBLE, wtype, cacheU, cacheV, ExecType.MR);
- setOutputDimensions(wsigmoid);
- setLineNumbers(wsigmoid);
- setLops(wsigmoid);
-
- //in contrast to wsloss no aggregation required
- }
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructSparkLopsWeightedSigmoid( WSigmoidType wtype )
- throws HopsException, LopsException
- {
- //NOTE: the common case for wsigmoid are factors U/V with a rank of 10s to 100s; the current runtime only
- //supports single block outer products (U/V rank <= blocksize, i.e., 1000 by default); we enforce this
- //by applying the hop rewrite for Weighted Squared Loss only if this constraint holds.
-
- //Notes: Any broadcast needs to fit twice in local memory because we partition the input in cp,
- //and needs to fit once in executor broadcast memory. The 2GB broadcast constraint is no longer
- //required because the max_int byte buffer constraint has been fixed in Spark 1.4
- double memBudgetExec = SparkExecutionContext.getBroadcastMemoryBudget();
- double memBudgetLocal = OptimizerUtils.getLocalMemBudget();
-
- Hop X = getInput().get(0);
- Hop U = getInput().get(1);
- Hop V = getInput().get(2);
-
- //MR operator selection, part1
- double m1Size = OptimizerUtils.estimateSize(U.getDim1(), U.getDim2()); //size U
- double m2Size = OptimizerUtils.estimateSize(V.getDim1(), V.getDim2()); //size V
- boolean isMapWsloss = (m1Size+m2Size < memBudgetExec
- && 2*m1Size<memBudgetLocal && 2*m2Size<memBudgetLocal);
-
- if( !FORCE_REPLICATION && isMapWsloss ) //broadcast
- {
- //map-side wsloss always with broadcast
- Lop wsigmoid = new WeightedSigmoid( X.constructLops(), U.constructLops(), V.constructLops(),
- DataType.MATRIX, ValueType.DOUBLE, wtype, ExecType.SPARK);
- setOutputDimensions(wsigmoid);
- setLineNumbers(wsigmoid);
- setLops( wsigmoid );
- }
- else //general case
- {
- //MR operator selection part 2
- boolean cacheU = !FORCE_REPLICATION && (m1Size < memBudgetExec && 2*m1Size < memBudgetLocal);
- boolean cacheV = !FORCE_REPLICATION && ((!cacheU && m2Size < memBudgetExec )
- || (cacheU && m1Size+m2Size < memBudgetExec)) && 2*m2Size < memBudgetLocal;
-
- //reduce-side wsloss w/ or without broadcast
- Lop wsigmoid = new WeightedSigmoidR(
- X.constructLops(), U.constructLops(), V.constructLops(),
- DataType.MATRIX, ValueType.DOUBLE, wtype, cacheU, cacheV, ExecType.SPARK);
- setOutputDimensions(wsigmoid);
- setLineNumbers(wsigmoid);
- setLops(wsigmoid);
- }
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructCPLopsWeightedDivMM(WDivMMType wtype)
- throws HopsException, LopsException
- {
- WeightedDivMM wdiv = new WeightedDivMM(
- getInput().get(0).constructLops(),
- getInput().get(1).constructLops(),
- getInput().get(2).constructLops(),
- getDataType(), getValueType(), wtype, ExecType.CP);
-
- //set degree of parallelism
- int k = OptimizerUtils.getConstrainedNumThreads(_maxNumThreads);
- wdiv.setNumThreads(k);
-
- setOutputDimensions( wdiv );
- setLineNumbers( wdiv );
- setLops( wdiv );
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructMRLopsWeightedDivMM( WDivMMType wtype )
- throws HopsException, LopsException
- {
- //NOTE: the common case for wdivmm are factors U/V with a rank of 10s to 100s; the current runtime only
- //supports single block outer products (U/V rank <= blocksize, i.e., 1000 by default); we enforce this
- //by applying the hop rewrite for Weighted Squared Loss only if this constraint holds.
-
- Hop W = getInput().get(0);
- Hop U = getInput().get(1);
- Hop V = getInput().get(2);
-
- //MR operator selection, part1
- double m1Size = OptimizerUtils.estimateSize(U.getDim1(), U.getDim2()); //size U
- double m2Size = OptimizerUtils.estimateSize(V.getDim1(), V.getDim2()); //size V
- boolean isMapWsloss = (m1Size+m2Size < OptimizerUtils.getRemoteMemBudgetMap(true));
-
- if( !FORCE_REPLICATION && isMapWsloss ) //broadcast
- {
- //partitioning of U
- boolean needPartU = !U.dimsKnown() || U.getDim1() * U.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- Lop lU = U.constructLops();
- if( needPartU ){ //requires partitioning
- lU = new DataPartition(lU, DataType.MATRIX, ValueType.DOUBLE, (m1Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), getRowsInBlock(), getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
- }
-
- //partitioning of V
- boolean needPartV = !V.dimsKnown() || V.getDim1() * V.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- Lop lV = V.constructLops();
- if( needPartV ){ //requires partitioning
- lV = new DataPartition(lV, DataType.MATRIX, ValueType.DOUBLE, (m2Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lV.getOutputParameters().setDimensions(V.getDim1(), V.getDim2(), getRowsInBlock(), getColsInBlock(), V.getNnz());
- setLineNumbers(lV);
- }
-
- //map-side wsloss always with broadcast
- Lop wdivmm = new WeightedDivMM( W.constructLops(), lU, lV,
- DataType.MATRIX, ValueType.DOUBLE, wtype, ExecType.MR);
- setOutputDimensions(wdivmm);
- setLineNumbers(wdivmm);
- setLops(wdivmm);
- }
- else //general case
- {
- //MR operator selection part 2 (both cannot happen for wdivmm, otherwise mapwdivmm)
- boolean cacheU = !FORCE_REPLICATION && (m1Size < OptimizerUtils.getRemoteMemBudgetReduce());
- boolean cacheV = !FORCE_REPLICATION && ((!cacheU && m2Size < OptimizerUtils.getRemoteMemBudgetReduce())
- || (cacheU && m1Size+m2Size < OptimizerUtils.getRemoteMemBudgetReduce()));
-
- Group grpW = new Group(W.constructLops(), Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpW.getOutputParameters().setDimensions(W.getDim1(), W.getDim2(), W.getRowsInBlock(), W.getColsInBlock(), W.getNnz());
- setLineNumbers(grpW);
-
- Lop lU = null;
- if( cacheU ) {
- //partitioning of U for read through distributed cache
- boolean needPartU = !U.dimsKnown() || U.getDim1() * U.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- lU = U.constructLops();
- if( needPartU ){ //requires partitioning
- lU = new DataPartition(lU, DataType.MATRIX, ValueType.DOUBLE, (m1Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), getRowsInBlock(), getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
- }
- }
- else {
- //replication of U for shuffle to target block
- Lop offset = createOffsetLop(V, false); //ncol of t(V) -> nrow of V determines num replicates
- lU = new RepMat(U.constructLops(), offset, true, V.getDataType(), V.getValueType());
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(),
- U.getRowsInBlock(), U.getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
-
- Group grpU = new Group(lU, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), U.getRowsInBlock(), U.getColsInBlock(), -1);
- setLineNumbers(grpU);
- lU = grpU;
- }
-
- Lop lV = null;
- if( cacheV ) {
- //partitioning of V for read through distributed cache
- boolean needPartV = !V.dimsKnown() || V.getDim1() * V.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- lV = V.constructLops();
- if( needPartV ){ //requires partitioning
- lV = new DataPartition(lV, DataType.MATRIX, ValueType.DOUBLE, (m2Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lV.getOutputParameters().setDimensions(V.getDim1(), V.getDim2(), getRowsInBlock(), getColsInBlock(), V.getNnz());
- setLineNumbers(lV);
- }
- }
- else {
- //replication of t(V) for shuffle to target block
- Transform ltV = new Transform( V.constructLops(), HopsTransf2Lops.get(ReOrgOp.TRANSPOSE), getDataType(), getValueType(), ExecType.MR);
- ltV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(),
- V.getColsInBlock(), V.getRowsInBlock(), V.getNnz());
- setLineNumbers(ltV);
-
- Lop offset = createOffsetLop(U, false); //nrow of U determines num replicates
- lV = new RepMat(ltV, offset, false, V.getDataType(), V.getValueType());
- lV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(),
- V.getColsInBlock(), V.getRowsInBlock(), V.getNnz());
- setLineNumbers(lV);
-
- Group grpV = new Group(lV, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(), V.getColsInBlock(), V.getRowsInBlock(), -1);
- setLineNumbers(grpV);
- lV = grpV;
- }
-
- //reduce-side wsloss w/ or without broadcast
- Lop wdivmm = new WeightedDivMMR( grpW, lU, lV,
- DataType.MATRIX, ValueType.DOUBLE, wtype, cacheU, cacheV, ExecType.MR);
- setOutputDimensions(wdivmm);
- setLineNumbers(wdivmm);
- setLops(wdivmm);
- }
-
- //in contrast to to wsloss/wsigmoid, wdivmm requires partial aggregation (for the final mm)
- Group grp = new Group(getLops(), Group.OperationTypes.Sort, getDataType(), getValueType());
- setOutputDimensions(grp);
- setLineNumbers(grp);
-
- Aggregate agg1 = new Aggregate(grp, HopsAgg2Lops.get(AggOp.SUM), getDataType(), getValueType(), ExecType.MR);
- // aggregation uses kahanSum but the inputs do not have correction values
- agg1.setupCorrectionLocation(CorrectionLocationType.NONE);
- setOutputDimensions(agg1);
- setLineNumbers(agg1);
-
- setLops(agg1);
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructSparkLopsWeightedDivMM( WDivMMType wtype )
- throws HopsException, LopsException
- {
- //NOTE: the common case for wdivmm are factors U/V with a rank of 10s to 100s; the current runtime only
- //supports single block outer products (U/V rank <= blocksize, i.e., 1000 by default); we enforce this
- //by applying the hop rewrite for Weighted DivMM only if this constraint holds.
-
- //Notes: Any broadcast needs to fit twice in local memory because we partition the input in cp,
- //and needs to fit once in executor broadcast memory. The 2GB broadcast constraint is no longer
- //required because the max_int byte buffer constraint has been fixed in Spark 1.4
- double memBudgetExec = SparkExecutionContext.getBroadcastMemoryBudget();
- double memBudgetLocal = OptimizerUtils.getLocalMemBudget();
-
- Hop W = getInput().get(0);
- Hop U = getInput().get(1);
- Hop V = getInput().get(2);
-
- //MR operator selection, part1
- double m1Size = OptimizerUtils.estimateSize(U.getDim1(), U.getDim2()); //size U
- double m2Size = OptimizerUtils.estimateSize(V.getDim1(), V.getDim2()); //size V
- boolean isMapWsloss = (m1Size+m2Size < memBudgetExec
- && 2*m1Size<memBudgetLocal && 2*m2Size<memBudgetLocal);
-
- if( !FORCE_REPLICATION && isMapWsloss ) //broadcast
- {
- //map-side wsloss always with broadcast
- Lop wdivmm = new WeightedDivMM( W.constructLops(), U.constructLops(), V.constructLops(),
- DataType.MATRIX, ValueType.DOUBLE, wtype, ExecType.SPARK);
- setOutputDimensions(wdivmm);
- setLineNumbers(wdivmm);
- setLops( wdivmm );
- }
- else //general case
- {
- //MR operator selection part 2
- boolean cacheU = !FORCE_REPLICATION && (m1Size < memBudgetExec && 2*m1Size < memBudgetLocal);
- boolean cacheV = !FORCE_REPLICATION && ((!cacheU && m2Size < memBudgetExec )
- || (cacheU && m1Size+m2Size < memBudgetExec)) && 2*m2Size < memBudgetLocal;
-
- //reduce-side wsloss w/ or without broadcast
- Lop wdivmm = new WeightedDivMMR(
- W.constructLops(), U.constructLops(), V.constructLops(),
- DataType.MATRIX, ValueType.DOUBLE, wtype, cacheU, cacheV, ExecType.SPARK);
- setOutputDimensions(wdivmm);
- setLineNumbers(wdivmm);
- setLops(wdivmm);
- }
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructCPLopsWeightedCeMM(WCeMMType wtype)
- throws HopsException, LopsException
- {
- WeightedCrossEntropy wcemm = new WeightedCrossEntropy(
- getInput().get(0).constructLops(),
- getInput().get(1).constructLops(),
- getInput().get(2).constructLops(),
- getDataType(), getValueType(), wtype, ExecType.CP);
-
- //set degree of parallelism
- int k = OptimizerUtils.getConstrainedNumThreads(_maxNumThreads);
- wcemm.setNumThreads(k);
-
- setOutputDimensions( wcemm );
- setLineNumbers( wcemm );
- setLops( wcemm );
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructMRLopsWeightedCeMM(WCeMMType wtype)
- throws HopsException, LopsException
- {
- //NOTE: the common case for wsloss are factors U/V with a rank of 10s to 100s; the current runtime only
- //supports single block outer products (U/V rank <= blocksize, i.e., 1000 by default); we enforce this
- //by applying the hop rewrite for Weighted Squared Loss only if this constraint holds.
-
- Hop X = getInput().get(0);
- Hop U = getInput().get(1);
- Hop V = getInput().get(2);
-
- //MR operator selection, part1
- double m1Size = OptimizerUtils.estimateSize(U.getDim1(), U.getDim2()); //size U
- double m2Size = OptimizerUtils.estimateSize(V.getDim1(), V.getDim2()); //size V
- boolean isMapWsloss = (m1Size+m2Size < OptimizerUtils.getRemoteMemBudgetMap(true));
-
- if( !FORCE_REPLICATION && isMapWsloss ) //broadcast
- {
- //partitioning of U
- boolean needPartU = !U.dimsKnown() || U.getDim1() * U.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- Lop lU = U.constructLops();
- if( needPartU ){ //requires partitioning
- lU = new DataPartition(lU, DataType.MATRIX, ValueType.DOUBLE, (m1Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), getRowsInBlock(), getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
- }
-
- //partitioning of V
- boolean needPartV = !V.dimsKnown() || V.getDim1() * V.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- Lop lV = V.constructLops();
- if( needPartV ){ //requires partitioning
- lV = new DataPartition(lV, DataType.MATRIX, ValueType.DOUBLE, (m2Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lV.getOutputParameters().setDimensions(V.getDim1(), V.getDim2(), getRowsInBlock(), getColsInBlock(), V.getNnz());
- setLineNumbers(lV);
- }
-
- //map-side wsloss always with broadcast
- Lop wcemm = new WeightedCrossEntropy( X.constructLops(), lU, lV,
- DataType.MATRIX, ValueType.DOUBLE, wtype, ExecType.MR);
- wcemm.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(wcemm);
-
- Group grp = new Group(wcemm, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grp.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(grp);
-
- Aggregate agg1 = new Aggregate(grp, HopsAgg2Lops.get(AggOp.SUM), DataType.MATRIX, ValueType.DOUBLE, ExecType.MR);
- agg1.setupCorrectionLocation(CorrectionLocationType.NONE); // aggregation uses kahanSum
- agg1.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(agg1);
-
- UnaryCP unary1 = new UnaryCP(agg1, HopsOpOp1LopsUS.get(OpOp1.CAST_AS_SCALAR), getDataType(), getValueType());
- unary1.getOutputParameters().setDimensions(0, 0, 0, 0, -1);
- setLineNumbers(unary1);
- setLops(unary1);
- }
- else //general case
- {
- //MR operator selection part 2
- boolean cacheU = !FORCE_REPLICATION && (m1Size < OptimizerUtils.getRemoteMemBudgetReduce());
- boolean cacheV = !FORCE_REPLICATION && ((!cacheU && m2Size < OptimizerUtils.getRemoteMemBudgetReduce())
- || (cacheU && m1Size+m2Size < OptimizerUtils.getRemoteMemBudgetReduce()));
-
- Group grpX = new Group(X.constructLops(), Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpX.getOutputParameters().setDimensions(X.getDim1(), X.getDim2(), X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(grpX);
-
- Lop lU = null;
- if( cacheU ) {
- //partitioning of U for read through distributed cache
- boolean needPartU = !U.dimsKnown() || U.getDim1() * U.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- lU = U.constructLops();
- if( needPartU ){ //requires partitioning
- lU = new DataPartition(lU, DataType.MATRIX, ValueType.DOUBLE, (m1Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), getRowsInBlock(), getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
- }
- }
- else {
- //replication of U for shuffle to target block
- Lop offset = createOffsetLop(V, false); //ncol of t(V) -> nrow of V determines num replicates
- lU = new RepMat(U.constructLops(), offset, true, V.getDataType(), V.getValueType());
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(),
- U.getRowsInBlock(), U.getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
-
- Group grpU = new Group(lU, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), U.getRowsInBlock(), U.getColsInBlock(), -1);
- setLineNumbers(grpU);
- lU = grpU;
- }
-
- Lop lV = null;
- if( cacheV ) {
- //partitioning of V for read through distributed cache
- boolean needPartV = !V.dimsKnown() || V.getDim1() * V.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- lV = V.constructLops();
- if( needPartV ){ //requires partitioning
- lV = new DataPartition(lV, DataType.MATRIX, ValueType.DOUBLE, (m2Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lV.getOutputParameters().setDimensions(V.getDim1(), V.getDim2(), getRowsInBlock(), getColsInBlock(), V.getNnz());
- setLineNumbers(lV);
- }
- }
- else {
- //replication of t(V) for shuffle to target block
- Transform ltV = new Transform( V.constructLops(), HopsTransf2Lops.get(ReOrgOp.TRANSPOSE), getDataType(), getValueType(), ExecType.MR);
- ltV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(),
- V.getColsInBlock(), V.getRowsInBlock(), V.getNnz());
- setLineNumbers(ltV);
-
- Lop offset = createOffsetLop(U, false); //nrow of U determines num replicates
- lV = new RepMat(ltV, offset, false, V.getDataType(), V.getValueType());
- lV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(),
- V.getColsInBlock(), V.getRowsInBlock(), V.getNnz());
- setLineNumbers(lV);
-
- Group grpV = new Group(lV, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(), V.getColsInBlock(), V.getRowsInBlock(), -1);
- setLineNumbers(grpV);
- lV = grpV;
- }
-
- //reduce-side wsloss w/ or without broadcast
- Lop wcemm = new WeightedCrossEntropyR(
- grpX, lU, lV, DataType.MATRIX, ValueType.DOUBLE, wtype, cacheU, cacheV, ExecType.MR);
- wcemm.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(wcemm);
-
- Group grp = new Group(wcemm, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grp.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(grp);
-
- Aggregate agg1 = new Aggregate(grp, HopsAgg2Lops.get(AggOp.SUM), DataType.MATRIX, ValueType.DOUBLE, ExecType.MR);
- agg1.setupCorrectionLocation(CorrectionLocationType.NONE); // aggregation uses kahanSum
- agg1.getOutputParameters().setDimensions(1, 1, X.getRowsInBlock(), X.getColsInBlock(), -1);
- setLineNumbers(agg1);
-
- UnaryCP unary1 = new UnaryCP(agg1, HopsOpOp1LopsUS.get(OpOp1.CAST_AS_SCALAR), getDataType(), getValueType());
- unary1.getOutputParameters().setDimensions(0, 0, 0, 0, -1);
- setLineNumbers(unary1);
- setLops(unary1);
- }
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructSparkLopsWeightedCeMM(WCeMMType wtype)
- throws HopsException, LopsException
- {
- //NOTE: the common case for wsloss are factors U/V with a rank of 10s to 100s; the current runtime only
- //supports single block outer products (U/V rank <= blocksize, i.e., 1000 by default); we enforce this
- //by applying the hop rewrite for Weighted Squared Loss only if this constraint holds.
-
- //Notes: Any broadcast needs to fit twice in local memory because we partition the input in cp,
- //and needs to fit once in executor broadcast memory. The 2GB broadcast constraint is no longer
- //required because the max_int byte buffer constraint has been fixed in Spark 1.4
- double memBudgetExec = SparkExecutionContext.getBroadcastMemoryBudget();
- double memBudgetLocal = OptimizerUtils.getLocalMemBudget();
-
- Hop X = getInput().get(0);
- Hop U = getInput().get(1);
- Hop V = getInput().get(2);
-
- //MR operator selection, part1
- double m1Size = OptimizerUtils.estimateSize(U.getDim1(), U.getDim2()); //size U
- double m2Size = OptimizerUtils.estimateSize(V.getDim1(), V.getDim2()); //size V
- boolean isMapWsloss = (m1Size+m2Size < memBudgetExec
- && 2*m1Size < memBudgetLocal && 2*m2Size < memBudgetLocal);
-
- if( !FORCE_REPLICATION && isMapWsloss ) //broadcast
- {
- //map-side wsloss always with broadcast
- Lop wsloss = new WeightedCrossEntropy( X.constructLops(), U.constructLops(), V.constructLops(),
- DataType.SCALAR, ValueType.DOUBLE, wtype, ExecType.SPARK);
- setOutputDimensions(wsloss);
- setLineNumbers(wsloss);
- setLops(wsloss);
- }
- else //general case
- {
- //MR operator selection part 2
- boolean cacheU = !FORCE_REPLICATION && (m1Size < memBudgetExec && 2*m1Size < memBudgetLocal);
- boolean cacheV = !FORCE_REPLICATION && ((!cacheU && m2Size < memBudgetExec )
- || (cacheU && m1Size+m2Size < memBudgetExec)) && 2*m2Size < memBudgetLocal;
-
- //reduce-side wsloss w/ or without broadcast
- Lop wcemm = new WeightedCrossEntropyR(
- X.constructLops(), U.constructLops(), V.constructLops(),
- DataType.SCALAR, ValueType.DOUBLE, wtype, cacheU, cacheV, ExecType.SPARK);
- setOutputDimensions(wcemm);
- setLineNumbers(wcemm);
- setLops(wcemm);
- }
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructCPLopsWeightedUMM(WUMMType wtype)
- throws HopsException, LopsException
- {
- Unary.OperationTypes uop = _uop!=null ?
- HopsOpOp1LopsU.get(_uop) : _sop==OpOp2.POW ?
- Unary.OperationTypes.POW2 : Unary.OperationTypes.MULTIPLY2;
-
- WeightedUnaryMM wsig = new WeightedUnaryMM(
- getInput().get(0).constructLops(),
- getInput().get(1).constructLops(),
- getInput().get(2).constructLops(),
- getDataType(), getValueType(), wtype, uop, ExecType.CP);
-
- //set degree of parallelism
- int k = OptimizerUtils.getConstrainedNumThreads(_maxNumThreads);
- wsig.setNumThreads(k);
-
- setOutputDimensions( wsig );
- setLineNumbers( wsig );
- setLops( wsig );
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructMRLopsWeightedUMM( WUMMType wtype )
- throws HopsException, LopsException
- {
- //NOTE: the common case for wsigmoid are factors U/V with a rank of 10s to 100s; the current runtime only
- //supports single block outer products (U/V rank <= blocksize, i.e., 1000 by default); we enforce this
- //by applying the hop rewrite for Weighted Squared Loss only if this constraint holds.
-
- Unary.OperationTypes uop = _uop!=null ?
- HopsOpOp1LopsU.get(_uop) : _sop==OpOp2.POW ?
- Unary.OperationTypes.POW2 : Unary.OperationTypes.MULTIPLY2;
-
- Hop X = getInput().get(0);
- Hop U = getInput().get(1);
- Hop V = getInput().get(2);
-
- //MR operator selection, part1
- double m1Size = OptimizerUtils.estimateSize(U.getDim1(), U.getDim2()); //size U
- double m2Size = OptimizerUtils.estimateSize(V.getDim1(), V.getDim2()); //size V
- boolean isMapWsloss = (m1Size+m2Size < OptimizerUtils.getRemoteMemBudgetMap(true));
-
- if( !FORCE_REPLICATION && isMapWsloss ) //broadcast
- {
- //partitioning of U
- boolean needPartU = !U.dimsKnown() || U.getDim1() * U.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- Lop lU = U.constructLops();
- if( needPartU ){ //requires partitioning
- lU = new DataPartition(lU, DataType.MATRIX, ValueType.DOUBLE, (m1Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), getRowsInBlock(), getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
- }
-
- //partitioning of V
- boolean needPartV = !V.dimsKnown() || V.getDim1() * V.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- Lop lV = V.constructLops();
- if( needPartV ){ //requires partitioning
- lV = new DataPartition(lV, DataType.MATRIX, ValueType.DOUBLE, (m2Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lV.getOutputParameters().setDimensions(V.getDim1(), V.getDim2(), getRowsInBlock(), getColsInBlock(), V.getNnz());
- setLineNumbers(lV);
- }
-
- //map-side wsloss always with broadcast
- Lop wumm = new WeightedUnaryMM( X.constructLops(), lU, lV,
- DataType.MATRIX, ValueType.DOUBLE, wtype, uop, ExecType.MR);
- setOutputDimensions(wumm);
- setLineNumbers(wumm);
- setLops( wumm );
-
- //in contrast to wsloss no aggregation required
- }
- else //general case
- {
- //MR operator selection part 2
- boolean cacheU = !FORCE_REPLICATION && (m1Size < OptimizerUtils.getRemoteMemBudgetReduce());
- boolean cacheV = !FORCE_REPLICATION && ((!cacheU && m2Size < OptimizerUtils.getRemoteMemBudgetReduce())
- || (cacheU && m1Size+m2Size < OptimizerUtils.getRemoteMemBudgetReduce()));
-
- Group grpX = new Group(X.constructLops(), Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpX.getOutputParameters().setDimensions(X.getDim1(), X.getDim2(), X.getRowsInBlock(), X.getColsInBlock(), X.getNnz());
- setLineNumbers(grpX);
-
- Lop lU = null;
- if( cacheU ) {
- //partitioning of U for read through distributed cache
- boolean needPartU = !U.dimsKnown() || U.getDim1() * U.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- lU = U.constructLops();
- if( needPartU ){ //requires partitioning
- lU = new DataPartition(lU, DataType.MATRIX, ValueType.DOUBLE, (m1Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), getRowsInBlock(), getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
- }
- }
- else {
- //replication of U for shuffle to target block
- Lop offset = createOffsetLop(V, false); //ncol of t(V) -> nrow of V determines num replicates
- lU = new RepMat(U.constructLops(), offset, true, V.getDataType(), V.getValueType());
- lU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(),
- U.getRowsInBlock(), U.getColsInBlock(), U.getNnz());
- setLineNumbers(lU);
-
- Group grpU = new Group(lU, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpU.getOutputParameters().setDimensions(U.getDim1(), U.getDim2(), U.getRowsInBlock(), U.getColsInBlock(), -1);
- setLineNumbers(grpU);
- lU = grpU;
- }
-
- Lop lV = null;
- if( cacheV ) {
- //partitioning of V for read through distributed cache
- boolean needPartV = !V.dimsKnown() || V.getDim1() * V.getDim2() > DistributedCacheInput.PARTITION_SIZE;
- lV = V.constructLops();
- if( needPartV ){ //requires partitioning
- lV = new DataPartition(lV, DataType.MATRIX, ValueType.DOUBLE, (m2Size>OptimizerUtils.getLocalMemBudget())?ExecType.MR:ExecType.CP, PDataPartitionFormat.ROW_BLOCK_WISE_N);
- lV.getOutputParameters().setDimensions(V.getDim1(), V.getDim2(), getRowsInBlock(), getColsInBlock(), V.getNnz());
- setLineNumbers(lV);
- }
- }
- else {
- //replication of t(V) for shuffle to target block
- Transform ltV = new Transform( V.constructLops(), HopsTransf2Lops.get(ReOrgOp.TRANSPOSE), getDataType(), getValueType(), ExecType.MR);
- ltV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(),
- V.getColsInBlock(), V.getRowsInBlock(), V.getNnz());
- setLineNumbers(ltV);
-
- Lop offset = createOffsetLop(U, false); //nrow of U determines num replicates
- lV = new RepMat(ltV, offset, false, V.getDataType(), V.getValueType());
- lV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(),
- V.getColsInBlock(), V.getRowsInBlock(), V.getNnz());
- setLineNumbers(lV);
-
- Group grpV = new Group(lV, Group.OperationTypes.Sort, DataType.MATRIX, ValueType.DOUBLE);
- grpV.getOutputParameters().setDimensions(V.getDim2(), V.getDim1(), V.getColsInBlock(), V.getRowsInBlock(), -1);
- setLineNumbers(grpV);
- lV = grpV;
- }
-
- //reduce-side wsloss w/ or without broadcast
- Lop wumm = new WeightedUnaryMMR(
- grpX, lU, lV, DataType.MATRIX, ValueType.DOUBLE, wtype, uop, cacheU, cacheV, ExecType.MR);
- setOutputDimensions(wumm);
- setLineNumbers(wumm);
- setLops(wumm);
-
- //in contrast to wsloss no aggregation required
- }
- }
-
- /**
- *
- * @param wtype
- * @throws HopsException
- * @throws LopsException
- */
- private void constructSparkLopsWeightedUMM( WUMMType wtype )
- throws HopsException, LopsException
- {
- //NOTE: the common case for wsigmoid are factors U/V with a rank of 10s to 100s; the current runtime only
- //supports single block outer products (U/V rank <= blocksize, i.e., 1000 by default); we enforce this
- //by applying the hop rewrite for Weighted Squared Loss only if this constraint holds.
-
- Unary.OperationTypes uop = _uop!=null ?
- HopsOpOp1LopsU.get(_uop) : _sop==OpOp2.POW ?
- Unary.OperationTypes.POW2 : Unary.OperationTypes.MULTIPLY2;
-
- //Notes: Any broadcast needs to fit twice in local memory because we partition the input in cp,
- //and needs to fit once in executor broadcast memory. The 2GB broadcast constraint is no longer
- //required because the max_int byte buffer constraint has been fixed in Spark 1.4
- double memBudgetExec = SparkExecutionContext.getBroadcastMemoryBudget();
- double memBudgetLocal = OptimizerUtils.getLocalMemBudget();
-
- Hop X = getInput().get(0);
- Hop U = getInput().get(1);
- Hop V = getInput().get(2);
-
- //MR operator selection, part1
- double m1Size = OptimizerUtils.estimateSize(U.getDim1(), U.getDim2()); //size U
- double m2Size = OptimizerUtils.estimateSize(V.getDim1(), V.getDim2()); //size V
- boolean isMapWsloss = (m1Size+m2Size < memBudgetExec
- && 2*m1Size<memBudgetLocal && 2*m2Size<memBudgetLocal);
-
- if( !FORCE_REPLICATION && isMapWsloss ) //broadcast
- {
- //map-side wsloss always with broadcast
- Lop wsigmoid = new WeightedUnaryMM( X.constructLops(), U.constructLops(), V.constructLops(),
- DataType.MATRIX, ValueType.DOUBLE, wtype, uop, ExecType.SPARK);
- setOutputDimensions(wsigmoid);
- setLineNumbers(wsigmoid);
- setLops( wsigmoid );
- }
- else //general case
- {
- //MR operator selection part 2
- boolean cacheU = !FORCE_REPLICATION && (m1Size < memBudgetExec && 2*m1Size < memBudgetLocal);
- boolean cacheV = !FORCE_REPLICATION && ((!cacheU && m2Size < memBudgetExec )
- || (cacheU && m1Size+m2Size < memBudgetExec)) && 2*m2Size < memBudgetLocal;
-
- //reduce-side wsloss w/ or without broadcast
- Lop wsigmoid = new WeightedUnaryMMR(
- X.constructLops(), U.constructLops(), V.constructLops(),
- DataType.MATRIX, ValueType.DOUBLE, wtype, uop, cacheU, cacheV, ExecType.SPARK);
- setOutputDimensions(wsigmoid);
- setLineNumbers(wsigmoid);
- setLops(wsigmoid);
- }
- }
-
- /**
- *
- * @return
- */
- private WeightsType checkWeightsType()
- {
- WeightsType ret = WeightsType.NONE;
- if( !(getInput().get(3) instanceof LiteralOp) ){
- if( _postWeights )
- ret = WeightsType.POST;
- else
- ret = WeightsType.PRE;
- }
- else if( _postWeights ){
- ret = WeightsType.POST_NZ;
- }
-
- return ret;
- }
-
- /**
- *
- * @return
- */
- private WSigmoidType checkWSigmoidType()
- {
-
- if( _logout && _minusin )
- return WSigmoidType.LOG_MINUS;
- else if( _logout )
- return WSigmoidType.LOG;
- else if( _minusin )
- return WSigmoidType.MINUS;
- else
- return WSigmoidType.BASIC;
- }
-
- /**
- *
- * @return
- */
- private WDivMMType checkWDivMMType()
- {
- switch( _baseType )
- {
- case 0: //BASIC
- return WDivMMType.MULT_BASIC;
- case 1: //LEFT
- if( _minus )
- return WDivMMType.MULT_MINUS_LEFT;
- else
- return _mult ? WDivMMType.MULT_LEFT : WDivMMType.DIV_LEFT;
- case 2: //RIGHT
- if( _minus )
- return WDivMMType.MULT_MINUS_RIGHT;
- else
- return _mult ? WDivMMType.MULT_RIGHT : WDivMMType.DIV_RIGHT;
- }
-
- return null;
- }
-
- @Override
- protected double computeOutputMemEstimate( long dim1, long dim2, long nnz )
- {
- switch( _op ) {
- case WSLOSS: //always scalar output
- case WCEMM:
- return OptimizerUtils.DOUBLE_SIZE;
-
- case WSIGMOID:
- case WDIVMM:
- case WUMM:
- double sp = OptimizerUtils.getSparsity(dim1, dim2, nnz);
- return OptimizerUtils.estimateSizeExactSparsity(dim1, dim2, sp);
-
- default:
- return 0;
- }
- }
-
- @Override
- protected double computeIntermediateMemEstimate( long dim1, long dim2, long nnz )
- {
- //no intermediates
- return 0;
- }
-
- @Override
- protected long[] inferOutputCharacteristics( MemoTable memo )
- {
- long[] ret = null;
-
- switch( _op ) {
- case WSLOSS: //always scalar output
- ret = null;
- break;
-
- case WSIGMOID:
- case WUMM: {
- MatrixCharacteristics mcW = memo.getAllInputStats(getInput().get(0));
- ret = new long[]{mcW.getRows(), mcW.getCols(), mcW.getNonZeros()};
- break;
- }
-
- case WDIVMM: {
- if( _baseType == 0 ){ //basic
- MatrixCharacteristics mcW = memo.getAllInputStats(getInput().get(0));
- ret = new long[]{mcW.getRows(), mcW.getCols(), mcW.getNonZeros()};
- }
- if( _baseType == 1 ) { //left (w/ transpose)
- MatrixCharacteristics mcV = memo.getAllInputStats(getInput().get(2));
- ret = new long[]{mcV.getRows(), mcV.getCols(), -1};
- }
- else { //right
- MatrixCharacteristics mcU = memo.getAllInputStats(getInput().get(1));
- ret = new long[]{mcU.getRows(), mcU.getCols(), -1};
- }
- break;
- }
-
- default:
- throw new RuntimeException("Memory for operation (" + _op + ") can not be estimated.");
- }
-
- return ret;
- }
-
- @Override
- protected ExecType optFindExecType()
- throws HopsException
- {
- checkAndSetForcedPlatform();
-
- ExecType REMOTE = OptimizerUtils.isSparkExecutionMode() ? ExecType.SPARK : ExecType.MR;
-
- if( _etypeForced != null )
- {
- _etype = _etypeForced;
- }
- else
- {
- if ( OptimizerUtils.isMemoryBasedOptLevel() ) {
- _etype = findExecTypeByMemEstimate();
- }
- else if ( (getInput().get(0).areDimsBelowThreshold()
- && getInput().get(1).areDimsBelowThreshold()
- && getInput().get(2).areDimsBelowThreshold()
- && getInput().get(3).areDimsBelowThreshold()) )
- _etype = ExecType.CP;
- else
- _etype = REMOTE;
-
- //check for valid CP dimensions and matrix size
- checkAndSetInvalidCPDimsAndSize();
- }
-
- //mark for recompile (forever)
- if( OptimizerUtils.ALLOW_DYN_RECOMPILATION && !dimsKnown(true) && _etype==REMOTE )
- setRequiresRecompile();
-
- return _etype;
- }
-
- @Override
- public void refreshSizeInformation()
- {
- switch( _op ) {
- case WSLOSS:
- //do nothing: always scalar
- break;
-
- case WSIGMOID:
- case WUMM: {
- Hop inW = getInput().get(0);
- setDim1( inW.getDim1() );
- setDim2( inW.getDim2() );
- setNnz( inW.getNnz() );
- break;
- }
-
- case WDIVMM: {
- if( _baseType == 0 ) { //basic
- Hop inW = getInput().get(0);
- setDim1( inW.getDim1() );
- setDim2( inW.getDim2() );
- setNnz( inW.getNnz() );
- }
- else if( _baseType == 1 ){ //left (w/ transpose)
- Hop inV = getInput().get(2);
- setDim1( inV.getDim1() );
- setDim2( inV.getDim2() );
- }
- else { //right
- Hop inU = getInput().get(1);
- setDim1( inU.getDim1() );
- setDim2( inU.getDim2() );
- }
- break;
- }
-
- default:
- break;
- }
- }
-
- @Override
- public Object clone() throws CloneNotSupportedException
- {
- QuaternaryOp ret = new QuaternaryOp();
-
- //copy generic attributes
- ret.clone(this, false);
-
- //copy specific attributes
- ret._op = _op;
- ret._postWeights = _postWeights;
- ret._logout = _logout;
- ret._minusin = _minusin;
- ret._baseType = _baseType;
- ret._mult = _mult;
- ret._minus = _minus;
- ret._umult = _umult;
- ret._uop = _uop;
- ret._sop = _sop;
- ret._maxNumThreads = _maxNumThreads;
-
- return ret;
- }
-
- @Override
- public boolean compare( Hop that )
- {
- if( !(that instanceof QuaternaryOp) )
- return false;
-
- QuaternaryOp that2 = (QuaternaryOp)that;
-
- //compare basic inputs and weights (always existing)
- boolean ret = (_op == that2._op
- && getInput().size() == getInput().size()
- && getInput().get(0) == that2.getInput().get(0)
- && getInput().get(1) == that2.getInput().get(1)
- && getInput().get(2) == that2.getInput().get(2) );
-
- //check for 4th argument if same size (see above)
- if( ret && getInput().size()==4 )
- ret &= (getInput().get(3) == that2.getInput().get(3));
-
- //compare specific parameters
- ret &= _postWeights == that2._postWeights;
- ret &= _logout == that2._logout;
- ret &= _minusin == that2._minusin;
- ret &= _baseType == that2._baseType;
- ret &= _mult == that2._mult;
- ret &= _minus == that2._minus;
- ret &= _umult == that2._umult;
- ret &= _uop == that2._uop;
- ret &= _sop == that2._sop;
- ret &= _maxNumThreads == that2._maxNumThreads;
-
- return ret;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/276d9257/src/main/java/com/ibm/bi/dml/hops/ReorgOp.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/hops/ReorgOp.java b/src/main/java/com/ibm/bi/dml/hops/ReorgOp.java
deleted file mode 100644
index 703fb4a..0000000
--- a/src/main/java/com/ibm/bi/dml/hops/ReorgOp.java
+++ /dev/null
@@ -1,608 +0,0 @@
-/**
- * (C) Copyright IBM Corp. 2010, 2015
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package com.ibm.bi.dml.hops;
-
-import java.util.ArrayList;
-
-import com.ibm.bi.dml.hops.rewrite.HopRewriteUtils;
-import com.ibm.bi.dml.lops.Aggregate;
-import com.ibm.bi.dml.lops.Group;
-import com.ibm.bi.dml.lops.Lop;
-import com.ibm.bi.dml.lops.LopsException;
-import com.ibm.bi.dml.lops.SortKeys;
-import com.ibm.bi.dml.lops.Transform;
-import com.ibm.bi.dml.lops.LopProperties.ExecType;
-import com.ibm.bi.dml.parser.Expression.DataType;
-import com.ibm.bi.dml.parser.Expression.ValueType;
-import com.ibm.bi.dml.runtime.matrix.MatrixCharacteristics;
-
-/**
- * Reorg (cell) operation: aij
- * Properties:
- * Symbol: ', rdiag, rshape, rsort
- * 1 Operand
- *
- * Semantic: change indices (in mapper or reducer)
- *
- *
- * NOTE MB: reshape integrated here because (1) ParameterizedBuiltinOp requires name-value pairs for params
- * and (2) most importantly semantic of reshape is exactly a reorg op.
- */
-
-public class ReorgOp extends Hop
-{
-
- public static boolean FORCE_DIST_SORT_INDEXES = false;
-
- public boolean bSortSPRewriteApplicable = false;
-
- private ReOrgOp op;
-
- private ReorgOp() {
- //default constructor for clone
- }
-
- public ReorgOp(String l, DataType dt, ValueType vt, ReOrgOp o, Hop inp)
- {
- super(l, dt, vt);
- op = o;
- getInput().add(0, inp);
- inp.getParent().add(this);
-
- //compute unknown dims and nnz
- refreshSizeInformation();
- }
-
- public ReorgOp(String l, DataType dt, ValueType vt, ReOrgOp o, ArrayList<Hop> inp)
- {
- super(l, dt, vt);
- op = o;
-
- for( int i=0; i<inp.size(); i++ ) {
- Hop in = inp.get(i);
- getInput().add(i, in);
- in.getParent().add(this);
- }
-
- //compute unknown dims and nnz
- refreshSizeInformation();
- }
-
- public ReOrgOp getOp()
- {
- return op;
- }
-
- @Override
- public String getOpString() {
- String s = new String("");
- s += "r(" + HopsTransf2String.get(op) + ")";
- return s;
- }
-
- @Override
- public Lop constructLops()
- throws HopsException, LopsException
- {
- //return already created lops
- if( getLops() != null )
- return getLops();
-
- ExecType et = optFindExecType();
-
- switch( op )
- {
- case TRANSPOSE:
- case DIAG:
- {
- Transform transform1 = new Transform( getInput().get(0).constructLops(),
- HopsTransf2Lops.get(op), getDataType(), getValueType(), et);
- setOutputDimensions(transform1);
- setLineNumbers(transform1);
- setLops(transform1);
-
- break;
- }
- case RESHAPE:
- {
- if( et==ExecType.MR )
- {
- Transform transform1 = new Transform( getInput().get(0).constructLops(),
- HopsTransf2Lops.get(op), getDataType(), getValueType(), et);
- setOutputDimensions(transform1);
- setLineNumbers(transform1);
- for( int i=1; i<=3; i++ ) //rows, cols, byrow
- {
- Lop ltmp = getInput().get(i).constructLops();
- transform1.addInput(ltmp);
- ltmp.addOutput(transform1);
- }
- transform1.setLevel(); //force order of added lops
-
- Group group1 = new Group(
- transform1, Group.OperationTypes.Sort, DataType.MATRIX,
- getValueType());
- setOutputDimensions(group1);
- setLineNumbers(group1);
-
- Aggregate agg1 = new Aggregate(
- group1, Aggregate.OperationTypes.Sum, DataType.MATRIX,
- getValueType(), et);
- setOutputDimensions(agg1);
- setLineNumbers(agg1);
-
- setLops(agg1);
- }
- else //CP/SPARK
- {
- Transform transform1 = new Transform( getInput().get(0).constructLops(),
- HopsTransf2Lops.get(op), getDataType(), getValueType(), et);
- setOutputDimensions(transform1);
- setLineNumbers(transform1);
-
- for( int i=1; i<=3; i++ ) //rows, cols, byrow
- {
- Lop ltmp = getInput().get(i).constructLops();
- transform1.addInput(ltmp);
- ltmp.addOutput(transform1);
- }
- transform1.setLevel(); //force order of added lops
-
- setLops(transform1);
- }
- break;
- }
- case SORT:
- {
- Hop input = getInput().get(0);
- Hop by = getInput().get(1);
- Hop desc = getInput().get(2);
- Hop ixret = getInput().get(3);
-
- if( et==ExecType.MR )
- {
-
- if( !(desc instanceof LiteralOp && ixret instanceof LiteralOp) ) {
- LOG.warn("Unsupported non-constant ordering parameters, using defaults and mark for recompilation.");
- setRequiresRecompile();
- desc = new LiteralOp(false);
- ixret = new LiteralOp(false);
- }
-
- //Step 1: extraction (if unknown ncol or multiple columns)
- Hop vinput = input;
- if( input.getDim2() != 1 ) {
- vinput = new IndexingOp("tmp1", getDataType(), getValueType(), input, new LiteralOp(1L),
- HopRewriteUtils.createValueHop(input, true), by, by, false, true);
- vinput.refreshSizeInformation();
- HopRewriteUtils.setOutputBlocksizes(vinput, getRowsInBlock(), getColsInBlock());
- HopRewriteUtils.copyLineNumbers(this, vinput);
- }
-
- //Step 2: Index vector sort
- Hop voutput = null;
- if( 2*OptimizerUtils.estimateSize(vinput.getDim1(), vinput.getDim2())
- > OptimizerUtils.getLocalMemBudget()
- || FORCE_DIST_SORT_INDEXES )
- {
- //large vector, fallback to MR sort
- //sort indexes according to given values
- SortKeys sort = new SortKeys(
- vinput.constructLops(), HopRewriteUtils.getBooleanValueSafe((LiteralOp)desc),
- SortKeys.OperationTypes.Indexes,
- vinput.getDataType(), vinput.getValueType(), ExecType.MR);
-
- sort.getOutputParameters().setDimensions(vinput.getDim1(), 1,
- vinput.getRowsInBlock(), vinput.getColsInBlock(), vinput.getNnz());
-
- setLineNumbers(sort);
-
- //note: this sortindexes includes also the shift by offsets and
- //final aggregate because sideways passing of offsets would
- //not nicely fit the current instruction model
-
- setLops(sort);
- voutput = this;
- }
- else
- {
- //small vector, use in-memory sort
- ArrayList<Hop> sinputs = new ArrayList<Hop>();
- sinputs.add(vinput);
- sinputs.add(new LiteralOp(1)); //by (always vector)
- sinputs.add(desc);
- sinputs.add(new LiteralOp(true)); //indexreturn (always indexes)
- voutput = new ReorgOp("tmp3", getDataType(), getValueType(), ReOrgOp.SORT, sinputs);
- HopRewriteUtils.copyLineNumbers(this, voutput);
- //explicitly construct CP lop; otherwise there is danger of infinite recursion if forced runtime platform.
- voutput.setLops( constructCPOrSparkSortLop(vinput, sinputs.get(1), sinputs.get(2), sinputs.get(3), ExecType.CP, false) );
- voutput.getLops().getOutputParameters().setDimensions(vinput.getDim1(), vinput.getDim2(), vinput.getRowsInBlock(), vinput.getColsInBlock(), vinput.getNnz());
- setLops( voutput.constructLops() );
- }
-
- //Step 3: Data permutation (only required for sorting data)
- // -- done via X' = table(seq(), IX') %*% X;
- if( !HopRewriteUtils.getBooleanValueSafe((LiteralOp)ixret) )
- {
- //generate seq
- DataGenOp seq = HopRewriteUtils.createSeqDataGenOp(voutput);
- seq.setName("tmp4");
- seq.refreshSizeInformation();
- seq.computeMemEstimate(new MemoTable()); //select exec type
- HopRewriteUtils.copyLineNumbers(this, seq);
-
- //generate table
- TernaryOp table = new TernaryOp("tmp5", DataType.MATRIX, ValueType.DOUBLE, OpOp3.CTABLE, seq, voutput, new LiteralOp(1L) );
- HopRewriteUtils.setOutputBlocksizes(table, getRowsInBlock(), getColsInBlock());
- table.refreshSizeInformation();
- table.setForcedExecType(ExecType.MR); //force MR
- HopRewriteUtils.copyLineNumbers(this, table);
- table.setDisjointInputs(true);
- table.setOutputEmptyBlocks(false);
-
- //generate matrix mult
- AggBinaryOp mmult = HopRewriteUtils.createMatrixMultiply(table, input);
- mmult.setForcedExecType(ExecType.MR); //force MR
-
- setLops( mmult.constructLops() );
-
- //cleanups
- HopRewriteUtils.removeChildReference(table, input);
- }
- }
- else //CP or Spark
- {
- if( et==ExecType.SPARK && !FORCE_DIST_SORT_INDEXES)
- bSortSPRewriteApplicable = isSortSPRewriteApplicable();
-
- Lop transform1 = constructCPOrSparkSortLop(input, by, desc, ixret, et, bSortSPRewriteApplicable);
- setOutputDimensions(transform1);
- setLineNumbers(transform1);
-
- setLops(transform1);
- }
- break;
- }
-
- default:
- throw new HopsException("Unsupported lops construction for operation type '"+op+"'.");
- }
-
- //add reblock/checkpoint lops if necessary
- constructAndSetLopsDataFlowProperties();
-
- return getLops();
- }
-
- private static Lop constructCPOrSparkSortLop( Hop input, Hop by, Hop desc, Hop ixret, ExecType et, boolean bSortIndInMem )
- throws HopsException, LopsException
- {
- Transform transform1 = new Transform( input.constructLops(), HopsTransf2Lops.get(ReOrgOp.SORT),
- input.getDataType(), input.getValueType(), et, bSortIndInMem);
-
- for( Hop c : new Hop[]{by,desc,ixret} ) {
- Lop ltmp = c.constructLops();
- transform1.addInput(ltmp);
- ltmp.addOutput(transform1);
- }
-
- transform1.setLevel(); //force order of added lops
-
- return transform1;
- }
-
- @Override
- protected double computeOutputMemEstimate( long dim1, long dim2, long nnz )
- {
- //no dedicated mem estimation per op type, because always propagated via refreshSizeInformation
- double sparsity = OptimizerUtils.getSparsity(dim1, dim2, nnz);
- return OptimizerUtils.estimateSizeExactSparsity(dim1, dim2, sparsity);
- }
-
- @Override
- protected double computeIntermediateMemEstimate( long dim1, long dim2, long nnz )
- {
- if( op == ReOrgOp.SORT )
- {
- Hop ixreturn = getInput().get(3);
- if( !(ixreturn instanceof LiteralOp && !HopRewriteUtils.getBooleanValueSafe((LiteralOp)ixreturn)
- && (dim2==1 || nnz==0) ) ) //NOT early abort case
- {
- //Version 2: memory requirements for temporary index int[] array,
- //(temporary double[] array already covered by output)
- return dim1 * 4;
-
- //Version 1: memory requirements for temporary index Integer[] array
- //8-16 (12) bytes for object, 4byte int payload, 4-8 (8) byte pointers.
- //return dim1 * 24;
- }
- }
-
- //default: no intermediate memory requirements
- return 0;
- }
-
- @Override
- protected long[] inferOutputCharacteristics( MemoTable memo )
- {
- long[] ret = null;
-
- Hop input = getInput().get(0);
- MatrixCharacteristics mc = memo.getAllInputStats(input);
-
- switch(op)
- {
- case TRANSPOSE:
- {
- // input is a [k1,k2] matrix and output is a [k2,k1] matrix
- // #nnz in output is exactly the same as in input
- if( mc.dimsKnown() )
- ret = new long[]{ mc.getCols(), mc.getRows(), mc.getNonZeros() };
- break;
- }
- case DIAG:
- {
- // NOTE: diag is overloaded according to the number of columns of the input
-
- long k = mc.getRows();
-
- // CASE a) DIAG V2M
- // input is a [1,k] or [k,1] matrix, and output is [k,k] matrix
- // #nnz in output is in the worst case k => sparsity = 1/k
- if( k == 1 )
- ret = new long[]{k, k, ((mc.getNonZeros()>=0) ? mc.getNonZeros() : k)};
-
- // CASE b) DIAG M2V
- // input is [k,k] matrix and output is [k,1] matrix
- // #nnz in the output is likely to be k (a dense matrix)
- if( k > 1 )
- ret = new long[]{k, 1, ((mc.getNonZeros()>=0) ? Math.min(k,mc.getNonZeros()) : k) };
-
- break;
- }
- case RESHAPE:
- {
- // input is a [k1,k2] matrix and output is a [k3,k4] matrix with k1*k2=k3*k4
- // #nnz in output is exactly the same as in input
- if( mc.dimsKnown() ) {
- if( _dim1 > 0 )
- ret = new long[]{ _dim1, mc.getRows()*mc.getCols()/_dim1, mc.getNonZeros()};
- else if( _dim2 > 0 )
- ret = new long[]{ mc.getRows()*mc.getCols()/_dim2, _dim2, mc.getNonZeros()};
- }
- break;
- }
- case SORT:
- {
- // input is a [k1,k2] matrix and output is a [k1,k3] matrix, where k3=k2 if no index return;
- // otherwise k3=1 (for the index vector)
- Hop input4 = getInput().get(3); //indexreturn
- boolean unknownIxRet = !(input4 instanceof LiteralOp);
-
- if( !unknownIxRet ) {
- boolean ixret = HopRewriteUtils.getBooleanValueSafe((LiteralOp)input4);
- long dim2 = ixret ? 1 : mc.getCols();
- long nnz = ixret ? mc.getRows() : mc.getNonZeros();
- ret = new long[]{ mc.getRows(), dim2, nnz};
- }
- else {
- ret = new long[]{ mc.getRows(), -1, -1};
- }
- }
- }
-
- return ret;
- }
-
-
- @Override
- public boolean allowsAllExecTypes()
- {
- return true;
- }
-
- @Override
- protected ExecType optFindExecType() throws HopsException {
-
- checkAndSetForcedPlatform();
-
- ExecType REMOTE = OptimizerUtils.isSparkExecutionMode() ? ExecType.SPARK : ExecType.MR;
-
- if( _etypeForced != null )
- {
- _etype = _etypeForced;
- }
- else
- {
- if ( OptimizerUtils.isMemoryBasedOptLevel() ) {
- _etype = findExecTypeByMemEstimate();
- }
- // Choose CP, if the input dimensions are below threshold or if the input is a vector
- else if ( getInput().get(0).areDimsBelowThreshold() || getInput().get(0).isVector() )
- {
- _etype = ExecType.CP;
- }
- else
- {
- _etype = REMOTE;
- }
-
- //check for valid CP dimensions and matrix size
- checkAndSetInvalidCPDimsAndSize();
- }
-
- //mark for recompile (forever)
- if( OptimizerUtils.ALLOW_DYN_RECOMPILATION && !dimsKnown(true) && _etype==REMOTE )
- setRequiresRecompile();
-
- return _etype;
- }
-
- @Override
- public void refreshSizeInformation()
- {
- Hop input1 = getInput().get(0);
-
- switch(op)
- {
- case TRANSPOSE:
- {
- // input is a [k1,k2] matrix and output is a [k2,k1] matrix
- // #nnz in output is exactly the same as in input
- setDim1(input1.getDim2());
- setDim2(input1.getDim1());
- setNnz(input1.getNnz());
- break;
- }
- case DIAG:
- {
- // NOTE: diag is overloaded according to the number of columns of the input
-
- long k = input1.getDim1();
- setDim1(k);
-
- // CASE a) DIAG_V2M
- // input is a [1,k] or [k,1] matrix, and output is [k,k] matrix
- // #nnz in output is in the worst case k => sparsity = 1/k
- if( input1.getDim2()==1 ) {
- setDim2(k);
- setNnz( (input1.getNnz()>=0) ? input1.getNnz() : k );
- }
-
- // CASE b) DIAG_M2V
- // input is [k,k] matrix and output is [k,1] matrix
- // #nnz in the output is likely to be k (a dense matrix)
- if( input1.getDim2()>1 ){
- setDim2(1);
- setNnz( (input1.getNnz()>=0) ? Math.min(k,input1.getNnz()) : k );
- }
-
- break;
- }
- case RESHAPE:
- {
- // input is a [k1,k2] matrix and output is a [k3,k4] matrix with k1*k2=k3*k4
- // #nnz in output is exactly the same as in input
- Hop input2 = getInput().get(1); //rows
- Hop input3 = getInput().get(2); //cols
- refreshRowsParameterInformation(input2); //refresh rows
- refreshColsParameterInformation(input3); //refresh cols
- setNnz(input1.getNnz());
- if( !dimsKnown() &&input1.dimsKnown() ) { //reshape allows to infer dims, if input and 1 dim known
- if(_dim1 > 0)
- _dim2 = (input1._dim1*input1._dim2)/_dim1;
- else if(_dim2 > 0)
- _dim1 = (input1._dim1*input1._dim2)/_dim2;
- }
- break;
- }
- case SORT:
- {
- // input is a [k1,k2] matrix and output is a [k1,k3] matrix, where k3=k2 if no index return;
- // otherwise k3=1 (for the index vector)
- Hop input4 = getInput().get(3); //indexreturn
- boolean unknownIxRet = !(input4 instanceof LiteralOp);
-
- _dim1 = input1.getDim1();
- if( !unknownIxRet ) {
- boolean ixret = HopRewriteUtils.getBooleanValueSafe((LiteralOp)input4);
- _dim2 = ixret ? 1 : input1.getDim2();
- _nnz = ixret ? input1.getDim1() : input1.getNnz();
- }
- else {
- _dim2 = -1;
- _nnz = -1;
- }
- break;
- }
- }
- }
-
- @Override
- public Object clone() throws CloneNotSupportedException
- {
- ReorgOp ret = new ReorgOp();
-
- //copy generic attributes
- ret.clone(this, false);
-
- //copy specific attributes
- ret.op = op;
-
- return ret;
- }
-
- @Override
- public boolean compare( Hop that )
- {
- if( !(that instanceof ReorgOp) )
- return false;
-
- ReorgOp that2 = (ReorgOp)that;
- boolean ret = (op == that2.op)
- && (getInput().size()==that.getInput().size());
-
- //compare all childs (see reshape, sort)
- if( ret ) //sizes matched
- for( int i=0; i<_input.size(); i++ )
- ret &= getInput().get(i) == that2.getInput().get(i);
-
- return ret;
- }
-
-
- @Override
- public void printMe() throws HopsException
- {
- if (LOG.isDebugEnabled()){
- if (getVisited() != VisitStatus.DONE) {
- super.printMe();
- LOG.debug(" Operation: " + op);
- for (Hop h : getInput()) {
- h.printMe();
- }
- }
- setVisited(VisitStatus.DONE);
- }
- }
-
- /**
- * This will check if there is sufficient memory locally (twice the size of second matrix, for original and sort data), and remotely (size of second matrix (sorted data)).
- * @return
- */
- private boolean isSortSPRewriteApplicable()
- {
- boolean ret = false;
- Hop input = getInput().get(0);
-
- //note: both cases (partitioned matrix, and sorted double array), require to
- //fit the broadcast twice into the local memory budget. Also, the memory
- //constraint only needs to take the rhs into account because the output is
- //guaranteed to be an aggregate of <=16KB
-
- double size = input.dimsKnown() ?
- OptimizerUtils.estimateSize(input.getDim1(), 1) : //dims known and estimate fits
- input.getOutputMemEstimate(); //dims unknown but worst-case estimate fits
-
- if( OptimizerUtils.checkSparkBroadcastMemoryBudget(size) ) {
- ret = true;
- }
-
- return ret;
- }
-}
\ No newline at end of file