edmStreamStallGrapher Namespace Reference

Classes
class	Point
class	Stack
class	StreamInfoElement

Functions
def	adjacentDiff (pairLists)
def	chooseParser (inputFile)
def	consolidateContiguousBlocks (numStreams, streamInfo)
def	createAsciiImage (processingSteps, numStreams, maxNameSize)
def	createPDFImage (pdfFile, shownStacks, processingSteps, numStreams, stalledModuleInfo)
def	findStalledModules (processingSteps, numStreams)
def	getTime (line)
def	mergeContiguousBlocks (blocks)
def	parseStallMonitorOutput (f)
def	parseTracerOutput (f)
def	printHelp ()
def	printStalledModulesInOrder (stalledModules)
def	readLogFile (inputFile)
def	reduceSortedPoints (ps)

Function Documentation

def edmStreamStallGrapher.adjacentDiff

(

pairLists

)

Definition at line 341 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 187 of file edmStreamStallGrapher.py.

References cmsRelvalreport.exit.

Referenced by readLogFile().

def chooseParser(inputFile):
    firstLine = inputFile.readline().rstrip()
    inputFile.seek(0) # Rewind back to beginning

    if firstLine.find("# Step") != -1:
        print "> ... Parsing StallMonitor output."
        return parseStallMonitorOutput
    elif firstLine.find("++") != -1:
        global kTracerInput
        kTracerInput = True
        print "> ... Parsing Tracer output."
        return parseTracerOutput
    else:
        inputFile.close()
        print "Unknown input format."
        exit(1)

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

def edmStreamStallGrapher.consolidateContiguousBlocks	(		numStreams,
			streamInfo
	)

Definition at line 379 of file edmStreamStallGrapher.py.

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

Referenced by createPDFImage().

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

    for s in xrange(numStreams):
        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 246 of file edmStreamStallGrapher.py.

References harvestTrackValidationPlots.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 in processingSteps:
        modulesActiveOnStream = modulesActiveOnStreams[s]
        waitTime = None
        if trans == kPrefetchEnd:
            modulesActiveOnStream[n] = time
            continue
        if trans == kStarted:
            if n != kSourceFindEvent:
                streamState[s] +=1
            if n in modulesActiveOnStream:
                waitTime = time - modulesActiveOnStream[n]
            if n == kSourceDelayedRead:
                if streamState[s] == 0:
                    waitTime = time-streamTime[s]
        if trans == kFinished:
            if n != kSourceDelayedRead and n!=kSourceFindEvent:
                modulesActiveOnStream.pop(n, None)
            if n != kSourceFindEvent:
                streamState[s] -=1
            streamTime[s] = time
        states = "%-*s: " % (maxNameSize,n)
        if trans == kStarted:
            states +="+ "
        if trans == kFinished:
            states +="- "
        for index, state in enumerate(streamState):
            if n==kSourceFindEvent and index == s:
                states +="* "
            else:
                states +=str(state)+" "
        if waitTime is not None:
            states += " %.2f"% (waitTime/1000.)
            if waitTime > kStallThreshold:
                states += " STALLED "+str(time/1000.)+" "+str(s)

        print states
    return stalledModules

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

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

Definition at line 420 of file edmStreamStallGrapher.py.

References PVValHelper.add(), adjacentDiff(), mps_alisetup.append, consolidateContiguousBlocks(), createAsciiImage(), cmsRelvalreport.exit, findStalledModules(), hpstanc_transforms.max, mergeContiguousBlocks(), min(), printHelp(), printStalledModulesInOrder(), readLogFile(), reduceSortedPoints(), MatrixUtil.remove(), and ComparisonHelper.zip().

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

    stalledModuleNames = set([x for x in stalledModuleInfo.iterkeys()])
    streamInfo = [[] for x in xrange(numStreams)]
    modulesActiveOnStreams = [{} for x in xrange(numStreams)]
    streamLastEventEndTimes = [None]*numStreams
    streamRunningTimes = [[] for x in xrange(numStreams)]
    maxNumberOfConcurrentModulesOnAStream = 1
    streamInvertedMessageFromModule = [set() for x in xrange(numStreams)]

    for n,trans,s,time in processingSteps:
        startTime = None
        if streamLastEventEndTimes[s] is None:
            streamLastEventEndTimes[s]=time
        if trans == kStarted:
            if n == kSourceFindEvent:
                # We assume the time from the end of the last event
                # for a stream until the start of a new event for that
                # stream is taken up by the source.
                startTime = streamLastEventEndTimes[s]
                moduleNames = set(n)
            else:
                activeModules = modulesActiveOnStreams[s]
                moduleNames = set(activeModules.iterkeys())
                if n in streamInvertedMessageFromModule[s] and kTracerInput:
                    # This is the rare case where a finished message
                    # is issued before the corresponding started.
                    streamInvertedMessageFromModule[s].remove(n)
                    continue
                activeModules[n] = time
                nModulesRunning = len(activeModules)
                streamRunningTimes[s].append(Point(time,1))
                maxNumberOfConcurrentModulesOnAStream = max(maxNumberOfConcurrentModulesOnAStream, nModulesRunning)
                if nModulesRunning > 1:
                    # Need to create a new time span to avoid overlaps in graph.
                    startTime = min(activeModules.itervalues())
                    for k in activeModules.iterkeys():
                        activeModules[k]=time

        if trans == kFinished:
            if n == kSourceFindEvent:
                streamLastEventEndTimes[s]=time
            else:
                activeModules = modulesActiveOnStreams[s]
                if n not in activeModules and kTracerInput:
                    # This is the rare case where a finished message
                    # is issued before the corresponding started.
                    streamInvertedMessageFromModule[s].add(n)
                    continue
                streamRunningTimes[s].append(Point(time,-1))
                startTime = activeModules[n]
                moduleNames = set(activeModules.iterkeys())
                del activeModules[n]
                nModulesRunning = len(activeModules)
                if nModulesRunning > 0:
                    # Reset start time for remaining modules to this time
                    # to avoid overlapping time ranges when making the plot.
                    for k in activeModules.iterkeys():
                        activeModules[k] = time
        if startTime is not None:
            c="green"
            if (kSourceDelayedRead in moduleNames) or (kSourceFindEvent in moduleNames):
                c = "orange"
            else:
                for n in moduleNames:
                    if n in stalledModuleNames:
                        c="red"
                        break
            streamInfo[s].append(StreamInfoElement(startTime, time-startTime, c))

    streamInfo = consolidateContiguousBlocks(numStreams, streamInfo)

    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': [], 'red': [], 'blue': [], 'orange': []}
    for i,perStreamInfo in enumerate(streamInfo):
        times=[(x.begin/1000., x.delta/1000.) for x in perStreamInfo] # Scale from msec to sec.
        colors=[x.color for x in perStreamInfo]
        ax.broken_barh(times,(i-0.4,height),facecolors=colors,edgecolors=colors,linewidth=0)
        for info in perStreamInfo:
            allStackTimes[info.color].append((info.begin, info.delta))

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

        for i,perStreamRunningTimes in enumerate(streamRunningTimes):
            perStreamTimes = sorted(perStreamRunningTimes, key=attrgetter('x'))
            perStreamTimes = reduceSortedPoints(perStreamTimes)
            streamHeight = 0
            preparedTimes = []
            for t1,t2 in zip(perStreamTimes, perStreamTimes[1:]):
                streamHeight += t1.y
                if streamHeight < 2:
                    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]
                theTimes = [(t[0]/1000.,t[1]/1000.) for t in theTS]
                yspan = (i-0.4+height,height*(nthreads-1))
                ax.broken_barh(theTimes, yspan, facecolors='blue', edgecolors='blue', linewidth=0)
                allStackTimes['blue'].extend(theTS*(nthreads-1))

    if shownStacks:
        print "> ... Generating stack"
        stack = Stack()
        for color in ['green','blue','red','orange']:
            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]
                axStack.broken_barh(finalxs, (0, height), facecolors=color, edgecolors=color, linewidth=0)

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

    fig.text(0.1, 0.95, "modules running", color = "green", 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')
    print "> ... Saving to file: '{}'".format(pdfFile)
    plt.savefig(pdfFile)

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

def edmStreamStallGrapher.findStalledModules	(		processingSteps,
			numStreams
	)

Definition at line 219 of file edmStreamStallGrapher.py.

Referenced by createPDFImage().

def findStalledModules(processingSteps, numStreams):
    streamTime = [0]*numStreams
    stalledModules = {}
    modulesActiveOnStream = [{} for x in xrange(numStreams)]
    for n,trans,s,time in processingSteps:
        waitTime = None
        modulesOnStream = modulesActiveOnStream[s]
        if trans == kPrefetchEnd:
            modulesOnStream[n] = time
        if trans == kStarted:
            if n in modulesOnStream:
                waitTime = time - modulesOnStream[n]
            if n == kSourceDelayedRead:
                if 0 == len(modulesOnStream):
                    waitTime = time - streamTime[s]
        if trans == kFinished:
            if n != kSourceDelayedRead and n!=kSourceFindEvent:
                modulesOnStream.pop(n, None)
            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 115 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

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

def edmStreamStallGrapher.mergeContiguousBlocks

(

blocks

)

Definition at line 402 of file edmStreamStallGrapher.py.

Referenced by createPDFImage().

def mergeContiguousBlocks(blocks):
    oldBlocks = blocks

    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.parseStallMonitorOutput

(

f

)

Definition at line 52 of file edmStreamStallGrapher.py.

References createfilelist.int, hpstanc_transforms.max, and split.

def parseStallMonitorOutput(f):
    processingSteps = []
    numStreams = 0
    maxNameSize = 0
    foundEventToStartFrom = False
    moduleNames = {}
    for rawl in f:
        l = rawl.strip()
        if not l or l[0] == '#':
            if len(l) > 5 and l[0:2] == "#M":
                (id,name)=tuple(l[2:].split())
                moduleNames[id] = name
            continue
        (step,payload) = tuple(l.split(None,1))
        payload=payload.split()

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

        numStreams = max(numStreams, stream+1)

        # 'S' = begin of event creation in source
        # 's' = end of event creation in source
        if step == 'S' or step == 's':
            name = kSourceFindEvent
            trans = kStarted
            # The start of an event is the end of the framework part
            if step == 's':
                trans = kFinished
        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
                name = moduleNames[moduleID]

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

            maxNameSize = max(maxNameSize, len(name))
            processingSteps.append((name,trans,stream,time))

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

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

def edmStreamStallGrapher.parseTracerOutput

(

f

)

Definition at line 123 of file edmStreamStallGrapher.py.

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

def parseTracerOutput(f):
    processingSteps = []
    numStreams = 0
    maxNameSize = 0
    startTime = 0
    for l in f:
        if l.find("processing event :") != -1:
            time = getTime(l)
            if startTime == 0:
                startTime = time
            time = time - startTime
            streamIndex = l.find("stream = ")
            stream = int(l[streamIndex+9:l.find(" ",streamIndex+10)])
            name = kSourceFindEvent
            trans = kFinished
            #the start of an event is the end of the framework part
            if l.find("starting:") != -1:
                trans = kStarted
            processingSteps.append((name,trans,stream,time))
            numStreams = max(numStreams, stream+1)
        if l.find("processing event for module") != -1:
            time = getTime(l)
            if startTime == 0:
                startTime = time
            time = time - startTime
            trans = kStarted
            stream = 0
            delayed = False
            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
            streamIndex = l.find("stream = ")
            stream = int( l[streamIndex+9:l.find(" ",streamIndex+10)])
            name = l.split("'")[1]
            maxNameSize = max(maxNameSize, len(name))
            processingSteps.append((name,trans,stream,time))
            numStreams = max(numStreams, stream+1)
        if l.find("event delayed read from source") != -1:
            time = getTime(l)
            if startTime == 0:
                startTime = time
            time = time - startTime
            trans = kStarted
            stream = 0
            delayed = False
            if l.find("finished:") != -1:
                trans = kFinished
            streamIndex = l.find("stream = ")
            stream = int(l[streamIndex+9:l.find(" ",streamIndex+10)])
            name = kSourceDelayedRead
            maxNameSize = max(maxNameSize, len(name))
            processingSteps.append((name,trans,stream,time))
            numStreams = max(numStreams, stream+1)
    f.close()
    return (processingSteps,numStreams,maxNameSize)


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

def edmStreamStallGrapher.printHelp

(

)

Definition at line 7 of file edmStreamStallGrapher.py.

Referenced by createPDFImage().

def printHelp():
    s = '''
To Use: Add the Tracer Service to the cmsRun job you want to check for
  stream stalls.  Make sure to use the 'printTimstamps' option
    cms.Service("Tracer", printTimestamps = cms.untracked.bool(True))
  After running the job, execute this script and pass the name of the
  log file to the script as the only command line argument.

To Read: The script will then print an 'ASCII art' stall graph which
  consists of the name of the module which either started or stopped
  running on a stream, and the number of modules running on each
  stream at that the moment in time. If the 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".  Once the first 4 events have finished
  processing, the program prints "FINISH INIT".  This is useful if one
  wants to ignore stalled caused by startup actions, e.g. reading
  conditions.

  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.'''
    return s


kStallThreshold=100 #in milliseconds
kTracerInput=False

#Stream states
kStarted=0
kFinished=1
kPrefetchEnd=2

#Special names

def edmStreamStallGrapher.printStalledModulesInOrder

(

stalledModules

)

Definition at line 289 of file edmStreamStallGrapher.py.

References join(), and hpstanc_transforms.max.

Referenced by createPDFImage().

def printStalledModulesInOrder(stalledModules):
    priorities = []
    maxNameSize = 0
    for name,t in stalledModules.iteritems():
        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.readLogFile

(

inputFile

)

Definition at line 205 of file edmStreamStallGrapher.py.

References chooseParser(), and tools.parseInput().

Referenced by createPDFImage().

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

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

def edmStreamStallGrapher.reduceSortedPoints

(

ps

)

Definition at line 325 of file edmStreamStallGrapher.py.

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

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

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