| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| package org.apache.sysml.yarn.ropt; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| |
| import org.apache.sysml.hops.DataOp; |
| import org.apache.sysml.hops.Hop; |
| import org.apache.sysml.hops.Hop.DataOpTypes; |
| import org.apache.sysml.hops.HopsException; |
| import org.apache.sysml.hops.OptimizerUtils; |
| import org.apache.sysml.hops.cost.CostEstimationWrapper; |
| import org.apache.sysml.hops.recompile.Recompiler; |
| import org.apache.sysml.lops.LopsException; |
| import org.apache.sysml.lops.LopProperties.ExecType; |
| import org.apache.sysml.parser.Expression.ValueType; |
| import org.apache.sysml.parser.ForStatementBlock; |
| import org.apache.sysml.parser.IfStatementBlock; |
| import org.apache.sysml.parser.StatementBlock; |
| import org.apache.sysml.parser.WhileStatementBlock; |
| import org.apache.sysml.parser.Expression.DataType; |
| import org.apache.sysml.runtime.DMLRuntimeException; |
| import org.apache.sysml.runtime.controlprogram.ForProgramBlock; |
| import org.apache.sysml.runtime.controlprogram.FunctionProgramBlock; |
| import org.apache.sysml.runtime.controlprogram.IfProgramBlock; |
| import org.apache.sysml.runtime.controlprogram.LocalVariableMap; |
| import org.apache.sysml.runtime.controlprogram.Program; |
| import org.apache.sysml.runtime.controlprogram.ProgramBlock; |
| import org.apache.sysml.runtime.controlprogram.WhileProgramBlock; |
| import org.apache.sysml.runtime.controlprogram.caching.MatrixObject; |
| import org.apache.sysml.runtime.controlprogram.context.ExecutionContext; |
| import org.apache.sysml.runtime.controlprogram.context.ExecutionContextFactory; |
| import org.apache.sysml.runtime.controlprogram.parfor.opt.OptTreeConverter; |
| import org.apache.sysml.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer; |
| import org.apache.sysml.runtime.controlprogram.parfor.stat.Timing; |
| import org.apache.sysml.runtime.instructions.Instruction; |
| import org.apache.sysml.runtime.instructions.MRJobInstruction; |
| import org.apache.sysml.runtime.matrix.MatrixCharacteristics; |
| import org.apache.sysml.runtime.matrix.MatrixDimensionsMetaData; |
| import org.apache.sysml.yarn.DMLYarnClient; |
| import org.apache.sysml.yarn.ropt.YarnOptimizerUtils.GridEnumType; |
| |
| /** |
| * TODO parallel version with exposed numThreads parameter |
| * |
| */ |
| public class ResourceOptimizer |
| { |
| |
| private static final Log LOG = LogFactory.getLog(ResourceOptimizer.class); |
| |
| //internal configuration parameters |
| public static final long MIN_CP_BUDGET = 512*1024*1024; //512MB |
| public static final boolean INCLUDE_PREDICATES = true; |
| public static final boolean PRUNING_SMALL = true; |
| public static final boolean PRUNING_UNKNOWN = true; |
| public static final boolean COSTS_MAX_PARALLELISM = true; |
| public static final boolean COST_INDIVIDUAL_BLOCKS = true; |
| |
| private static long _cntCompilePB = 0; |
| private static long _cntCostPB = 0; |
| |
| |
| /** |
| * |
| * @param prog |
| * @param cc |
| * @param cptype |
| * @param mrtype |
| * @return |
| * @throws DMLRuntimeException |
| */ |
| public synchronized static ResourceConfig optimizeResourceConfig( ArrayList<ProgramBlock> prog, YarnClusterConfig cc, GridEnumType cptype, GridEnumType mrtype ) |
| throws DMLRuntimeException |
| { |
| ResourceConfig ROpt = null; |
| |
| try |
| { |
| //init statistics and counters |
| Timing time = new Timing(true); |
| initStatistics(); |
| |
| //get constraints (yarn-specific: force higher min to limit degree of parallelism) |
| long max = (long)(YarnOptimizerUtils.toB(cc.getMaxAllocationMB()) / DMLYarnClient.MEM_FACTOR); |
| long minCP = (long) Math.max(YarnOptimizerUtils.toB(cc.getMinAllocationMB()) / DMLYarnClient.MEM_FACTOR, MIN_CP_BUDGET); |
| long minMR = YarnOptimizerUtils.computeMinContraint(minCP, max, cc.getAvgNumCores()); |
| |
| //enumerate grid points for given types (refers to jvm max heap) |
| ArrayList<Long> SRc = enumerateGridPoints(prog, minCP, max, cptype); |
| ArrayList<Long> SRm = enumerateGridPoints(prog, minMR, max, mrtype); |
| |
| //init resource config and global costs |
| ROpt = new ResourceConfig(prog, minMR); |
| double costOpt = Double.MAX_VALUE; |
| |
| for( Long rc : SRc ) //enumerate CP memory rc |
| { |
| //baseline compile and pruning |
| ArrayList<ProgramBlock> B = compileProgram(prog, null, rc, minMR); //unrolled Bp |
| ArrayList<ProgramBlock> Bp = pruneProgramBlocks( B ); |
| LOG.debug("Enum (rc="+rc+"): |B|="+B.size()+", |Bp|="+Bp.size()); |
| |
| //init local memo table [resource, cost] |
| double[][] memo = initLocalMemoTable( Bp, minMR ); |
| |
| for( int i=0; i<Bp.size(); i++ ) //for all relevant blocks |
| { |
| ProgramBlock pb = Bp.get(i); |
| |
| for( Long rm : SRm ) //for each MR memory |
| { |
| //recompile program block |
| recompileProgramBlock(pb, rc, rm); |
| |
| //local costing and memo table maintenance (cost entire program to account for |
| //in-memory status of variables and loops) |
| double lcost = getProgramCosts( pb ); |
| if( lcost < memo[i][1] ) { //accept new local opt |
| memo[i][0] = rm; |
| memo[i][1] = lcost; |
| //LOG.debug("Enum (rc="+rc+"): found new local opt w/ cost="+lcost); |
| } |
| //LOG.debug("Enum (rc="+rc+", rm="+rm+"): lcost="+lcost+", mincost="+memo[i][1]); |
| } |
| } |
| |
| //global costing |
| double[][] gmemo = initGlobalMemoTable(B, Bp, memo, minMR); |
| recompileProgramBlocks(B, rc, gmemo); |
| double gcost = getProgramCosts(B.get(0).getProgram()); |
| if( gcost < costOpt ){ //accept new global opt |
| ROpt.setCPResource(rc.longValue()); |
| ROpt.setMRResources(B, gmemo); |
| costOpt = gcost; |
| LOG.debug("Enum (rc="+rc+"): found new opt w/ cost="+gcost); |
| } |
| } |
| |
| //print optimization summary |
| LOG.info("Optimization summary:"); |
| LOG.info("-- optimal plan (rc, rm): "+YarnOptimizerUtils.toMB(ROpt.getCPResource())+"MB, "+YarnOptimizerUtils.toMB(ROpt.getMaxMRResource())+"MB"); |
| LOG.info("-- costs of optimal plan: "+costOpt); |
| LOG.info("-- # of block compiles: "+_cntCompilePB); |
| LOG.info("-- # of block costings: "+_cntCostPB); |
| LOG.info("-- optimization time: "+String.format("%.3f", (double)time.stop()/1000)+" sec."); |
| LOG.info("-- optimal plan details: "+ROpt.serialize()); |
| } |
| catch(Exception ex) |
| { |
| throw new DMLRuntimeException(ex); |
| } |
| |
| return ROpt; |
| } |
| |
| /** |
| * |
| * @param prog |
| * @param B |
| * @param rc |
| * @return |
| * @throws IOException |
| * @throws LopsException |
| * @throws HopsException |
| * @throws DMLRuntimeException |
| */ |
| public static ArrayList<ProgramBlock> compileProgram( ArrayList<ProgramBlock> prog, ResourceConfig rc ) |
| throws DMLRuntimeException, HopsException, LopsException, IOException |
| { |
| //recompile program block hierarchy to list of blocks and apply optimized resource configuration |
| ArrayList<ProgramBlock> B = compileProgram(prog, null, rc.getCPResource(), rc.getMaxMRResource()); |
| ResourceOptimizer.recompileProgramBlocks(B, rc.getCPResource(), rc.getMRResourcesMemo()); |
| |
| return B; |
| } |
| |
| |
| /** |
| * |
| * @param prog |
| * @param B |
| * @return |
| * @throws IOException |
| * @throws LopsException |
| * @throws HopsException |
| * @throws DMLRuntimeException |
| */ |
| private static ArrayList<ProgramBlock> compileProgram( ArrayList<ProgramBlock> prog, ArrayList<ProgramBlock> B, double cp, double mr ) |
| throws DMLRuntimeException, HopsException, LopsException, IOException |
| { |
| if( B == null ) //init |
| { |
| B = new ArrayList<ProgramBlock>(); |
| |
| InfrastructureAnalyzer.setLocalMaxMemory( (long)cp ); |
| InfrastructureAnalyzer.setRemoteMaxMemoryMap( (long)mr ); |
| InfrastructureAnalyzer.setRemoteMaxMemoryReduce( (long)mr ); |
| OptimizerUtils.resetDefaultSize(); //dependent on cp, mr |
| } |
| |
| for( ProgramBlock pb : prog ) |
| compileProgram( pb, B, cp, mr ); |
| |
| return B; |
| } |
| |
| /** |
| * |
| * @param pb |
| * @param Bp |
| * @return |
| * @throws IOException |
| * @throws LopsException |
| * @throws HopsException |
| * @throws DMLRuntimeException |
| */ |
| private static ArrayList<ProgramBlock> compileProgram( ProgramBlock pb, ArrayList<ProgramBlock> B, double cp, double mr ) |
| throws DMLRuntimeException, HopsException, LopsException, IOException |
| { |
| if (pb instanceof FunctionProgramBlock) |
| { |
| FunctionProgramBlock fpb = (FunctionProgramBlock)pb; |
| compileProgram(fpb.getChildBlocks(), B, cp, mr); |
| } |
| else if (pb instanceof WhileProgramBlock) |
| { |
| WhileProgramBlock wpb = (WhileProgramBlock)pb; |
| WhileStatementBlock sb = (WhileStatementBlock) pb.getStatementBlock(); |
| if( INCLUDE_PREDICATES && sb!=null && sb.getPredicateHops()!=null ){ |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb.getPredicateHops(), new LocalVariableMap(), null, false, 0); |
| wpb.setPredicate( inst ); |
| B.add(wpb); |
| _cntCompilePB ++; |
| } |
| compileProgram(wpb.getChildBlocks(), B, cp, mr); |
| } |
| else if (pb instanceof IfProgramBlock) |
| { |
| IfProgramBlock ipb = (IfProgramBlock)pb; |
| IfStatementBlock sb = (IfStatementBlock) ipb.getStatementBlock(); |
| if( INCLUDE_PREDICATES && sb!=null && sb.getPredicateHops()!=null ){ |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb.getPredicateHops(), new LocalVariableMap(), null, false, 0); |
| ipb.setPredicate( inst ); |
| B.add(ipb); |
| _cntCompilePB ++; |
| } |
| compileProgram(ipb.getChildBlocksIfBody(), B, cp, mr); |
| compileProgram(ipb.getChildBlocksElseBody(), B, cp, mr); |
| } |
| else if (pb instanceof ForProgramBlock) //incl parfor |
| { |
| ForProgramBlock fpb = (ForProgramBlock)pb; |
| ForStatementBlock sb = (ForStatementBlock) fpb.getStatementBlock(); |
| if( INCLUDE_PREDICATES && sb!=null ){ |
| if( sb.getFromHops()!=null ){ |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb.getFromHops(), new LocalVariableMap(), null, false, 0); |
| fpb.setFromInstructions( inst ); |
| } |
| if( sb.getToHops()!=null ){ |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb.getToHops(), new LocalVariableMap(), null, false, 0); |
| fpb.setToInstructions( inst ); |
| } |
| if( sb.getIncrementHops()!=null ){ |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb.getIncrementHops(), new LocalVariableMap(), null, false, 0); |
| fpb.setIncrementInstructions( inst ); |
| } |
| B.add(fpb); |
| _cntCompilePB ++; |
| } |
| compileProgram(fpb.getChildBlocks(), B, cp, mr); |
| } |
| else |
| { |
| StatementBlock sb = pb.getStatementBlock(); |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb, sb.get_hops(), |
| new LocalVariableMap(), null, false, 0); |
| pb.setInstructions( inst ); |
| B.add(pb); |
| _cntCompilePB ++; |
| } |
| |
| return B; |
| } |
| |
| |
| /** |
| * |
| * @param pbs |
| * @param cp |
| * @param memo |
| * @throws DMLRuntimeException |
| * @throws HopsException |
| * @throws LopsException |
| * @throws IOException |
| */ |
| private static void recompileProgramBlocks( ArrayList<ProgramBlock> pbs, long cp, double[][] memo ) |
| throws DMLRuntimeException, HopsException, LopsException, IOException |
| { |
| for( int i=0; i<pbs.size(); i++ ) |
| { |
| ProgramBlock pb = pbs.get(i); |
| long mr = (long)memo[i][0]; |
| recompileProgramBlock(pb, cp, mr); |
| } |
| } |
| |
| /** |
| * |
| * @param pb |
| * @param cp |
| * @param mr |
| * @throws DMLRuntimeException |
| * @throws HopsException |
| * @throws LopsException |
| * @throws IOException |
| */ |
| private static void recompileProgramBlock( ProgramBlock pb, long cp, long mr ) |
| throws DMLRuntimeException, HopsException, LopsException, IOException |
| { |
| //init compiler memory budget |
| InfrastructureAnalyzer.setLocalMaxMemory( cp ); |
| InfrastructureAnalyzer.setRemoteMaxMemoryMap( mr ); |
| InfrastructureAnalyzer.setRemoteMaxMemoryReduce( mr ); |
| OptimizerUtils.resetDefaultSize(); //dependent on cp, mr |
| |
| //recompile instructions (incl predicates) |
| if (pb instanceof WhileProgramBlock) |
| { |
| WhileProgramBlock wpb = (WhileProgramBlock)pb; |
| WhileStatementBlock sb = (WhileStatementBlock) pb.getStatementBlock(); |
| if( INCLUDE_PREDICATES && sb!=null && sb.getPredicateHops()!=null ){ |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb.getPredicateHops(), new LocalVariableMap(), null, false, 0); |
| inst = annotateMRJobInstructions(inst, cp, mr); |
| wpb.setPredicate( inst ); |
| } |
| } |
| else if (pb instanceof IfProgramBlock) |
| { |
| IfProgramBlock ipb = (IfProgramBlock)pb; |
| IfStatementBlock sb = (IfStatementBlock) ipb.getStatementBlock(); |
| if( INCLUDE_PREDICATES && sb!=null && sb.getPredicateHops()!=null ){ |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb.getPredicateHops(), new LocalVariableMap(), null, false, 0); |
| inst = annotateMRJobInstructions(inst, cp, mr); |
| ipb.setPredicate( inst ); |
| } |
| } |
| else if (pb instanceof ForProgramBlock) //incl parfor |
| { |
| ForProgramBlock fpb = (ForProgramBlock)pb; |
| ForStatementBlock sb = (ForStatementBlock) fpb.getStatementBlock(); |
| if( INCLUDE_PREDICATES && sb!=null ){ |
| if( sb.getFromHops()!=null ){ |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb.getFromHops(), new LocalVariableMap(), null, false, 0); |
| inst = annotateMRJobInstructions(inst, cp, mr); |
| fpb.setFromInstructions( inst ); |
| } |
| if( sb.getToHops()!=null ){ |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb.getToHops(), new LocalVariableMap(), null, false, 0); |
| inst = annotateMRJobInstructions(inst, cp, mr); |
| fpb.setToInstructions( inst ); |
| } |
| if( sb.getIncrementHops()!=null ){ |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb.getIncrementHops(), new LocalVariableMap(), null, false, 0); |
| inst = annotateMRJobInstructions(inst, cp, mr); |
| fpb.setIncrementInstructions( inst ); |
| } |
| } |
| } |
| else //last-level program blocks |
| { |
| StatementBlock sb = pb.getStatementBlock(); |
| ArrayList<Instruction> inst = Recompiler.recompileHopsDag(sb, sb.get_hops(), |
| new LocalVariableMap(), null, false, 0); |
| inst = annotateMRJobInstructions(inst, cp, mr); |
| pb.setInstructions( inst ); |
| } |
| |
| _cntCompilePB ++; |
| } |
| |
| /** |
| * |
| * @param inst |
| * @return |
| * @throws DMLRuntimeException |
| */ |
| private static ArrayList<Instruction> annotateMRJobInstructions( ArrayList<Instruction> inst, long cp, long mr ) |
| throws DMLRuntimeException |
| { |
| //check for empty instruction lists (e.g., predicates) |
| if( inst == null || !COSTS_MAX_PARALLELISM ) |
| return inst; |
| |
| try |
| { |
| for( int i=0; i<inst.size(); i++ ) |
| { |
| Instruction linst = inst.get(i); |
| if( linst instanceof MRJobInstruction ){ |
| //copy mr job instruction |
| MRJobResourceInstruction newlinst = new MRJobResourceInstruction((MRJobInstruction)linst); |
| |
| //compute and annotate |
| long maxMemPerNode = (long)YarnClusterAnalyzer.getMaxAllocationBytes(); |
| long nNodes = YarnClusterAnalyzer.getNumNodes(); |
| long totalMem = nNodes * maxMemPerNode; |
| long maxMRTasks = (long)(totalMem - DMLYarnClient.computeMemoryAllocation(cp)) |
| / (long)DMLYarnClient.computeMemoryAllocation(mr); |
| newlinst.setMaxMRTasks( maxMRTasks ); |
| |
| //write enhanced instruction back |
| inst.set(i, newlinst); |
| } |
| } |
| } |
| catch(Exception ex) |
| { |
| throw new DMLRuntimeException(ex); |
| } |
| |
| return inst; |
| } |
| |
| |
| /** |
| * |
| * @param pb |
| * @return |
| * @throws DMLRuntimeException |
| * @throws HopsException |
| */ |
| private static double getProgramCosts( ProgramBlock pb ) |
| throws DMLRuntimeException, HopsException |
| { |
| double val = 0; |
| if( COST_INDIVIDUAL_BLOCKS ) { |
| LocalVariableMap vars = new LocalVariableMap(); |
| collectReadVariables(pb.getStatementBlock().get_hops(), vars); |
| ExecutionContext ec = ExecutionContextFactory.createContext(false, null); |
| ec.setVariables(vars); |
| val = CostEstimationWrapper.getTimeEstimate(pb, ec, false); |
| } |
| else{ |
| //we need to cost the entire program in order to take in-memory status into account |
| ExecutionContext ec = ExecutionContextFactory.createContext(); |
| val = CostEstimationWrapper.getTimeEstimate(pb.getProgram(), ec); |
| } |
| |
| _cntCostPB ++; |
| return val; |
| } |
| |
| |
| /** |
| * |
| * @param prog |
| * @throws DMLRuntimeException |
| */ |
| private static double getProgramCosts( Program prog ) |
| throws DMLRuntimeException |
| { |
| //we need to cost the entire program in order to take in-memory status into account |
| ExecutionContext ec = ExecutionContextFactory.createContext(); |
| double val = CostEstimationWrapper.getTimeEstimate(prog, ec); |
| _cntCostPB ++; |
| |
| return val; |
| } |
| |
| /** |
| * |
| * @param hops |
| * @param vars |
| */ |
| private static void collectReadVariables( ArrayList<Hop> hops, LocalVariableMap vars ) |
| { |
| if( hops!=null ) { |
| Hop.resetVisitStatus(hops); |
| for( Hop hop : hops ) |
| collectReadVariables(hop, vars); |
| } |
| } |
| |
| /** |
| * |
| * @param hop |
| * @param vars |
| */ |
| private static void collectReadVariables( Hop hop, LocalVariableMap vars ) |
| { |
| if( hop == null ) |
| return; |
| |
| //process childs |
| for(Hop hi : hop.getInput()) |
| collectReadVariables( hi, vars ); |
| |
| //investigate hop exec type and known dimensions |
| if( hop instanceof DataOp && hop.getDataType()==DataType.MATRIX |
| && (((DataOp)hop).getDataOpType()==DataOpTypes.TRANSIENTREAD |
| || ((DataOp)hop).getDataOpType()==DataOpTypes.PERSISTENTREAD) ) |
| { |
| String varname = hop.getName(); |
| MatrixCharacteristics mc = new MatrixCharacteristics(hop.getDim1(), hop.getDim2(), |
| (int)hop.getRowsInBlock(), (int)hop.getColsInBlock(), hop.getNnz()); |
| MatrixDimensionsMetaData md = new MatrixDimensionsMetaData(mc); |
| MatrixObject mo = new MatrixObject(ValueType.DOUBLE, "/tmp", md); |
| vars.put(varname, mo); |
| } |
| |
| hop.setVisited(Hop.VisitStatus.DONE); |
| } |
| |
| /** |
| * |
| * @param B |
| * @return |
| * @throws HopsException |
| */ |
| private static ArrayList<ProgramBlock> pruneProgramBlocks( ArrayList<ProgramBlock> B ) |
| throws HopsException |
| { |
| //prune all program blocks w/o mr instructions (mr budget does not matter) |
| if( PRUNING_SMALL ){ |
| ArrayList<ProgramBlock> Bp = new ArrayList<ProgramBlock>(); |
| for( ProgramBlock pb : B ) |
| if( OptTreeConverter.containsMRJobInstruction(pb.getInstructions(), false, true) ) |
| Bp.add( pb ); |
| B = Bp; |
| } |
| |
| //prune all program blocks, where all mr hops are due to unknowns |
| if( PRUNING_UNKNOWN ){ |
| ArrayList<ProgramBlock> Bp = new ArrayList<ProgramBlock>(); |
| for( ProgramBlock pb : B ) |
| if( !pruneHasOnlyUnknownMR(pb) ) |
| Bp.add( pb ); |
| B = Bp; |
| } |
| |
| return B; |
| } |
| |
| /** |
| * |
| * @param pb |
| * @return |
| * @throws HopsException |
| */ |
| private static boolean pruneHasOnlyUnknownMR( ProgramBlock pb ) |
| throws HopsException |
| { |
| if (pb instanceof WhileProgramBlock) |
| { |
| WhileStatementBlock sb = (WhileStatementBlock) pb.getStatementBlock(); |
| sb.getPredicateHops().resetVisitStatus(); |
| return pruneHasOnlyUnknownMR(sb.getPredicateHops()); |
| } |
| else if (pb instanceof IfProgramBlock) |
| { |
| IfStatementBlock sb = (IfStatementBlock) pb.getStatementBlock(); |
| sb.getPredicateHops().resetVisitStatus(); |
| return pruneHasOnlyUnknownMR(sb.getPredicateHops()); |
| } |
| else if (pb instanceof ForProgramBlock) //incl parfor |
| { |
| ForStatementBlock sb = (ForStatementBlock) pb.getStatementBlock(); |
| sb.getFromHops().resetVisitStatus(); |
| sb.getToHops().resetVisitStatus(); |
| sb.getIncrementHops().resetVisitStatus(); |
| return pruneHasOnlyUnknownMR(sb.getFromHops()) |
| && pruneHasOnlyUnknownMR(sb.getToHops()) |
| && pruneHasOnlyUnknownMR(sb.getIncrementHops()); |
| } |
| else //last-level program blocks |
| { |
| StatementBlock sb = pb.getStatementBlock(); |
| return pruneHasOnlyUnknownMR(sb.get_hops()); |
| } |
| } |
| |
| |
| /** |
| * |
| * @param sb |
| * @return |
| * @throws HopsException |
| */ |
| private static boolean pruneHasOnlyUnknownMR( ArrayList<Hop> hops ) |
| throws HopsException |
| { |
| boolean ret = false; |
| |
| if( hops!=null ){ |
| ret = true; |
| Hop.resetVisitStatus(hops); |
| for( Hop hop : hops ) |
| ret &= pruneHasOnlyUnknownMR(hop); |
| } |
| |
| return ret; |
| } |
| |
| /** |
| * |
| * @param hop |
| * @return |
| */ |
| private static boolean pruneHasOnlyUnknownMR( Hop hop ) |
| { |
| if( hop == null || hop.getVisited() == Hop.VisitStatus.DONE ) |
| return true; |
| |
| boolean ret = true; |
| |
| //process childs |
| for(Hop hi : hop.getInput()) |
| ret &= pruneHasOnlyUnknownMR( hi ); |
| |
| //investigate hop exec type and known dimensions |
| if( hop.getExecType()==ExecType.MR ) { |
| boolean lret = false; |
| |
| //1) operator output dimensions unknown |
| lret |= !hop.dimsKnown(); |
| |
| //2) operator output dimensions known but inputs unknown |
| //(use cases for e.g. AggUnary with scalar output, Binary with one known input) |
| for(Hop hi : hop.getInput()) |
| lret |= !hi.dimsKnown(); |
| |
| ret &= lret; |
| } |
| |
| hop.setVisited(Hop.VisitStatus.DONE); |
| |
| return ret; |
| } |
| |
| |
| /** |
| * |
| * @param prog |
| * @param cc |
| * @param type |
| * @return |
| * @throws DMLRuntimeException |
| * @throws HopsException |
| */ |
| private static ArrayList<Long> enumerateGridPoints( ArrayList<ProgramBlock> prog, long min, long max, GridEnumType type ) |
| throws DMLRuntimeException, HopsException |
| { |
| //create enumerator |
| GridEnumeration ge = null; |
| switch( type ){ |
| case EQUI_GRID: |
| ge = new GridEnumerationEqui(prog, min, max); break; |
| case EXP_GRID: |
| ge = new GridEnumerationExp(prog, min, max); break; |
| case MEM_EQUI_GRID: |
| ge = new GridEnumerationMemory(prog, min, max); break; |
| case HYBRID_MEM_EXP_GRID: |
| ge = new GridEnumerationHybrid(prog, min, max); break; |
| default: |
| throw new DMLRuntimeException("Unsupported grid enumeration type: "+type); |
| } |
| |
| //generate points |
| ArrayList<Long> ret = ge.enumerateGridPoints(); |
| LOG.debug("Gen: min="+YarnOptimizerUtils.toMB(min)+", max="+YarnOptimizerUtils.toMB(max)+", npoints="+ret.size()); |
| |
| return ret; |
| } |
| |
| /** |
| * |
| * @param Bp |
| * @param min |
| * @return |
| * @throws DMLRuntimeException |
| */ |
| private static double[][] initLocalMemoTable( ArrayList<ProgramBlock> Bp, double min ) |
| throws DMLRuntimeException |
| { |
| //allocate memo structure |
| int len = Bp.size(); |
| double[][] memo = new double[len][2]; |
| |
| //init with min resource and current costs |
| for( int i=0; i<len; i++ ) |
| { |
| ProgramBlock pb = Bp.get(i); |
| ExecutionContext ec = ExecutionContextFactory.createContext(); |
| memo[i][0] = min; |
| memo[i][1] = CostEstimationWrapper.getTimeEstimate(pb.getProgram(), ec); |
| } |
| |
| return memo; |
| } |
| |
| /** |
| * |
| * @param B |
| * @param Bp |
| * @param lmemo |
| * @param min |
| * @return |
| */ |
| private static double[][] initGlobalMemoTable( ArrayList<ProgramBlock> B, ArrayList<ProgramBlock> Bp, double[][] lmemo, double min ) |
| { |
| //allocate memo structure |
| int len = B.size(); |
| int lenp = Bp.size(); //lenp<=len |
| double[][] memo = new double[len][2]; |
| |
| //init with min resources |
| for( int i=0; i<len; i++ ) { |
| memo[i][0] = min; |
| memo[i][1] = -1; |
| } |
| |
| //overwrite existing values |
| int j = 0; |
| for( int i=0; i<len && j<lenp; i++ ) |
| { |
| ProgramBlock pb = B.get(i); |
| if( pb != Bp.get(j) ) |
| continue; |
| |
| //map local memo entry |
| memo[i][0] = lmemo[j][0]; |
| memo[i][1] = -1; |
| j++; |
| } |
| |
| return memo; |
| } |
| |
| /** |
| * |
| */ |
| public static void initStatistics() |
| { |
| _cntCompilePB = 0; |
| _cntCostPB = 0; |
| } |
| |
| |
| //////// |
| // old code |
| |
| public static long jvmToPhy(long jvm, boolean mrRealRun) { |
| long ret = (long) Math.ceil((double)jvm * DMLYarnClient.MEM_FACTOR); |
| if (mrRealRun) { |
| long lowerBound = (long)YarnClusterAnalyzer.getMinMRContarinerPhyMB() * 1024 * 1024; |
| if (ret < lowerBound) |
| return lowerBound; |
| } |
| return ret; |
| } |
| |
| public static long budgetToJvm(double budget) { |
| return (long) Math.ceil(budget / OptimizerUtils.MEM_UTIL_FACTOR); |
| } |
| |
| |
| public static double phyToBudget(long physical) throws IOException { |
| return (double)physical / DMLYarnClient.MEM_FACTOR * OptimizerUtils.MEM_UTIL_FACTOR; |
| } |
| |
| } |
