edmStreamStallGrapher Namespace Reference

Classes
class	Point
class	Stack
class	StallMonitorParser
class	StreamInfoElement
class	TracerParser

Functions
def	adjacentDiff (pairLists)
def	chooseParser (inputFile)
def	consolidateContiguousBlocks (numStreams, streamInfo)
def	createAsciiImage (processingSteps, numStreams, maxNameSize)
def	createPDFImage (pdfFile, shownStacks, processingSteps, numStreams, stalledModuleInfo, displayExternalWork, checkOrder)
def	findStalledModules (processingSteps, numStreams)
def	getTime (line)
def	mergeContiguousBlocks (blocks)
def	parseTracerOutput (f)
def	plotPerStreamAboveFirstAndPrepareStack (points, allStackTimes, ax, stream, height, streamHeightCut, doPlot, addToStackTimes, color, threadOffset)
def	printHelp ()
def	printStalledModulesInOrder (stalledModules)
def	processingStepsFromStallMonitorOutput (f, moduleNames)
def	readLogFile (inputFile)
def	reduceSortedPoints (ps)

Function Documentation

def edmStreamStallGrapher.adjacentDiff

(

pairLists

)

Definition at line 445 of file edmStreamStallGrapher.py.

Referenced by createPDFImage().

def adjacentDiff(*pairLists):
    points = []
    for pairList in pairLists:
        points += [Point(x[0], 1) for x in pairList if x[1] != 0]
        points += [Point(sum(x),-1) for x in pairList if x[1] != 0]
    points.sort(key=attrgetter('x'))
    return points

def edmStreamStallGrapher.chooseParser

(

inputFile

)

Definition at line 278 of file edmStreamStallGrapher.py.

References cmsRelvalreport.exit, and edm.print().

Referenced by readLogFile().

def chooseParser(inputFile):

    firstLine = inputFile.readline().rstrip()
    for i in range(3):
        inputFile.readline()
    # Often the Tracer log file starts with 4 lines not from the Tracer
    fifthLine = inputFile.readline().rstrip()
    inputFile.seek(0) # Rewind back to beginning
    if (firstLine.find("# Transition") != -1) or (firstLine.find("# Step") != -1):
        print("> ... Parsing StallMonitor output.")
        return StallMonitorParser

    if firstLine.find("++") != -1 or fifthLine.find("++") != -1:
        global kTracerInput
        kTracerInput = True
        print("> ... Parsing Tracer output.")
        return TracerParser
    else:
        inputFile.close()
        print("Unknown input format.")
        exit(1)

#----------------------------------------------

def edmStreamStallGrapher.consolidateContiguousBlocks	(		numStreams,
			streamInfo
	)

Definition at line 483 of file edmStreamStallGrapher.py.

References mps_setup.append, and timeUnitHelper.unpack().

Referenced by createPDFImage().

def consolidateContiguousBlocks(numStreams, streamInfo):
    oldStreamInfo = streamInfo
    streamInfo = [[] for x in xrange(numStreams)]

    for s in xrange(numStreams):
        if oldStreamInfo[s]:
            lastStartTime,lastTimeLength,lastColor = oldStreamInfo[s][0].unpack()
            for info in oldStreamInfo[s][1:]:
                start,length,color = info.unpack()
                if color == lastColor and lastStartTime+lastTimeLength == start:
                    lastTimeLength += length
                else:
                    streamInfo[s].append(StreamInfoElement(lastStartTime,lastTimeLength,lastColor))
                    lastStartTime = start
                    lastTimeLength = length
                    lastColor = color
            streamInfo[s].append(StreamInfoElement(lastStartTime,lastTimeLength,lastColor))

    return streamInfo

#----------------------------------------------
# Consolidating contiguous blocks with the same color drastically
# reduces the size of the pdf file.  Same functionality as the
# previous function, but with slightly different implementation.

def edmStreamStallGrapher.createAsciiImage	(		processingSteps,
			numStreams,
			maxNameSize
	)

Definition at line 349 of file edmStreamStallGrapher.py.

References edm.print(), and str.

Referenced by createPDFImage().

def createAsciiImage(processingSteps, numStreams, maxNameSize):
    streamTime = [0]*numStreams
    streamState = [0]*numStreams
    modulesActiveOnStreams = [{} for x in xrange(numStreams)]
    for n,trans,s,time,isEvent in processingSteps:
        waitTime = None
        modulesActiveOnStream = modulesActiveOnStreams[s]
        if trans == kPrefetchEnd:
            modulesActiveOnStream[n] = time
            continue
        elif trans == kStartedAcquire or trans == kStarted:
            if n in modulesActiveOnStream:
                waitTime = time - modulesActiveOnStream[n]
                modulesActiveOnStream.pop(n, None)
            streamState[s] +=1
        elif trans == kFinishedAcquire or trans == kFinished:
            streamState[s] -=1
            streamTime[s] = time
        elif trans == kStartedSourceDelayedRead:
            if streamState[s] == 0:
                waitTime = time - streamTime[s]
        elif trans == kStartedSource:
            modulesActiveOnStream.clear()
        elif trans == kFinishedSource or trans == kFinishedSourceDelayedRead:
            streamTime[s] = time
        states = "%-*s: " % (maxNameSize,n)
        if trans == kStartedAcquire or trans == kStarted or trans == kStartedSourceDelayedRead or trans == kStartedSource:
            states +="+ "
        else:
            states +="- "
        for index, state in enumerate(streamState):
            if n==kSourceFindEvent and index == s:
                states +="* "
            else:
                states +=str(state)+" "
        states += " -- " + str(time/1000.) + " " + str(s) + " "
        if waitTime is not None:
            states += " %.2f"% (waitTime/1000.)
            if waitTime > kStallThreshold:
                states += " STALLED"

        print(states)

#----------------------------------------------

def edmStreamStallGrapher.createPDFImage	(		pdfFile,
			shownStacks,
			processingSteps,
			numStreams,
			stalledModuleInfo,
			displayExternalWork,
			checkOrder
	)

Definition at line 557 of file edmStreamStallGrapher.py.

References PVValHelper.add(), adjacentDiff(), mps_setup.append, consolidateContiguousBlocks(), createAsciiImage(), cmsRelvalreport.exit, findStalledModules(), list(), SiStripPI.max, mergeContiguousBlocks(), plotPerStreamAboveFirstAndPrepareStack(), edm.print(), printHelp(), printStalledModulesInOrder(), readLogFile(), reduceSortedPoints(), MatrixUtil.remove(), and ComparisonHelper.zip().

def createPDFImage(pdfFile, shownStacks, processingSteps, numStreams, stalledModuleInfo, displayExternalWork, checkOrder):

    stalledModuleNames = set([x for x in stalledModuleInfo.iterkeys()])
    streamLowestRow = [[] for x in xrange(numStreams)]
    modulesActiveOnStreams = [set() for x in xrange(numStreams)]
    acquireActiveOnStreams = [set() for x in xrange(numStreams)]
    externalWorkOnStreams  = [set() for x in xrange(numStreams)]
    previousFinishTime = [None for x in xrange(numStreams)]
    streamRunningTimes = [[] for x in xrange(numStreams)]
    streamExternalWorkRunningTimes = [[] for x in xrange(numStreams)]
    maxNumberOfConcurrentModulesOnAStream = 1
    externalWorkModulesInJob = False
    previousTime = [0 for x in xrange(numStreams)]

    # The next five variables are only used to check for out of order transitions
    finishBeforeStart = [set() for x in xrange(numStreams)]
    finishAcquireBeforeStart = [set() for x in xrange(numStreams)]
    countSource = [0 for x in xrange(numStreams)]
    countDelayedSource = [0 for x in xrange(numStreams)]
    countExternalWork = [defaultdict(int) for x in xrange(numStreams)]

    timeOffset = None
    for n,trans,s,time,isEvent in processingSteps:
        if timeOffset is None:
            timeOffset = time
        startTime = None
        time -=timeOffset
        # force the time to monotonically increase on each stream
        if time < previousTime[s]:
            time = previousTime[s]
        previousTime[s] = time

        activeModules = modulesActiveOnStreams[s]
        acquireModules = acquireActiveOnStreams[s]
        externalWorkModules = externalWorkOnStreams[s]

        if trans == kStarted or trans == kStartedSourceDelayedRead or trans == kStartedAcquire or trans == kStartedSource :
            if checkOrder:
                # Note that the code which checks the order of transitions assumes that
                # all the transitions exist in the input. It is checking only for order
                # problems, usually a start before a finish. Problems are fixed and
                # silently ignored. Nothing gets plotted for transitions that are
                # in the wrong order.
                if trans == kStarted:
                    countExternalWork[s][n] -= 1
                    if n in finishBeforeStart[s]:
                        finishBeforeStart[s].remove(n)
                        continue
                elif trans == kStartedAcquire:
                    if n in finishAcquireBeforeStart[s]:
                        finishAcquireBeforeStart[s].remove(n)
                        continue

            if trans == kStartedSourceDelayedRead:
                countDelayedSource[s] += 1
                if countDelayedSource[s] < 1:
                    continue
            elif trans == kStartedSource:
                countSource[s] += 1
                if countSource[s] < 1:
                    continue

            moduleNames = activeModules.copy()
            moduleNames.update(acquireModules)
            if trans == kStartedAcquire:
                 acquireModules.add(n)
            else:
                 activeModules.add(n)
            streamRunningTimes[s].append(Point(time,1))
            if moduleNames or externalWorkModules:
                startTime = previousFinishTime[s]
            previousFinishTime[s] = time

            if trans == kStarted and n in externalWorkModules:
                externalWorkModules.remove(n)
                streamExternalWorkRunningTimes[s].append(Point(time, -1))
            else:
                nTotalModules = len(activeModules) + len(acquireModules) + len(externalWorkModules)
                maxNumberOfConcurrentModulesOnAStream = max(maxNumberOfConcurrentModulesOnAStream, nTotalModules)
        elif trans == kFinished or trans == kFinishedSourceDelayedRead or trans == kFinishedAcquire or trans == kFinishedSource :
            if checkOrder:
                if trans == kFinished:
                    if n not in activeModules:
                        finishBeforeStart[s].add(n)
                        continue

            if trans == kFinishedSourceDelayedRead:
                countDelayedSource[s] -= 1
                if countDelayedSource[s] < 0:
                    continue
            elif trans == kFinishedSource:
                countSource[s] -= 1
                if countSource[s] < 0:
                    continue

            if trans == kFinishedAcquire:
                if checkOrder:
                    countExternalWork[s][n] += 1
                if displayExternalWork:
                    externalWorkModulesInJob = True
                    if (not checkOrder) or countExternalWork[s][n] > 0:
                        externalWorkModules.add(n)
                        streamExternalWorkRunningTimes[s].append(Point(time,+1))
                if checkOrder and n not in acquireModules:
                    finishAcquireBeforeStart[s].add(n)
                    continue
            streamRunningTimes[s].append(Point(time,-1))
            startTime = previousFinishTime[s]
            previousFinishTime[s] = time
            moduleNames = activeModules.copy()
            moduleNames.update(acquireModules)

            if trans == kFinishedAcquire:
                acquireModules.remove(n)
            elif trans == kFinishedSourceDelayedRead:
                if countDelayedSource[s] == 0:
                    activeModules.remove(n)
            elif trans == kFinishedSource:
                if countSource[s] == 0:
                    activeModules.remove(n)
            else:
                activeModules.remove(n)

        if startTime is not None:
            c="green"
            if not isEvent:
              c="limegreen"
            if not moduleNames:
                c = "darkviolet"
            elif (kSourceDelayedRead in moduleNames) or (kSourceFindEvent in moduleNames):
                c = "orange"
            else:
                for n in moduleNames:
                    if n in stalledModuleNames:
                        c="red"
                        break
            streamLowestRow[s].append(StreamInfoElement(startTime, time-startTime, c))
    streamLowestRow = consolidateContiguousBlocks(numStreams, streamLowestRow)

    nr = 1
    if shownStacks:
        nr += 1
    fig, ax = plt.subplots(nrows=nr, squeeze=True)
    axStack = None
    if shownStacks:
        [xH,yH] = fig.get_size_inches()
        fig.set_size_inches(xH,yH*4/3)
        ax = plt.subplot2grid((4,1),(0,0), rowspan=3)
        axStack = plt.subplot2grid((4,1),(3,0))

    ax.set_xlabel("Time (sec)")
    ax.set_ylabel("Stream ID")
    ax.set_ylim(-0.5,numStreams-0.5)
    ax.yaxis.set_ticks(xrange(numStreams))

    height = 0.8/maxNumberOfConcurrentModulesOnAStream
    allStackTimes={'green': [],'limegreen':[], 'red': [], 'blue': [], 'orange': [], 'darkviolet': []}
    for iStream,lowestRow in enumerate(streamLowestRow):
        times=[(x.begin/1000., x.delta/1000.) for x in lowestRow] # Scale from msec to sec.
        colors=[x.color for x in lowestRow]
        # for each stream, plot the lowest row
        ax.broken_barh(times,(iStream-0.4,height),facecolors=colors,edgecolors=colors,linewidth=0)
        # record them also for inclusion in the stack plot
        # the darkviolet ones get counted later so do not count them here
        for info in lowestRow:
            if not info.color == 'darkviolet':
                allStackTimes[info.color].append((info.begin, info.delta))

    # Now superimpose the number of concurrently running modules on to the graph.
    if maxNumberOfConcurrentModulesOnAStream > 1 or externalWorkModulesInJob:

        for i,perStreamRunningTimes in enumerate(streamRunningTimes):

            perStreamTimesWithExtendedWork = list(perStreamRunningTimes)
            perStreamTimesWithExtendedWork.extend(streamExternalWorkRunningTimes[i])

            plotPerStreamAboveFirstAndPrepareStack(perStreamTimesWithExtendedWork,
                                                   allStackTimes, ax, i, height,
                                                   streamHeightCut=2,
                                                   doPlot=True,
                                                   addToStackTimes=False,
                                                   color='darkviolet',
                                                   threadOffset=1)

            plotPerStreamAboveFirstAndPrepareStack(perStreamRunningTimes,
                                                   allStackTimes, ax, i, height,
                                                   streamHeightCut=2,
                                                   doPlot=True,
                                                   addToStackTimes=True,
                                                   color='blue',
                                                   threadOffset=1)

            plotPerStreamAboveFirstAndPrepareStack(streamExternalWorkRunningTimes[i],
                                                   allStackTimes, ax, i, height,
                                                   streamHeightCut=1,
                                                   doPlot=False,
                                                   addToStackTimes=True,
                                                   color='darkviolet',
                                                   threadOffset=0)

    if shownStacks:
        print("> ... Generating stack")
        stack = Stack()
        for color in ['green','limegreen','blue','red','orange','darkviolet']:
            tmp = allStackTimes[color]
            tmp = reduceSortedPoints(adjacentDiff(tmp))
            stack.update(color, tmp)

        for stk in reversed(stack.data):
            color = stk[0]

            # Now arrange list in a manner that it can be grouped by the height of the block
            height = 0
            xs = []
            for p1,p2 in zip(stk[1], stk[1][1:]):
                height += p1.y
                xs.append((p1.x, p2.x-p1.x, height))
            xs.sort(key = itemgetter(2))
            xs = mergeContiguousBlocks(xs)

            for height, xpairs in groupby(xs, itemgetter(2)):
                finalxs = [(e[0]/1000.,e[1]/1000.) for e in xpairs]
                # plot the stacked plot, one color and one height on each call to broken_barh
                axStack.broken_barh(finalxs, (0, height), facecolors=color, edgecolors=color, linewidth=0)

        axStack.set_xlabel("Time (sec)");
        axStack.set_ylabel("# modules");
        axStack.set_xlim(ax.get_xlim())
        axStack.tick_params(top='off')

    fig.text(0.1, 0.95, "modules running event", color = "green", horizontalalignment = 'left')
    fig.text(0.1, 0.92, "modules running other", color = "limegreen", horizontalalignment = 'left')
    fig.text(0.5, 0.95, "stalled module running", color = "red", horizontalalignment = 'center')
    fig.text(0.9, 0.95, "read from input", color = "orange", horizontalalignment = 'right')
    fig.text(0.5, 0.92, "multiple modules running", color = "blue", horizontalalignment = 'center')
    if displayExternalWork:
        fig.text(0.9, 0.92, "external work", color = "darkviolet", horizontalalignment = 'right')
    print("> ... Saving to file: '{}'".format(pdfFile))
    plt.savefig(pdfFile)

#=======================================

def edmStreamStallGrapher.findStalledModules	(		processingSteps,
			numStreams
	)

Definition at line 316 of file edmStreamStallGrapher.py.

Referenced by createPDFImage().

def findStalledModules(processingSteps, numStreams):
    streamTime = [0]*numStreams
    streamState = [0]*numStreams
    stalledModules = {}
    modulesActiveOnStream = [{} for x in xrange(numStreams)]
    for n,trans,s,time,isEvent in processingSteps:

        waitTime = None
        modulesOnStream = modulesActiveOnStream[s]
        if trans == kPrefetchEnd:
            modulesOnStream[n] = time
        elif trans == kStarted or trans == kStartedAcquire:
            if n in modulesOnStream:
                waitTime = time - modulesOnStream[n]
                modulesOnStream.pop(n, None)
            streamState[s] +=1
        elif trans == kFinished or trans == kFinishedAcquire:
            streamState[s] -=1
            streamTime[s] = time
        elif trans == kStartedSourceDelayedRead:
            if streamState[s] == 0:
                waitTime = time - streamTime[s]
        elif trans == kStartedSource:
            modulesOnStream.clear()
        elif trans == kFinishedSource or trans == kFinishedSourceDelayedRead:
            streamTime[s] = time
        if waitTime is not None:
            if waitTime > kStallThreshold:
                t = stalledModules.setdefault(n,[])
                t.append(waitTime)
    return stalledModules

#----------------------------------------------

def edmStreamStallGrapher.getTime

(

line

)

Definition at line 179 of file edmStreamStallGrapher.py.

References objects.autophobj.float, and createfilelist.int.

Referenced by EcalPerEvtLaserAnalyzer.endJob(), EcalLaserAnalyzer2.endJob(), EcalLaserAnalyzer.endJob(), parseTracerOutput(), and pos::PixelFEDTestDAC.writeXMLHeader().

def getTime(line):
    time = line.split(" ")[1]
    time = time.split(":")
    time = int(time[0])*60*60+int(time[1])*60+float(time[2])
    time = int(1000*time) # convert to milliseconds
    return time

#----------------------------------------------
# The next function parses the Tracer output.
# Here are some differences to consider if you use Tracer output
# instead of the StallMonitor output.
# - The time in the text of the Tracer output is not as precise
# as the StallMonitor (.01 s vs .001 s)
# - The MessageLogger bases the time on when the message printed
# and not when it was initially queued up to print which smears
# the accuracy of the times.
# - Both of the previous things can produce some strange effects
# in the output plots.
# - The file size of the Tracer text file is much larger.
# - The CPU work needed to parse the Tracer files is larger.
# - The Tracer log file is expected to have "++" in the first
# or fifth line. If there are extraneous lines at the beginning
# you have to remove them.
# - The ascii printout out will have one extraneous line
# near the end for the SourceFindEvent start.
# - The only advantage I can see is that you have only
# one output file to handle instead of two, the regular
# log file and the StallMonitor output.
# We might should just delete the Tracer option because it is
# clearly inferior ...

def edmStreamStallGrapher.mergeContiguousBlocks

(

blocks

)

Definition at line 507 of file edmStreamStallGrapher.py.

Referenced by createPDFImage(), and plotPerStreamAboveFirstAndPrepareStack().

def mergeContiguousBlocks(blocks):
    oldBlocks = blocks

    blocks = []
    if not oldBlocks:
        return blocks

    lastStartTime,lastTimeLength,lastHeight = oldBlocks[0]
    for start,length,height in oldBlocks[1:]:
        if height == lastHeight and lastStartTime+lastTimeLength == start:
            lastTimeLength += length
        else:
            blocks.append((lastStartTime,lastTimeLength,lastHeight))
            lastStartTime = start
            lastTimeLength = length
            lastHeight = height
    blocks.append((lastStartTime,lastTimeLength,lastHeight))

    return blocks

#----------------------------------------------

def edmStreamStallGrapher.parseTracerOutput

(

f

)

Definition at line 209 of file edmStreamStallGrapher.py.

References getTime(), createfilelist.int, and SiStripPI.max.

def parseTracerOutput(f):
    processingSteps = []
    numStreams = 0
    maxNameSize = 0
    startTime = 0
    streamsThatSawFirstEvent = set()
    for l in f:
        trans = None
        # We estimate the start and stop of the source
        # by the end of the previous event and start of
        # the event. This is historical, probably because
        # the Tracer output for the begin and end of the
        # source event does not include the stream number.
        if l.find("processing event :") != -1:
            name = kSourceFindEvent
            trans = kStartedSource
            # the end of the source is estimated using the start of the event
            if l.find("starting:") != -1:
                trans = kFinishedSource
        elif l.find("processing event for module") != -1:
            trans = kStarted
            if l.find("finished:") != -1:
                if l.find("prefetching") != -1:
                    trans = kPrefetchEnd
                else:
                    trans = kFinished
            else:
                if l.find("prefetching") != -1:
                    #skip this since we don't care about prefetch starts
                    continue
            name = l.split("'")[1]
        elif l.find("processing event acquire for module:") != -1:
            trans = kStartedAcquire
            if l.find("finished:") != -1:
                trans = kFinishedAcquire
            name = l.split("'")[1]
        elif l.find("event delayed read from source") != -1:
            trans = kStartedSourceDelayedRead
            if l.find("finished:") != -1:
                trans = kFinishedSourceDelayedRead
            name = kSourceDelayedRead
        if trans is not None:
            time = getTime(l)
            if startTime == 0:
                startTime = time
            time = time - startTime
            streamIndex = l.find("stream = ")
            stream = int(l[streamIndex+9:l.find(" ",streamIndex+10)])
            maxNameSize = max(maxNameSize, len(name))

            if trans == kFinishedSource and not stream in streamsThatSawFirstEvent:
                # This is wrong but there is no way to estimate the time better
                # because there is no previous event for the first event.
                processingSteps.append((name,kStartedSource,stream,time,True))
                streamsThatSawFirstEvent.add(stream)

            processingSteps.append((name,trans,stream,time, True))
            numStreams = max(numStreams, stream+1)

    f.close()
    return (processingSteps,numStreams,maxNameSize)

def edmStreamStallGrapher.plotPerStreamAboveFirstAndPrepareStack	(		points,
			allStackTimes,
			ax,
			stream,
			height,
			streamHeightCut,
			doPlot,
			addToStackTimes,
			color,
			threadOffset
	)

Definition at line 528 of file edmStreamStallGrapher.py.

References mergeContiguousBlocks(), reduceSortedPoints(), and ComparisonHelper.zip().

Referenced by createPDFImage().

def plotPerStreamAboveFirstAndPrepareStack(points, allStackTimes, ax, stream, height, streamHeightCut, doPlot, addToStackTimes, color, threadOffset):
    points = sorted(points, key=attrgetter('x'))
    points = reduceSortedPoints(points)
    streamHeight = 0
    preparedTimes = []
    for t1,t2 in zip(points, points[1:]):
        streamHeight += t1.y
        # We make a cut here when plotting because the first row for
        # each stream was already plotted previously and we do not
        # need to plot it again. And also we want to count things
        # properly in allStackTimes. We want to avoid double counting
        # or missing running modules and this is complicated because
        # we counted the modules in the first row already.
        if streamHeight < streamHeightCut:
            continue
        preparedTimes.append((t1.x,t2.x-t1.x, streamHeight))
    preparedTimes.sort(key=itemgetter(2))
    preparedTimes = mergeContiguousBlocks(preparedTimes)

    for nthreads, ts in groupby(preparedTimes, itemgetter(2)):
        theTS = [(t[0],t[1]) for t in ts]
        if doPlot:
            theTimes = [(t[0]/1000.,t[1]/1000.) for t in theTS]
            yspan = (stream-0.4+height,height*(nthreads-1))
            ax.broken_barh(theTimes, yspan, facecolors=color, edgecolors=color, linewidth=0)
        if addToStackTimes:
            allStackTimes[color].extend(theTS*(nthreads-threadOffset))

#----------------------------------------------

def edmStreamStallGrapher.printHelp

(

)

Definition at line 10 of file edmStreamStallGrapher.py.

Referenced by createPDFImage().

def printHelp():
    s = '''
To Use: Add the StallMonitor Service to the cmsRun job you want to check for
  stream stalls. Use something like this in the configuration:

  process.add_(cms.Service("StallMonitor", fileName = cms.untracked.string("stallMonitor.log")))

  After running the job, execute this script and pass the name of the
  StallMonitor log file to the script.

  By default, the script will then print an 'ASCII art' stall graph
  which consists of a line of text for each time a module or the
  source stops or starts. Each line contains the name of the module
  which either started or stopped running, and the number of modules
  running on each stream at that moment in time. After that will be
  the time and stream number. Then if a module just started, you
  will also see the amount of time the module spent between finishing
  its prefetching and starting.  The state of a module is represented
  by a symbol:

    plus  ("+") the stream has just finished waiting and is starting a module
    minus ("-") the stream just finished running a module

  If a module had to wait more than 0.1 seconds, the end of the line
  will have "STALLED". Startup actions, e.g. reading conditions,
  may affect results for the first few events.

  Using the command line arguments described above you can make the
  program create a PDF file with actual graphs instead of the 'ASCII art'
  output.

  Once the graph is completed, the program outputs the list of modules
  which had the greatest total stall times. The list is sorted by
  total stall time and written in descending order. In addition, the
  list of all stall times for the module is given.

  There is an inferior alternative (an old obsolete way).
  Instead of using the StallMonitor Service, you can use the
  Tracer Service.  Make sure to use the 'printTimestamps' option
  cms.Service("Tracer", printTimestamps = cms.untracked.bool(True))
  There are problems associated with this and it is not recommended.'''
    return s

kStallThreshold=100 #in milliseconds
kTracerInput=False

#Stream states
kStarted=0
kFinished=1
kPrefetchEnd=2
kStartedAcquire=3
kFinishedAcquire=4
kStartedSource=5
kFinishedSource=6
kStartedSourceDelayedRead=7
kFinishedSourceDelayedRead=8

#Special names

def edmStreamStallGrapher.printStalledModulesInOrder

(

stalledModules

)

Definition at line 393 of file edmStreamStallGrapher.py.

References join(), SiStripPI.max, and edm.print().

Referenced by createPDFImage().

def printStalledModulesInOrder(stalledModules):
    priorities = []
    maxNameSize = 0
    for name,t in six.iteritems(stalledModules):
        maxNameSize = max(maxNameSize, len(name))
        t.sort(reverse=True)
        priorities.append((name,sum(t),t))

    def sumSort(i,j):
        return cmp(i[1],j[1])
    priorities.sort(cmp=sumSort, reverse=True)

    nameColumn = "Stalled Module"
    maxNameSize = max(maxNameSize, len(nameColumn))

    stallColumn = "Tot Stall Time"
    stallColumnLength = len(stallColumn)

    print("%-*s" % (maxNameSize, nameColumn), "%-*s"%(stallColumnLength,stallColumn), " Stall Times")
    for n,s,t in priorities:
        paddedName = "%-*s:" % (maxNameSize,n)
        print(paddedName, "%-*.2f"%(stallColumnLength,s/1000.), ", ".join([ "%.2f"%(x/1000.) for x in t]))

#--------------------------------------------------------

def edmStreamStallGrapher.processingStepsFromStallMonitorOutput	(		f,
			moduleNames
	)

Definition at line 72 of file edmStreamStallGrapher.py.

References createfilelist.int.

Referenced by edmStreamStallGrapher.StallMonitorParser.processingSteps().

def processingStepsFromStallMonitorOutput(f,moduleNames):
    for rawl in f:
        l = rawl.strip()
        if not l or l[0] == '#':
            continue
        (step,payload) = tuple(l.split(None,1))
        payload=payload.split()

        # Ignore these
        if step == 'E' or step == 'e':
            continue

        # Payload format is:
        #  <stream id> <..other fields..> <time since begin job>
        stream = int(payload[0])
        time = int(payload[-1])
        trans = None
        isEvent = True

        name = None
        # 'S' = begin of event creation in source
        # 's' = end of event creation in source
        if step == 'S' or step == 's':
            name = kSourceFindEvent
            trans = kStartedSource
            # The start of an event is the end of the framework part
            if step == 's':
                trans = kFinishedSource
        else:
            # moduleID is the second payload argument for all steps below
            moduleID = payload[1]

            # 'p' = end of module prefetching
            # 'M' = begin of module processing
            # 'm' = end of module processing
            if step == 'p' or step == 'M' or step == 'm':
                trans = kStarted
                if step == 'p':
                    trans = kPrefetchEnd
                elif step == 'm':
                    trans = kFinished
                if step == 'm' or step == 'M':
                    isEvent = (int(payload[2]) == 0)
                name = moduleNames[moduleID]

            # 'A' = begin of module acquire function
            # 'a' = end of module acquire function
            elif step == 'A' or step == 'a':
                trans = kStartedAcquire
                if step == 'a':
                    trans = kFinishedAcquire
                name = moduleNames[moduleID]

            # Delayed read from source
            # 'R' = begin of delayed read from source
            # 'r' = end of delayed read from source
            elif step == 'R' or step == 'r':
                trans = kStartedSourceDelayedRead
                if step == 'r':
                    trans = kFinishedSourceDelayedRead
                name = kSourceDelayedRead

        if trans is not None:
            yield (name,trans,stream,time, isEvent)
    
    return

def edmStreamStallGrapher.readLogFile

(

inputFile

)

Definition at line 301 of file edmStreamStallGrapher.py.

References chooseParser().

Referenced by createPDFImage().

def readLogFile(inputFile):
    parseInput = chooseParser(inputFile)
    return parseInput(inputFile)

#----------------------------------------------
#
# modules: The time between prefetch finished and 'start processing' is
#   the time it took to acquire any resources which is by definition the
#   stall time.
#
# source: The source just records how long it spent doing work,
#   not how long it was stalled. We can get a lower bound on the stall
#   time for delayed reads by measuring the time the stream was doing
#   no work up till the start of the source delayed read.
#

def edmStreamStallGrapher.reduceSortedPoints

(

ps

)

Definition at line 429 of file edmStreamStallGrapher.py.

Referenced by createPDFImage(), plotPerStreamAboveFirstAndPrepareStack(), and edmStreamStallGrapher.Stack.update().

def reduceSortedPoints(ps):
    if len(ps) < 2:
        return ps
    reducedPoints = []
    tmp = Point(ps[0].x, ps[0].y)
    for p in ps[1:]:
        if tmp.x == p.x:
            tmp.y += p.y
        else:
            reducedPoints.append(tmp)
            tmp = Point(p.x, p.y)
    reducedPoints.append(tmp)
    reducedPoints = [p for p in reducedPoints if p.y != 0]
    return reducedPoints

# -------------------------------------------