diff --git a/common/wuanalysis/anarule.cpp b/common/wuanalysis/anarule.cpp index 5f0924f9073..8c5dc2c6db9 100644 --- a/common/wuanalysis/anarule.cpp +++ b/common/wuanalysis/anarule.cpp @@ -135,14 +135,44 @@ class IoSkewRule : public AActivityRule stat_type timeAvgLocalExecute = activity.getStatRaw(StTimeLocalExecute, StAvgX); stat_type cost; - //If one node didn't spill then it is possible the skew caused all the lost time unsigned actkind = activity.getAttr(WaKind); - if ((actkind==TAKspillread||actkind==TAKspillwrite) && activity.getStatRaw(stat, StMinX) == 0) + if ((actkind==TAKspillread||actkind==TAKspillwrite) && (activity.getStatRaw(stat, StMinX) == 0)) + { + //If one node didn't spill then it is possible the skew caused all the lost time cost = timeMaxLocalExecute; + result.set(ANA_IOSKEW_RECORDS_ID, cost, "Uneven worker spilling is causing uneven %s time", category); + } else + { + bool fileSkew = false; + if (stat == StTimeDiskWriteIO) + { + IWuEdge *inputEdge; + unsigned i=0; + while (!fileSkew && (inputEdge=activity.queryInput(i++))) + { + //output IO is likely to be skewed because # rows written by the workers are skewed + if (inputEdge->getStatRaw(StNumRowsProcessed, StSkewMax)>options.queryOption(watOptSkewThreshold)) + fileSkew = true; + } + } + if (stat == StTimeDiskReadIO) + { + IWuEdge *outputEdge; + unsigned i=0; + while (!fileSkew && (outputEdge=activity.queryOutput(i++))) + { + //input IO is likely to be skewed because # rows read by the workers are skewed + if (outputEdge->getStatRaw(StNumRowsProcessed, StSkewMax)>options.queryOption(watOptSkewThreshold)) + fileSkew = true; + } + } cost = (timeMaxLocalExecute - timeAvgLocalExecute); - - result.set(ANA_IOSKEW_RECORDS_ID, cost, "Significant skew in records causes uneven %s time", category); + if (fileSkew) + result.set(ANA_IOSKEW_RECORDS_ID, cost, "Significant skew in datafile partitioning is causing uneven %s time", category); + else + result.set(ANA_IOSKEW_RECORDS_ID, cost, "Significant skew in IO performance may be causing uneven %s time", category); + } updateInformation(result, activity); return true; }