| /* |
| * 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.sysds.utils; |
| |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.HashSet; |
| import java.util.Map; |
| import java.util.List; |
| import java.util.Map.Entry; |
| import java.util.Stack; |
| |
| import org.apache.commons.lang3.mutable.MutableInt; |
| import org.apache.sysds.hops.Hop; |
| import org.apache.sysds.hops.LiteralOp; |
| import org.apache.sysds.hops.OptimizerUtils; |
| import org.apache.sysds.hops.codegen.cplan.CNode; |
| import org.apache.sysds.hops.codegen.cplan.CNodeMultiAgg; |
| import org.apache.sysds.hops.codegen.cplan.CNodeTpl; |
| import org.apache.sysds.hops.ipa.FunctionCallGraph; |
| import org.apache.sysds.lops.Lop; |
| import org.apache.sysds.parser.DMLProgram; |
| import org.apache.sysds.parser.ForStatement; |
| import org.apache.sysds.parser.ForStatementBlock; |
| import org.apache.sysds.parser.FunctionStatement; |
| import org.apache.sysds.parser.FunctionStatementBlock; |
| import org.apache.sysds.parser.IfStatement; |
| import org.apache.sysds.parser.IfStatementBlock; |
| import org.apache.sysds.parser.ParForStatementBlock; |
| import org.apache.sysds.parser.StatementBlock; |
| import org.apache.sysds.parser.WhileStatement; |
| import org.apache.sysds.parser.WhileStatementBlock; |
| import org.apache.sysds.runtime.controlprogram.BasicProgramBlock; |
| import org.apache.sysds.runtime.controlprogram.ForProgramBlock; |
| import org.apache.sysds.runtime.controlprogram.FunctionProgramBlock; |
| import org.apache.sysds.runtime.controlprogram.IfProgramBlock; |
| import org.apache.sysds.runtime.controlprogram.ParForProgramBlock; |
| import org.apache.sysds.runtime.controlprogram.Program; |
| import org.apache.sysds.runtime.controlprogram.ProgramBlock; |
| import org.apache.sysds.runtime.controlprogram.WhileProgramBlock; |
| import org.apache.sysds.runtime.controlprogram.context.SparkExecutionContext; |
| import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer; |
| import org.apache.sysds.runtime.instructions.Instruction; |
| import org.apache.sysds.runtime.instructions.cp.CPInstruction; |
| import org.apache.sysds.runtime.instructions.fed.FEDInstruction; |
| import org.apache.sysds.runtime.instructions.gpu.GPUInstruction; |
| import org.apache.sysds.runtime.instructions.spark.CSVReblockSPInstruction; |
| import org.apache.sysds.runtime.instructions.spark.CheckpointSPInstruction; |
| import org.apache.sysds.runtime.instructions.spark.ReblockSPInstruction; |
| import org.apache.sysds.runtime.instructions.spark.SPInstruction; |
| import org.apache.sysds.runtime.lineage.LineageItem; |
| import org.apache.sysds.runtime.lineage.LineageItemUtils; |
| |
| public class Explain |
| { |
| //internal configuration parameters |
| private static final boolean REPLACE_SPECIAL_CHARACTERS = true; |
| private static final boolean SHOW_MEM_ABOVE_BUDGET = true; |
| private static final boolean SHOW_LITERAL_HOPS = false; |
| private static final boolean SHOW_DATA_DEPENDENCIES = true; |
| private static final boolean SHOW_DATA_FLOW_PROPERTIES = true; |
| |
| //different explain levels |
| public enum ExplainType { |
| NONE, // explain disabled |
| HOPS, // explain program and hops |
| RUNTIME, // explain runtime program (default) |
| RECOMPILE_HOPS, // explain hops, incl recompile |
| RECOMPILE_RUNTIME; // explain runtime program, incl recompile |
| |
| public boolean isHopsType(boolean recompile) { |
| return (this==RECOMPILE_HOPS || (!recompile && this==HOPS)); |
| } |
| public boolean isRuntimeType(boolean recompile) { |
| return (this==RECOMPILE_RUNTIME || (!recompile && this==RUNTIME)); |
| } |
| } |
| |
| public static class ExplainCounts { |
| public int numCPInst = 0; |
| public int numJobs = 0; |
| public int numReblocks = 0; |
| public int numChkpts = 0; |
| } |
| |
| ////////////// |
| // public explain interface |
| |
| public static String display(DMLProgram prog, Program rtprog, ExplainType type, ExplainCounts counts) { |
| if( counts == null ) |
| counts = countDistributedOperations(rtprog); |
| |
| //explain plan of program (hops or runtime) |
| return "# EXPLAIN ("+type.name()+"):\n" |
| + Explain.explainMemoryBudget(counts)+"\n" |
| + Explain.explainDegreeOfParallelism(counts) |
| + Explain.explain(prog, rtprog, type, counts); |
| } |
| |
| public static String explainMemoryBudget() { |
| return explainMemoryBudget(new ExplainCounts()); |
| } |
| |
| public static String explainMemoryBudget(ExplainCounts counts) { |
| StringBuilder sb = new StringBuilder(); |
| sb.append( "# Memory Budget local/remote = " ); |
| sb.append( OptimizerUtils.toMB(OptimizerUtils.getLocalMemBudget()) ); |
| sb.append( "MB/" ); |
| |
| if( OptimizerUtils.isSparkExecutionMode() ) { |
| //avoid unnecessary lazy spark context creation on access to memory configurations |
| if( counts.numJobs-counts.numReblocks-counts.numChkpts <= 0 |
| || !SparkExecutionContext.isSparkContextCreated() ) { |
| sb.append( "?MB/?MB/?MB" ); |
| } |
| else { //default |
| sb.append( OptimizerUtils.toMB(SparkExecutionContext.getDataMemoryBudget(true, false)) ); |
| sb.append( "MB/" ); |
| sb.append( OptimizerUtils.toMB(SparkExecutionContext.getDataMemoryBudget(false, false)) ); |
| sb.append( "MB/" ); |
| sb.append( OptimizerUtils.toMB(SparkExecutionContext.getBroadcastMemoryBudget()) ); |
| sb.append( "MB" ); |
| } |
| } |
| else { |
| sb.append( "?MB/?MB" ); |
| } |
| |
| return sb.toString(); |
| } |
| |
| public static String explainDegreeOfParallelism() { |
| return explainDegreeOfParallelism(new ExplainCounts()); |
| } |
| |
| public static String explainDegreeOfParallelism(ExplainCounts counts) { |
| int lk = InfrastructureAnalyzer.getLocalParallelism(); |
| StringBuilder sb = new StringBuilder(); |
| sb.append( "# Degree of Parallelism (vcores) local/remote = " ); |
| sb.append( lk ); |
| sb.append( "/" ); |
| |
| if( OptimizerUtils.isSparkExecutionMode() ) { |
| if( counts.numJobs-counts.numReblocks-counts.numChkpts <= 0 |
| || !SparkExecutionContext.isSparkContextCreated() ) { |
| //avoid unnecessary lazy spark context creation on access to memory configurations |
| sb.append( "?" ); |
| } |
| else { //default |
| sb.append( SparkExecutionContext.getDefaultParallelism(false) ); |
| } |
| } |
| |
| return sb.toString(); |
| } |
| |
| public static String explain(DMLProgram prog, Program rtprog, ExplainType type) { |
| return explain(prog, rtprog, type, null); |
| } |
| |
| public static String explain(DMLProgram prog, Program rtprog, ExplainType type, ExplainCounts counts) { |
| //dispatch to individual explain utils |
| switch( type ) { |
| //explain hops with stats |
| case HOPS: |
| case RECOMPILE_HOPS: |
| return explain(prog); |
| //explain runtime program |
| case RUNTIME: |
| case RECOMPILE_RUNTIME: |
| return explain(rtprog, counts); |
| case NONE: |
| //do nothing |
| } |
| |
| return null; |
| } |
| |
| public static String explain(DMLProgram prog) |
| { |
| StringBuilder sb = new StringBuilder(); |
| |
| //create header |
| sb.append("\nPROGRAM\n"); |
| |
| // Explain functions (if exists) |
| if( prog.hasFunctionStatementBlocks() ) { |
| sb.append("--FUNCTIONS\n"); |
| |
| //show function call graph |
| sb.append("----FUNCTION CALL GRAPH\n"); |
| sb.append("------MAIN PROGRAM\n"); |
| FunctionCallGraph fgraph = new FunctionCallGraph(prog); |
| sb.append(explainFunctionCallGraph(fgraph, new HashSet<String>(), null, 3)); |
| |
| //show individual functions |
| for (String namespace : prog.getNamespaces().keySet()) { |
| for (String fname : prog.getFunctionStatementBlocks(namespace).keySet()) { |
| FunctionStatementBlock fsb = prog.getFunctionStatementBlock(namespace, fname); |
| FunctionStatement fstmt = (FunctionStatement) fsb.getStatement(0); |
| String fkey = DMLProgram.constructFunctionKey(namespace, fname); |
| sb.append("----FUNCTION " + fkey + " [recompile="+fsb.isRecompileOnce()+"]\n"); |
| for (StatementBlock current : fstmt.getBody()) |
| sb.append(explainStatementBlock(current, 3)); |
| } |
| } |
| } |
| |
| // Explain main program |
| sb.append("--MAIN PROGRAM\n"); |
| for( StatementBlock sblk : prog.getStatementBlocks() ) |
| sb.append(explainStatementBlock(sblk, 2)); |
| |
| return sb.toString(); |
| } |
| |
| public static String explain( Program rtprog ) { |
| return explain(rtprog, null); |
| } |
| |
| public static String explain( Program rtprog, ExplainCounts counts ) |
| { |
| //counts number of instructions |
| boolean sparkExec = OptimizerUtils.isSparkExecutionMode(); |
| if( counts == null ) { |
| counts = new ExplainCounts(); |
| countCompiledInstructions(rtprog, counts, true, sparkExec); |
| } |
| |
| StringBuilder sb = new StringBuilder(); |
| |
| //create header |
| sb.append("\nPROGRAM ( size CP/"+(sparkExec?"SP":"MR")+" = "); |
| sb.append(counts.numCPInst); |
| sb.append("/"); |
| sb.append(counts.numJobs); |
| sb.append(" )\n"); |
| |
| //explain functions (if exists) |
| Map<String, FunctionProgramBlock> funcMap = rtprog.getFunctionProgramBlocks(); |
| if( funcMap != null && !funcMap.isEmpty() ) |
| { |
| sb.append("--FUNCTIONS\n"); |
| |
| //show function call graph |
| if( !rtprog.getProgramBlocks().isEmpty() && |
| rtprog.getProgramBlocks().get(0).getStatementBlock() != null ) |
| { |
| sb.append("----FUNCTION CALL GRAPH\n"); |
| sb.append("------MAIN PROGRAM\n"); |
| DMLProgram prog = rtprog.getProgramBlocks().get(0).getStatementBlock().getDMLProg(); |
| FunctionCallGraph fgraph = new FunctionCallGraph(prog); |
| sb.append(explainFunctionCallGraph(fgraph, new HashSet<String>(), null, 3)); |
| } |
| |
| //show individual functions |
| for( Entry<String, FunctionProgramBlock> e : funcMap.entrySet() ) { |
| String fkey = e.getKey(); |
| FunctionProgramBlock fpb = e.getValue(); |
| //explain optimized function |
| sb.append("----FUNCTION "+fkey+" [recompile="+fpb.isRecompileOnce()+"]\n"); |
| for( ProgramBlock pb : fpb.getChildBlocks() ) |
| sb.append( explainProgramBlock(pb,3) ); |
| //explain unoptimized function |
| if( rtprog.containsFunctionProgramBlock(fkey, false) ) { |
| FunctionProgramBlock fpb2 = rtprog.getFunctionProgramBlock(fkey, false); |
| sb.append("----FUNCTION "+fkey+" (unoptimized) [recompile="+fpb2.isRecompileOnce()+"]\n"); |
| for( ProgramBlock pb : fpb2.getChildBlocks() ) |
| sb.append( explainProgramBlock(pb,3) ); |
| } |
| } |
| } |
| |
| //explain main program |
| sb.append("--MAIN PROGRAM\n"); |
| for( ProgramBlock pb : rtprog.getProgramBlocks() ) |
| sb.append( explainProgramBlock(pb,2) ); |
| |
| return sb.toString(); |
| } |
| |
| public static String explain( ProgramBlock pb ) { |
| return explainProgramBlock(pb, 0); |
| } |
| |
| public static String explain( ArrayList<Instruction> inst ) { |
| return explainInstructions(inst, 0); |
| } |
| |
| public static String explain( ArrayList<Instruction> inst, int level ) { |
| return explainInstructions(inst, level); |
| } |
| |
| public static String explain( Instruction inst ) { |
| return explainGenericInstruction(inst, 0); |
| } |
| |
| public static String explain( StatementBlock sb ) { |
| return explainStatementBlock(sb, 0); |
| } |
| |
| public static String explainHops( ArrayList<Hop> hops ) { |
| return explainHops(hops, 0); |
| } |
| |
| public static String explainHops( ArrayList<Hop> hops, int level ) { |
| StringBuilder sb = new StringBuilder(); |
| Hop.resetVisitStatus(hops); |
| for( Hop hop : hops ) |
| sb.append(explainHop(hop, level)); |
| Hop.resetVisitStatus(hops); |
| return sb.toString(); |
| } |
| |
| public static String explain( Hop hop ) { |
| return explain(hop, 0); |
| } |
| |
| public static String explain( Hop hop, int level ) { |
| hop.resetVisitStatus(); |
| String ret = explainHop(hop, level); |
| hop.resetVisitStatus(); |
| return ret; |
| } |
| |
| public static String explainLineageItems( LineageItem[] lis ) { |
| return explainLineageItems(lis, 0); |
| } |
| |
| public static String explainLineageItems( LineageItem[] lis, int level ) { |
| StringBuilder sb = new StringBuilder(); |
| LineageItem.resetVisitStatusNR(lis); |
| for( LineageItem li : lis ) |
| sb.append(explainLineageItemNR(li, level)); |
| LineageItem.resetVisitStatusNR(lis); |
| return sb.toString(); |
| } |
| |
| public static String explain( LineageItem li ) { |
| li.resetVisitStatusNR(); |
| String s = explain(li, 0); |
| //s += rExplainDedupItems(li, new ArrayList<>()); |
| li.resetVisitStatusNR(); |
| return s; |
| } |
| |
| private static String explain( LineageItem li, int level ) { |
| li.resetVisitStatusNR(); |
| String ret = explainLineageItemNR(li, level); |
| li.resetVisitStatusNR(); |
| return ret; |
| } |
| |
| @Deprecated |
| @SuppressWarnings("unused") |
| private static String rExplainDedupItems(LineageItem li, List<String> paths) { |
| if (li.isVisited()) |
| return ""; |
| StringBuilder sb = new StringBuilder(); |
| |
| if (li.getType() == LineageItem.LineageItemType.Dedup && !paths.contains(li.getData())) { |
| sb.append("\n").append("dedup").append(li.getData()).append(":\n"); |
| sb.append(Explain.explain(li, 0)); |
| paths.add(li.getData()); |
| } |
| |
| if (li.getInputs() != null) |
| for (LineageItem in : li.getInputs()) |
| sb.append(rExplainDedupItems(in, paths)); |
| |
| li.setVisited(); |
| return sb.toString(); |
| } |
| |
| |
| public static String explainCPlan( CNodeTpl cplan ) { |
| StringBuilder sb = new StringBuilder(); |
| |
| //create template header |
| sb.append("\n----------------------------------------\n"); |
| sb.append("CPLAN: "+cplan.getTemplateInfo()+"\n"); |
| sb.append("--inputs: "+Arrays.toString(cplan.getInputNames())+"\n"); |
| sb.append("----------------------------------------\n"); |
| |
| //explain body dag |
| cplan.resetVisitStatusOutputs(); |
| if( cplan instanceof CNodeMultiAgg ) |
| for( CNode output : ((CNodeMultiAgg)cplan).getOutputs() ) |
| sb.append(explainCNode(output, 1)); |
| else |
| sb.append(explainCNode(cplan.getOutput(), 1)); |
| cplan.resetVisitStatusOutputs(); |
| sb.append("----------------------------------------\n"); |
| |
| return sb.toString(); |
| } |
| |
| public static String explain( CNode node ) { |
| return explain(node, 0); |
| } |
| |
| public static String explain( CNode node, int level ) { |
| return explainCNode(node, level); |
| } |
| |
| /** |
| * Counts the number of compiled MRJob/Spark instructions in the |
| * given runtime program. |
| * |
| * @param rtprog runtime program |
| * @return counts |
| */ |
| public static ExplainCounts countDistributedOperations( Program rtprog ) { |
| ExplainCounts counts = new ExplainCounts(); |
| Explain.countCompiledInstructions(rtprog, counts, true, true); |
| return counts; |
| } |
| |
| public static String getIdentation( int level ) { |
| return createOffset(level); |
| } |
| |
| ////////////// |
| // internal explain HOPS |
| |
| private static String explainStatementBlock(StatementBlock sb, int level) |
| { |
| StringBuilder builder = new StringBuilder(); |
| String offset = createOffset(level); |
| |
| if (sb instanceof WhileStatementBlock) { |
| WhileStatementBlock wsb = (WhileStatementBlock) sb; |
| builder.append(offset); |
| if( !wsb.getUpdateInPlaceVars().isEmpty() ) |
| builder.append("WHILE (lines "+wsb.getBeginLine()+"-"+wsb.getEndLine()+") [in-place="+wsb.getUpdateInPlaceVars().toString()+"]\n"); |
| else |
| builder.append("WHILE (lines "+wsb.getBeginLine()+"-"+wsb.getEndLine()+")\n"); |
| builder.append(explainHop(wsb.getPredicateHops(), level+1)); |
| |
| WhileStatement ws = (WhileStatement)sb.getStatement(0); |
| for (StatementBlock current : ws.getBody()) |
| builder.append(explainStatementBlock(current, level+1)); |
| |
| } |
| else if (sb instanceof IfStatementBlock) { |
| IfStatementBlock ifsb = (IfStatementBlock) sb; |
| builder.append(offset); |
| builder.append("IF (lines "+ifsb.getBeginLine()+"-"+ifsb.getEndLine()+")\n"); |
| builder.append(explainHop(ifsb.getPredicateHops(), level+1)); |
| |
| IfStatement ifs = (IfStatement) sb.getStatement(0); |
| for (StatementBlock current : ifs.getIfBody()) |
| builder.append(explainStatementBlock(current, level+1)); |
| if( !ifs.getElseBody().isEmpty() ) { |
| builder.append(offset); |
| builder.append("ELSE\n"); |
| } |
| for (StatementBlock current : ifs.getElseBody()) |
| builder.append(explainStatementBlock(current, level+1)); |
| |
| } |
| else if (sb instanceof ForStatementBlock) { |
| ForStatementBlock fsb = (ForStatementBlock) sb; |
| builder.append(offset); |
| if (sb instanceof ParForStatementBlock) { |
| if( !fsb.getUpdateInPlaceVars().isEmpty() ) |
| builder.append("PARFOR (lines "+fsb.getBeginLine()+"-"+fsb.getEndLine()+") [in-place="+fsb.getUpdateInPlaceVars().toString()+"]\n"); |
| else |
| builder.append("PARFOR (lines "+fsb.getBeginLine()+"-"+fsb.getEndLine()+")\n"); |
| } |
| else { |
| if( !fsb.getUpdateInPlaceVars().isEmpty() ) |
| builder.append("FOR (lines "+fsb.getBeginLine()+"-"+fsb.getEndLine()+") [in-place="+fsb.getUpdateInPlaceVars().toString()+"]\n"); |
| else |
| builder.append("FOR (lines "+fsb.getBeginLine()+"-"+fsb.getEndLine()+")\n"); |
| } |
| if (fsb.getFromHops() != null) |
| builder.append(explainHop(fsb.getFromHops(), level+1)); |
| if (fsb.getToHops() != null) |
| builder.append(explainHop(fsb.getToHops(), level+1)); |
| if (fsb.getIncrementHops() != null) |
| builder.append(explainHop(fsb.getIncrementHops(), level+1)); |
| |
| ForStatement fs = (ForStatement)sb.getStatement(0); |
| for (StatementBlock current : fs.getBody()) |
| builder.append(explainStatementBlock(current, level+1)); |
| |
| } |
| else if (sb instanceof FunctionStatementBlock) { |
| FunctionStatement fsb = (FunctionStatement) sb.getStatement(0); |
| for (StatementBlock current : fsb.getBody()) |
| builder.append(explainStatementBlock(current, level+1)); |
| |
| } |
| else { |
| // For generic StatementBlock |
| builder.append(offset); |
| builder.append("GENERIC (lines "+sb.getBeginLine()+"-"+sb.getEndLine()+") [recompile=" + sb.requiresRecompilation() + "]\n"); |
| ArrayList<Hop> hopsDAG = sb.getHops(); |
| if( hopsDAG != null && !hopsDAG.isEmpty() ) { |
| Hop.resetVisitStatus(hopsDAG); |
| for (Hop hop : hopsDAG) |
| builder.append(explainHop(hop, level+1)); |
| Hop.resetVisitStatus(hopsDAG); |
| } |
| } |
| |
| return builder.toString(); |
| } |
| |
| /** |
| * Do a post-order traverse through the Hop DAG and explain each Hop |
| * |
| * @param hop high-level operator |
| * @param level offset |
| * @return string explanation of Hop DAG |
| */ |
| private static String explainHop(Hop hop, int level) { |
| if( hop.isVisited() || (!SHOW_LITERAL_HOPS && hop instanceof LiteralOp) ) |
| return ""; |
| |
| StringBuilder sb = new StringBuilder(); |
| String offset = createOffset(level); |
| |
| for( Hop input : hop.getInput() ) |
| sb.append(explainHop(input, level)); |
| |
| //indentation |
| sb.append(offset); |
| |
| //hop id |
| if( SHOW_DATA_DEPENDENCIES ) |
| sb.append("("+hop.getHopID()+") "); |
| |
| //operation string |
| sb.append(hop.getOpString()); |
| |
| //input hop references |
| if( SHOW_DATA_DEPENDENCIES ) { |
| StringBuilder childs = new StringBuilder(); |
| childs.append(" ("); |
| boolean childAdded = false; |
| for( Hop input : hop.getInput() ) |
| if( SHOW_LITERAL_HOPS || !(input instanceof LiteralOp) ){ |
| childs.append(childAdded?",":""); |
| childs.append(input.getHopID()); |
| childAdded = true; |
| } |
| childs.append(")"); |
| if( childAdded ) |
| sb.append(childs.toString()); |
| } |
| |
| //matrix characteristics |
| sb.append(" [" + hop.getDim1() + "," |
| + hop.getDim2() + "," |
| + hop.getBlocksize() + "," |
| + hop.getNnz()); |
| |
| if (hop.getUpdateType().isInPlace()) |
| sb.append("," + hop.getUpdateType().toString().toLowerCase()); |
| |
| sb.append("]"); |
| |
| //memory estimates |
| sb.append(" [" + showMem(hop.getInputMemEstimate(), false) + "," |
| + showMem(hop.getIntermediateMemEstimate(), false) + "," |
| + showMem(hop.getOutputMemEstimate(), false) + " -> " |
| + showMem(hop.getMemEstimate(), true) + "]"); |
| |
| //data flow properties |
| if( SHOW_DATA_FLOW_PROPERTIES ) { |
| if( hop.requiresReblock() && hop.requiresCheckpoint() ) |
| sb.append(" [rblk,chkpt]"); |
| else if( hop.requiresReblock() ) |
| sb.append(" [rblk]"); |
| else if( hop.requiresCheckpoint() ) |
| sb.append(" [chkpt]"); |
| } |
| |
| //exec type |
| if (hop.getExecType() != null) |
| sb.append(", " + hop.getExecType()); |
| |
| sb.append('\n'); |
| |
| hop.setVisited(); |
| |
| return sb.toString(); |
| } |
| |
| private static String explainLineageItemNR(LineageItem item, int level) { |
| //NOTE: in contrast to similar non-recursive functions like resetVisitStatusNR, |
| // we maintain a more complex stack to ensure DFS ordering where current nodes |
| // are added after the subtree underneath is processed (backwards compatibility) |
| Stack<LineageItem> stackItem = new Stack<>(); |
| Stack<MutableInt> stackPos = new Stack<>(); |
| stackItem.push(item); stackPos.push(new MutableInt(0)); |
| StringBuilder sb = new StringBuilder(); |
| while( !stackItem.empty() ) { |
| LineageItem tmpItem = stackItem.peek(); |
| MutableInt tmpPos = stackPos.peek(); |
| //check ascent condition - no item processing |
| if( tmpItem.isVisited() ) { |
| stackItem.pop(); stackPos.pop(); |
| } |
| //check ascent condition - append item |
| else if( tmpItem.getInputs() == null |
| || tmpItem.getOpcode().startsWith(LineageItemUtils.LPLACEHOLDER) |
| // don't trace beyond if a placeholder is found |
| || tmpItem.getInputs().length <= tmpPos.intValue() ) { |
| sb.append(createOffset(level)); |
| sb.append(tmpItem.toString()); |
| sb.append('\n'); |
| stackItem.pop(); stackPos.pop(); |
| tmpItem.setVisited(); |
| } |
| //check descent condition |
| else if( tmpItem.getInputs() != null ) { |
| stackItem.push(tmpItem.getInputs()[tmpPos.intValue()]); |
| tmpPos.increment(); |
| stackPos.push(new MutableInt(0)); |
| } |
| } |
| return sb.toString(); |
| } |
| |
| @Deprecated |
| @SuppressWarnings("unused") |
| private static String explainLineageItem(LineageItem li, int level) { |
| if( li.isVisited()) |
| return ""; |
| |
| StringBuilder sb = new StringBuilder(); |
| String offset = createOffset(level); |
| |
| if (li.getInputs() != null) |
| for( LineageItem input : li.getInputs() ) |
| sb.append(explainLineageItem(input, level)); |
| |
| sb.append(offset); |
| sb.append(li.toString()); |
| sb.append('\n'); |
| |
| li.setVisited(); |
| |
| return sb.toString(); |
| } |
| |
| ////////////// |
| // internal explain CNODE |
| |
| private static String explainCNode(CNode cnode, int level) { |
| if( cnode.isVisited() ) |
| return ""; |
| |
| StringBuilder sb = new StringBuilder(); |
| String offset = createOffset(level); |
| |
| for( CNode input : cnode.getInput() ) |
| sb.append(explainCNode(input, level)); |
| |
| //indentation |
| sb.append(offset); |
| |
| //hop id |
| if( SHOW_DATA_DEPENDENCIES ) |
| sb.append("("+cnode.getID()+") "); |
| |
| //operation string |
| sb.append(cnode.toString()); |
| |
| //input hop references |
| if( SHOW_DATA_DEPENDENCIES ) { |
| StringBuilder childs = new StringBuilder(); |
| childs.append(" ("); |
| boolean childAdded = false; |
| for( CNode input : cnode.getInput() ) { |
| childs.append(childAdded?",":""); |
| childs.append(input.getID()); |
| childAdded = true; |
| } |
| childs.append(")"); |
| if( childAdded ) |
| sb.append(childs.toString()); |
| } |
| |
| sb.append('\n'); |
| cnode.setVisited(); |
| |
| return sb.toString(); |
| } |
| |
| ////////////// |
| // internal explain RUNTIME |
| |
| private static String explainProgramBlock( ProgramBlock pb, int level ) |
| { |
| StringBuilder sb = new StringBuilder(); |
| String offset = createOffset(level); |
| |
| if (pb instanceof FunctionProgramBlock ) { |
| FunctionProgramBlock fpb = (FunctionProgramBlock)pb; |
| for( ProgramBlock pbc : fpb.getChildBlocks() ) |
| sb.append( explainProgramBlock( pbc, level+1) ); |
| } |
| else if (pb instanceof WhileProgramBlock) { |
| WhileProgramBlock wpb = (WhileProgramBlock) pb; |
| StatementBlock wsb = pb.getStatementBlock(); |
| sb.append(offset); |
| if( wsb != null && !wsb.getUpdateInPlaceVars().isEmpty() ) |
| sb.append("WHILE (lines "+wpb.getBeginLine()+"-"+wpb.getEndLine()+") [in-place="+wsb.getUpdateInPlaceVars().toString()+"]\n"); |
| else |
| sb.append("WHILE (lines "+wpb.getBeginLine()+"-"+wpb.getEndLine()+")\n"); |
| sb.append(explainInstructions(wpb.getPredicate(), level+1)); |
| for( ProgramBlock pbc : wpb.getChildBlocks() ) |
| sb.append( explainProgramBlock( pbc, level+1) ); |
| if( wpb.getExitInstruction() != null ) |
| sb.append(explainInstructions(wpb.getExitInstruction(), level+1)); |
| } |
| else if (pb instanceof IfProgramBlock) { |
| IfProgramBlock ipb = (IfProgramBlock) pb; |
| sb.append(offset); |
| sb.append("IF (lines "+ipb.getBeginLine()+"-"+ipb.getEndLine()+")\n"); |
| sb.append(explainInstructions(ipb.getPredicate(), level+1)); |
| for( ProgramBlock pbc : ipb.getChildBlocksIfBody() ) |
| sb.append( explainProgramBlock( pbc, level+1) ); |
| if( !ipb.getChildBlocksElseBody().isEmpty() ) { |
| sb.append(offset); |
| sb.append("ELSE\n"); |
| for( ProgramBlock pbc : ipb.getChildBlocksElseBody() ) |
| sb.append( explainProgramBlock( pbc, level+1) ); |
| } |
| if( ipb.getExitInstruction() != null ) |
| sb.append(explainInstructions(ipb.getExitInstruction(), level+1)); |
| } |
| else if (pb instanceof ForProgramBlock) { //incl parfor |
| ForProgramBlock fpb = (ForProgramBlock) pb; |
| StatementBlock fsb = pb.getStatementBlock(); |
| sb.append(offset); |
| if( pb instanceof ParForProgramBlock ) |
| sb.append("PARFOR (lines "+fpb.getBeginLine()+"-"+fpb.getEndLine()+")\n"); |
| else { |
| if( fsb != null && !fsb.getUpdateInPlaceVars().isEmpty() ) |
| sb.append("FOR (lines "+fpb.getBeginLine()+"-"+fpb.getEndLine()+") [in-place="+fsb.getUpdateInPlaceVars().toString()+"]\n"); |
| else |
| sb.append("FOR (lines "+fpb.getBeginLine()+"-"+fpb.getEndLine()+")\n"); |
| } |
| sb.append(explainInstructions(fpb.getFromInstructions(), level+1)); |
| sb.append(explainInstructions(fpb.getToInstructions(), level+1)); |
| sb.append(explainInstructions(fpb.getIncrementInstructions(), level+1)); |
| for( ProgramBlock pbc : fpb.getChildBlocks() ) |
| sb.append( explainProgramBlock( pbc, level+1) ); |
| if( fpb.getExitInstruction() != null ) |
| sb.append(explainInstructions(fpb.getExitInstruction(), level+1)); |
| } |
| else if( pb instanceof BasicProgramBlock ) { |
| BasicProgramBlock bpb = (BasicProgramBlock) pb; |
| sb.append(offset); |
| if( pb.getStatementBlock()!=null ) |
| sb.append("GENERIC (lines "+pb.getBeginLine()+"-"+pb.getEndLine()+") [recompile="+pb.getStatementBlock().requiresRecompilation()+"]\n"); |
| else |
| sb.append("GENERIC (lines "+pb.getBeginLine()+"-"+pb.getEndLine()+") \n"); |
| sb.append(explainInstructions(bpb.getInstructions(), level+1)); |
| } |
| |
| return sb.toString(); |
| } |
| |
| private static String explainInstructions( ArrayList<Instruction> instSet, int level ) { |
| StringBuilder sb = new StringBuilder(); |
| String offsetInst = createOffset(level); |
| for( Instruction inst : instSet ) { |
| String tmp = explainGenericInstruction(inst, level); |
| sb.append( offsetInst ); |
| sb.append( tmp ); |
| sb.append( '\n' ); |
| } |
| |
| return sb.toString(); |
| } |
| |
| private static String explainInstructions( Instruction inst, int level ) { |
| StringBuilder sb = new StringBuilder(); |
| sb.append( createOffset(level) ); |
| sb.append( explainGenericInstruction(inst, level) ); |
| sb.append( '\n' ); |
| return sb.toString(); |
| } |
| |
| private static String explainGenericInstruction( Instruction inst, int level ) |
| { |
| String tmp = null; |
| if ( inst instanceof SPInstruction || inst instanceof CPInstruction || inst instanceof GPUInstruction || |
| inst instanceof FEDInstruction ) |
| tmp = inst.toString(); |
| |
| if( REPLACE_SPECIAL_CHARACTERS ){ |
| tmp = tmp.replaceAll(Lop.OPERAND_DELIMITOR, " "); |
| tmp = tmp.replaceAll(Lop.DATATYPE_PREFIX, "."); |
| tmp = tmp.replaceAll(Lop.INSTRUCTION_DELIMITOR, ", "); |
| } |
| |
| return tmp; |
| } |
| |
| @SuppressWarnings("unused") |
| private static String showMem(double mem, boolean units) |
| { |
| if( !SHOW_MEM_ABOVE_BUDGET && mem >= OptimizerUtils.DEFAULT_SIZE ) |
| return "MAX"; |
| return OptimizerUtils.toMB(mem) + (units?"MB":""); |
| } |
| |
| private static String createOffset( int level ) |
| { |
| StringBuilder sb = new StringBuilder(); |
| for( int i=0; i<level; i++ ) |
| sb.append("--"); |
| return sb.toString(); |
| } |
| |
| private static void countCompiledInstructions( Program rtprog, ExplainCounts counts, boolean CP, boolean SP ) |
| { |
| //analyze DML-bodied functions |
| for( FunctionProgramBlock fpb : rtprog.getFunctionProgramBlocks().values() ) |
| countCompiledInstructions( fpb, counts, CP, SP ); |
| |
| //analyze main program |
| for( ProgramBlock pb : rtprog.getProgramBlocks() ) |
| countCompiledInstructions( pb, counts, CP, SP ); |
| } |
| |
| /** |
| * Recursively counts the number of compiled MRJob instructions in the |
| * given runtime program block. |
| * |
| * @param pb program block |
| * @param counts explain countst |
| * @param CP if true, count CP instructions |
| * @param SP if true, count Spark instructions |
| */ |
| private static void countCompiledInstructions(ProgramBlock pb, ExplainCounts counts, boolean CP, boolean SP) |
| { |
| if (pb instanceof WhileProgramBlock) { |
| WhileProgramBlock tmp = (WhileProgramBlock)pb; |
| countCompiledInstructions(tmp.getPredicate(), counts, CP, SP); |
| for (ProgramBlock pb2 : tmp.getChildBlocks()) |
| countCompiledInstructions(pb2, counts, CP, SP); |
| } |
| else if (pb instanceof IfProgramBlock) { |
| IfProgramBlock tmp = (IfProgramBlock)pb; |
| countCompiledInstructions(tmp.getPredicate(), counts, CP, SP); |
| for( ProgramBlock pb2 : tmp.getChildBlocksIfBody() ) |
| countCompiledInstructions(pb2, counts, CP, SP); |
| for( ProgramBlock pb2 : tmp.getChildBlocksElseBody() ) |
| countCompiledInstructions(pb2, counts, CP, SP); |
| } |
| else if (pb instanceof ForProgramBlock) { //includes ParFORProgramBlock |
| ForProgramBlock tmp = (ForProgramBlock)pb; |
| countCompiledInstructions(tmp.getFromInstructions(), counts, CP, SP); |
| countCompiledInstructions(tmp.getToInstructions(), counts, CP, SP); |
| countCompiledInstructions(tmp.getIncrementInstructions(), counts, CP, SP); |
| for( ProgramBlock pb2 : tmp.getChildBlocks() ) |
| countCompiledInstructions(pb2, counts, CP, SP); |
| //additional parfor jobs counted during runtime |
| } |
| else if ( pb instanceof FunctionProgramBlock ) { |
| FunctionProgramBlock fpb = (FunctionProgramBlock)pb; |
| for( ProgramBlock pb2 : fpb.getChildBlocks() ) |
| countCompiledInstructions(pb2, counts, CP, SP); |
| } |
| else if( pb instanceof BasicProgramBlock ) { |
| BasicProgramBlock bpb = (BasicProgramBlock) pb; |
| countCompiledInstructions(bpb.getInstructions(), counts, CP, SP); |
| } |
| } |
| |
| /** |
| * Count the number of Hadoop instructions, CP instructions, Spark |
| * instructions, and/or Spark reblock instructions in a list of |
| * instructions. |
| * |
| * @param instSet |
| * list of instructions |
| * @param counts |
| * explain counts |
| * @param CP |
| * if true, count CP instructions |
| * @param SP |
| * if true, count Spark instructions and Spark reblock |
| * instructions |
| */ |
| private static void countCompiledInstructions( ArrayList<Instruction> instSet, ExplainCounts counts, boolean CP, boolean SP ) |
| { |
| for( Instruction inst : instSet ) |
| { |
| if( CP && inst instanceof CPInstruction ) |
| counts.numCPInst++; |
| else if( SP && inst instanceof SPInstruction ) |
| counts.numJobs++; |
| |
| //keep track of reblocks (in order to prevent unnecessary spark context creation) |
| if( SP && (inst instanceof CSVReblockSPInstruction || inst instanceof ReblockSPInstruction) ) |
| counts.numReblocks++; |
| if( SP && inst instanceof CheckpointSPInstruction ) |
| counts.numChkpts++; |
| } |
| } |
| |
| private static String explainFunctionCallGraph(FunctionCallGraph fgraph, HashSet<String> fstack, String fkey, int level) |
| { |
| StringBuilder builder = new StringBuilder(); |
| String offset = createOffset(level); |
| Collection<String> cfkeys = fgraph.getCalledFunctions(fkey); |
| if( cfkeys != null ) { |
| for( String cfkey : cfkeys ) { |
| if( fstack.contains(cfkey) && fgraph.isRecursiveFunction(cfkey) ) |
| builder.append(offset + "--" + cfkey + " (recursive)\n"); |
| else { |
| fstack.add(cfkey); |
| builder.append(offset + "--" + cfkey + "\n"); |
| builder.append(explainFunctionCallGraph(fgraph, fstack, cfkey, level+1)); |
| fstack.remove(cfkey); |
| } |
| } |
| } |
| |
| return builder.toString(); |
| } |
| } |
