GMPBase.py

Go to the documentation of this file.

from Gaudi.Configuration import *
from Configurables import EvtCounter
from GaudiPython import AppMgr, gbl, setOwnership, PyAlgorithm, SUCCESS,FAILURE, InterfaceCast
from ROOT import TBufferFile, TBuffer
import multiprocessing
from multiprocessing import Process, Queue, JoinableQueue, Event
from multiprocessing import cpu_count, current_process
from multiprocessing.queues import Empty
from pTools import *
import time, sys, os
from ROOT import TParallelMergingFile
# This script contains the bases for the Gaudi MultiProcessing (GMP)
# classes

# There are three classes :
#   Reader
#   Worker
#   Writer

# Each class needs to perform communication with the others
# For this, we need a means of communication, which will be based on
# the python multiprocessing package
# This is provided in SPI pytools package
# cmt line : use pytools v1.1 LCG_Interfaces
# The PYTHONPATH env variable may need to be modified, as this might
# still point to 1.0_python2.5

# Each class will need Queues, and a defined method for using these
# queues.
# For example, as long as there is something in the Queue, both ends
# of the queue must be open
# Also, there needs to be proper Termination flags and criteria
# The System should be error proof.


# Constants -------------------------------------------------------------------
NAP = 0.001
MB  = 1024.0*1024.0
# waits to guard against hanging, in seconds
WAIT_INITIALISE   = 60*5
WAIT_FIRST_EVENT  = 60*3
WAIT_SINGLE_EVENT = 60*6
WAIT_FINALISE     = 60*2
STEP_INITIALISE   = 10
STEP_EVENT        = 2
STEP_FINALISE     = 10

# My switch for direct switching on/off Smaps Algorithm in GaudiPython AppMgr
SMAPS = False

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

# definitions
# ----------
# used to convert stored histos (in AIDA format) to ROOT format
aida2root = gbl.Gaudi.Utils.Aida2ROOT.aida2root

# used to check which type of histo we are dealing with
# i.e. if currentHisto in aidatypes : pass
aidatypes = ( gbl.AIDA.IHistogram,
              gbl.AIDA.IHistogram1D,
              gbl.AIDA.IHistogram2D,
              gbl.AIDA.IHistogram3D,
              gbl.AIDA.IProfile1D,
              gbl.AIDA.IProfile2D,
              gbl.AIDA.IBaseHistogram  )  # extra?

# similar to aidatypes
thtypes   = ( gbl.TH1D, gbl.TH2D, gbl.TH3D, gbl.TProfile, gbl.TProfile2D )

# Types of OutputStream in Gaudi
WRITERTYPES  = {  'EvtCollectionStream'    : "tuples",
                  'InputCopyStream'        : "events",
                  'OutputStream'           : "events",
                  'RecordStream'           : "records",
                  'RunRecordStream'        : "records",
                  'SequentialOutputStream' : "events",
                  'TagCollectionStream'    : "tuples"   }

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

class MiniWriter( object ) :
    '''
    A class to represent a writer in the GaudiPython configuration
    It can be non-trivial to access the name of the output file; it may be
    specified in the DataSvc, or just on the writer, may be a list, or string
    Also, there are three different types of writer (events, records, tuples)
    so this bootstrap class provides easy access to this info while configuring
    '''
    def __init__( self, writer, wType, config ) :
        self.w     = writer
        self.wType = wType
        # set parameters for directly accessing the correct
        #   part of the configuration, so that later, we can do
        #   config[ key ].Output = modified(output)
        self.key          = None
        self.output       = None
        self.ItemList     = None
        self.OptItemList  = None
        #
        self.wName = writer.getName()
        # Now process the Writer, find where the output is named
        self.woutput     = None
        self.datasvcName = None
        self.svcOutput   = None
        if hasattr( self.w, "Output" ) :
            self.woutput = self.w.Output
            self.getItemLists( config )
            self.set( self.wName, self.w.Output )
            return
        else :
            # if there is no output file, get it via the datasvc
            # (every writer always has a datasvc property)
            self.datasvcName = self.w.EvtDataSvc
            datasvc = config[ self.datasvcName ]
            if hasattr( datasvc, "Output" ) :
                self.getItemLists( config )
                self.set( self.datasvcName, datasvc.Output )
                return

    def getNewName( self, replaceThis, withThis, extra='' ) :
        # replace one pattern in the output name string
        #  with another, and return the Output name
        # It *might* be in a list, so check for this
        #
        # @param extra : might need to add ROOT flags
        #                i.e.: OPT='RECREATE', or such
        assert replaceThis.__class__.__name__ == 'str'
        assert    withThis.__class__.__name__ == 'str'
        old = self.output
        lst = False
        if old.__class__.__name__ == 'list' :
            old = self.output[0]
            lst = True
        new = old.replace( replaceThis, withThis )
        new += extra
        if lst :
            return [ new ]
        else :
            return new

    def getItemLists( self, config ) :
        # the item list
        if hasattr( self.w, "ItemList" ) :
            self.ItemList = self.w.ItemList
        else :
            datasvc = config[ self.w.EvtDataSvc ]
            if hasattr( datasvc, "ItemList" ) :
                self.ItemList = datasvc.ItemList
        # The option item list; possibly not a valid variable
        if hasattr( self.w, "OptItemList" ) :
            self.OptItemList = self.w.OptItemList
        else :
            datasvc = config[ self.w.EvtDataSvc ]
            if hasattr( datasvc, "OptItemList" ) :
                self.OptItemList = datasvc.OptItemList
        return

    def set( self, key, output ) :
        self.key         = key
        self.output      = output
        return

    def __repr__( self ) :
        s  = ""
        line = '-'*80
        s += (line+'\n')
        s += "Writer         : %s\n"%( self.wName  )
        s += "Writer Type    : %s\n"%( self.wType  )
        s += "Writer Output  : %s\n"%( self.output )
        s += "DataSvc        : %s\n"%( self.datasvcName )
        s += "DataSvc Output : %s\n"%( self.svcOutput   )
        s += '\n'
        s += "Key for config : %s\n"%( self.key    )
        s += "Output File    : %s\n"%( self.output )
        s += "ItemList       : %s\n"%( self.ItemList )
        s += "OptItemList    : %s\n"%( self.OptItemList )
        s += (line+'\n')
        return s

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

class CollectHistograms( PyAlgorithm ) :
    '''
    GaudiPython algorithm used to clean up histos on the Reader and Workers
    Only has a finalize method()
    This retrieves a dictionary of path:histo objects and sends it to the
    writer.  It then waits for a None flag : THIS IS IMPORTANT, as if
    the algorithm returns before ALL histos have been COMPLETELY RECEIVED
    at the writer end, there will be an error.
    '''
    def __init__( self, gmpcomponent ) :
        PyAlgorithm.__init__( self )
        self._gmpc = gmpcomponent
        self.log = self._gmpc.log
        return None
    def execute( self ) :
        return SUCCESS
    def finalize( self ) :
    self.log.info('CollectHistograms Finalise (%s)'%(self._gmpc.nodeType))
        self._gmpc.hDict = self._gmpc.dumpHistograms( )
        ks = self._gmpc.hDict.keys()
        self.log.info('%i Objects in Histogram Store'%(len(ks)))

        # crashes occurred due to Memory Error during the sending of hundreds
        # histos on slc5 machines, so instead, break into chunks
        # send 100 at a time
        chunk = 100
        reps = len(ks)/chunk + 1
        for i in xrange(reps) :
            someKeys = ks[i*chunk : (i+1)*chunk]
            smalld = dict( [(key, self._gmpc.hDict[key]) for key in someKeys] )
            self._gmpc.hq.put( (self._gmpc.nodeID, smalld) )
        # "finished" Notification
    self.log.debug('Signalling end of histos to Writer')
        self._gmpc.hq.put( 'HISTOS_SENT' )
        self.log.debug( 'Waiting on Sync Event' )
        self._gmpc.sEvent.wait()
        self.log.debug( 'Histo Sync Event set, clearing and returning' )
        self._gmpc.hvt.clearStore()
        root = gbl.DataObject()
        setOwnership(root, False)
        self._gmpc.hvt.setRoot( '/stat', root )
        return SUCCESS

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

class EventCommunicator( object ) :
    # This class is responsible for communicating Gaudi Events via Queues
    # Events are communicated as TBufferFiles, filled either by the
    # TESSerializer, or the GaudiSvc, "IPCSvc"
    def __init__( self, GMPComponent, qin, qout ) :
        self._gmpc = GMPComponent
        self.log   = self._gmpc.log
        # maximum capacity of Queues
        self.maxsize = 50
        # central variables : Queues
        self.qin  = qin
        self.qout = qout

        # flags
        self.allsent = False
        self.allrecv = False

        # statistics storage
        self.nSent    = 0    # counter : items sent
        self.nRecv    = 0    # counter : items received
        self.sizeSent = 0    # measure : size of events sent ( tbuf.Length() )
        self.sizeRecv = 0    # measure : size of events in   ( tbuf.Length() )
        self.qinTime  = 0    # time    : receiving from self.qin
        self.qoutTime = 0    # time    : sending on qout

    def send( self, item ) :
        # This class manages the sending of a TBufferFile Event to a Queue
        # The actual item to be sent is a tuple : ( evtNumber, TBufferFile )
        assert item.__class__.__name__ == 'tuple'
        startTransmission = time.time()
        self.qout.put( item )
        # allow the background thread to feed the Queue; not 100% guaranteed to
        # finish before next line
        while self.qout._buffer : time.sleep( NAP )
        self.qoutTime += time.time()-startTransmission
        self.sizeSent += item[1].Length()
        self.nSent += 1
        return SUCCESS

    def receive( self, timeout=None ) :
        # Receive items from self.qin
        startWait = time.time()
        try :
            itemIn = self.qin.get( timeout=timeout )
        except Empty :
            return None
        self.qinTime += time.time()-startWait
        self.nRecv += 1
        if itemIn.__class__.__name__ == 'tuple' :
            self.sizeRecv += itemIn[1].Length()
        else :
            self.nRecv -= 1
        try :
            self.qin.task_done()
        except :
            self._gmpc.log.warning('TASK_DONE called too often by : %s'\
                                    %(self._gmpc.nodeType))
        return itemIn

    def finalize( self ) :
        self.log.info('Finalize Event Communicator : %s %s'%(self._gmpc, self._gmpc.nodeType))
        # Reader sends one flag for each worker
        # Workers send one flag each
        # Writer sends nothing (it's at the end of the chain)
        if   self._gmpc.nodeType == 'Reader' : downstream = self._gmpc.nWorkers
        elif self._gmpc.nodeType == 'Writer' : downstream = 0
        elif self._gmpc.nodeType == 'Worker' : downstream = 1

        for i in xrange(downstream) :
            self.qout.put( 'FINISHED' )
        if self._gmpc.nodeType != 'Writer' :
            self.qout.join()
        # Now some reporting...
        self.statistics( )

    def statistics( self ) :
        self.log.name = '%s-%i Audit '%(self._gmpc.nodeType,self._gmpc.nodeID)
        self.log.info ( 'Items Sent     : %i'%(self.nSent) )
        self.log.info ( 'Items Received : %i'%(self.nRecv) )
        self.log.info ( 'Data  Sent     : %i'%(self.sizeSent) )
        self.log.info ( 'Data  Received : %i'%(self.sizeRecv) )
        self.log.info ( 'Q-out Time     : %5.2f'%(self.qoutTime) )
        self.log.info ( 'Q-in  Time     : %5.2f'%(self.qinTime ) )

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

class TESSerializer( object ) :
    def __init__( self, gaudiTESSerializer, evtDataSvc,
                        nodeType, nodeID, log ) :
        self.T   = gaudiTESSerializer
        self.evt = evtDataSvc
        self.buffersIn  = []
        self.buffersOut = []
        self.nIn     = 0
        self.nOut    = 0
        self.tDump   = 0.0
        self.tLoad   = 0.0
        # logging
        self.nodeType = nodeType
        self.nodeID   = nodeID
        self.log      = log
    def Load( self, tbuf ) :
        root = gbl.DataObject()
        setOwnership( root, False )
        self.evt.setRoot( '/Event', root )
        t = time.time()
        self.T.loadBuffer( tbuf )
        self.tLoad   += (time.time() - t)
        self.nIn     += 1
        self.buffersIn.append( tbuf.Length() )
    def Dump( self ) :
        t = time.time()
        tb = TBufferFile( TBuffer.kWrite )
        self.T.dumpBuffer(tb)
        self.tDump += ( time.time()-t )
        self.nOut  += 1
        self.buffersOut.append( tb.Length() )
        return tb
    def Report( self ) :
        evIn       = "Events Loaded    : %i"%( self.nIn  )
        evOut      = "Events Dumped    : %i"%( self.nOut )
        din = sum( self.buffersIn )
        dataIn     = "Data Loaded      : %i"%(din)
        dataInMb   = "Data Loaded (MB) : %5.2f Mb"%(din/MB)
        if self.nIn :
            avgIn      = "Avg Buf Loaded   : %5.2f Mb"\
                          %( din/(self.nIn*MB) )
            maxIn      = "Max Buf Loaded   : %5.2f Mb"\
                          %( max(self.buffersIn)/MB )
        else :
            avgIn      = "Avg Buf Loaded   : N/A"
            maxIn      = "Max Buf Loaded   : N/A"
        dout = sum( self.buffersOut )
        dataOut    = "Data Dumped      : %i"%(dout)
        dataOutMb  = "Data Dumped (MB) : %5.2f Mb"%(dout/MB)
        if self.nOut :
            avgOut     = "Avg Buf Dumped   : %5.2f Mb"\
                          %( din/(self.nOut*MB) )
            maxOut     = "Max Buf Dumped   : %5.2f Mb"\
                          %( max(self.buffersOut)/MB )
        else :
            avgOut     = "Avg Buf Dumped   : N/A"
            maxOut     = "Max Buf Dumped   : N/A"
        dumpTime   = "Total Dump Time  : %5.2f"%( self.tDump )
        loadTime   = "Total Load Time  : %5.2f"%( self.tLoad )

        lines =  evIn     ,\
                 evOut    ,\
                 dataIn   ,\
                 dataInMb ,\
                 avgIn    ,\
                 maxIn    ,\
                 dataOut  ,\
                 dataOutMb,\
                 avgOut   ,\
                 maxOut   ,\
                 dumpTime ,\
                 loadTime
        self.log.name = "%s-%i TESSerializer"%(self.nodeType, self.nodeID)
        for line in lines :
            self.log.info( line )
        self.log.name = "%s-%i"%(self.nodeType, self.nodeID)

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

class GMPComponent( object ) :
    # This class will be the template for Reader, Worker and Writer
    # containing all common components
    # nodeId will be a numerical identifier for the node
    # -1 for reader
    # -2 for writer
    # 0,...,nWorkers-1 for the Workers
    def __init__( self, nodeType, nodeID, queues, events, params, subworkers   ) :
        # declare a Gaudi MultiProcessing Node
        # the nodeType is going to be one of Reader, Worker, Writer
        # qPair is going to be a tuple of ( qin, qout )
        # for sending and receiving
        # if nodeType is "Writer", it will be a list of qPairs,
        # as there's one queue-in from each Worker
        #
        # params is a tuple of (nWorkers, config, log)

        self.nodeType = nodeType
        current_process().name = nodeType

        # Synchronisation Event() objects for keeping track of the system
        self.initEvent, eventLoopSyncer, self.finalEvent = events
        self.eventLoopSyncer, self.lastEvent = eventLoopSyncer   # unpack tuple

        # necessary for knowledge of the system
        self.nWorkers, self.sEvent, self.config, self.log = params
        self.subworkers = subworkers
        self.nodeID   = nodeID

        # describe the state of the node by the current Event Number
        self.currentEvent = None

        # Unpack the Queues : (events, histos, filerecords)
        self.queues = queues
        self.num = 0

        ks = self.config.keys()
        self.app = None
        list = ["Brunel", "DaVinci", "Boole", "Gauss"]
        for k in list:
           if k in ks: self.app = k

    def Start( self ) :
        # define the separate process
        qPair, histq, fq = self.queues

        # Set up the Queue Mechanisms ( Event Communicators )
        if self.nodeType == 'Reader' or self.nodeType == 'Worker' :
            # Reader or Worker Node
            qin, qout = qPair
            self.evcom = EventCommunicator( self, qin, qout )
        else :
            # Writer : many queues in, no queue out
            assert self.nodeType == 'Writer'
            self.evcoms = []
            qsin = qPair[0]
            for q in qsin :
                ec = EventCommunicator( self, q, None )
                self.evcoms.append( ec )
        # Histogram Queue
        self.hq = histq
        # FileRecords Queue
        self.fq = fq

        # Universal Counters (available to all nodes)
        # Use sensibly!!!
        self.nIn  = 0
        self.nOut = 0

        # Status Flag (possibly remove later)
        self.stat = SUCCESS

        # Set logger name
        self.log.name = '%s-%i'%(self.nodeType, self.nodeID)

        # Heuristic variables
        # time for init, run, final, firstEventTime, totalTime
        self.iTime       = 0.0
        self.rTime       = 0.0
        self.fTime       = 0.0
        self.firstEvTime = 0.0
        self.tTime       = 0.0

        self.proc = Process( target=self.Engine )
        # Fork and start the separate process
        self.proc.start()




    def Engine( self ) :
        # This will be the method carried out by the Node
        # Different for all
        pass

    def processConfiguration( self ) :
        # Different for all ; customize Configuration for multicore
        pass

    def SetupGaudiPython( self ) :
        # This method will initialize the GaudiPython Tools
        # such as the AppMgr and so on
        self.a   = AppMgr()
        if SMAPS :
            from AlgSmapShot import SmapShot
            smapsLog = self.nodeType+'-'+str(self.nodeID)+'.smp'
            ss = SmapShot( logname=smapsLog )
            self.a.addAlgorithm( ss )
        self.evt = self.a.evtsvc()
        self.hvt = self.a.histsvc()
        self.fsr = self.a.filerecordsvc()
        self.inc = self.a.service('IncidentSvc','IIncidentSvc')
        self.pers = self.a.service( 'EventPersistencySvc', 'IAddressCreator' )
        self.ts  = gbl.GaudiMP.TESSerializer( self.evt._idp, self.pers )
        self.TS  = TESSerializer( self.ts, self.evt,
                                  self.nodeType, self.nodeID, self.log )
        return SUCCESS

    def StartGaudiPython( self ) :
        self.a.initialize()
        self.a.start()
        return SUCCESS

    def LoadTES( self, tbufferfile ) :
        root = gbl.DataObject()
        setOwnership(root, False)
        self.evt.setRoot( '/Event', root )
        self.ts.loadBuffer(tbufferfile)

    def getEventNumber( self ) :
      if self.app != 'Gauss':
        # Using getList or getHistoNames can result in the EventSelector
        # re-initialising connection to RootDBase, which costs a lot of
        # time... try to build a set of Header paths??

        # First Attempt : Unpacked Event Data
        lst = [ '/Event/Gen/Header',
                '/Event/Rec/Header' ]
        for l in lst :
            path = l
            try :
                n = self.evt[path].evtNumber()

                return n
            except :
                # No evt number at this path
                continue

        # second attepmt : try DAQ/RawEvent data
        # The Evt Number is in bank type 16, bank 0, data pt 4
        try :
            n = self.evt['/Event/DAQ/RawEvent'].banks(16)[0].data()[4]

            return n
        except :
            pass

        # Default Action
        if self.nIn > 0 or self.nOut > 0 :
            pass
        else :
            self.log.warning('Could not determine Event Number')
        return -1
      else:
           if self.nodeID == -1:
             self.num = self.num + 1

             return self.num

    def IdentifyWriters( self ) :
        #
        # Identify Writers in the Configuration
        #
        d = {}
        keys = [ "events", "records", "tuples", "histos" ]
        for k in keys :
            d[k] = []

        # Identify Writers and Classify
        wkeys = WRITERTYPES.keys()
        for v in self.config.values() :
            if v.__class__.__name__ in wkeys :
                writerType = WRITERTYPES[ v.__class__.__name__ ]
                d[writerType].append( MiniWriter(v, writerType, self.config) )
                if self.nodeID == 0 :
                    self.log.info('Writer Found : %s'%(v.name()))

        # Now Check for the Histogram Service
        if 'HistogramPersistencySvc' in  self.config.keys() :
            hfile =self.config['HistogramPersistencySvc'].getProp('OutputFile')
            d[ "histos" ].append( hfile )
        return d

    def dumpHistograms( self ) :
        '''
        Method used by the GaudiPython algorithm CollectHistos
        to obtain a dictionary of form { path : object }
        representing the Histogram Store
        '''
        nlist = self.hvt.getHistoNames( )
        histDict = {}
        objects = 0 ; histos = 0
        if nlist :
            for n in nlist :
                o = self.hvt[ n ]
                if type(o) in aidatypes :
                    o = aida2root(o)
                    histos  += 1
                else :
                    objects += 1
                histDict[ n ] = o
        else :
            print 'WARNING : no histograms to recover?'
        return histDict

    def Initialize( self ) :
        start = time.time()
        self.processConfiguration( )
        self.SetupGaudiPython( )
        self.initEvent.set()
        self.StartGaudiPython( )

        if self.app == 'Gauss':

            tool = self.a.tool( "ToolSvc.EvtCounter" )
            self.cntr = InterfaceCast( gbl.IEventCounter )( tool.getInterface() )
        else:
            self.cntr = None

        self.iTime = time.time() - start

    def Finalize( self ) :
        start = time.time()
        self.a.stop()
        self.a.finalize()
        self.log.info( '%s-%i Finalized'%(self.nodeType, self.nodeID) )
        self.finalEvent.set()
        self.fTime = time.time() - start

    def Report( self ) :
        self.log.name = "%s-%i Audit"%(self.nodeType, self.nodeID)
        allTime  = "Alive Time     : %5.2f"%(self.tTime)
        initTime = "Init Time      : %5.2f"%(self.iTime)
        frstTime = "1st Event Time : %5.2f"%(self.firstEvTime)
        runTime  = "Run Time       : %5.2f"%(self.rTime)
        finTime  = "Finalise Time  : %5.2f"%(self.fTime)
        tup = ( allTime, initTime, frstTime, runTime, finTime )
        for t in tup :
            self.log.info( t )
        self.log.name = "%s-%i"%(self.nodeType, self.nodeID)
        # and report from the TESSerializer
        self.TS.Report()

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

class Reader( GMPComponent )  :
    def __init__( self, queues, events, params, subworkers  ) :
        GMPComponent.__init__(self, 'Reader', -1, queues, events, params, subworkers  )

    def processConfiguration( self ) :
        # Reader :
        #   No algorithms
        #   No output
        #   No histos
        self.config[ 'ApplicationMgr' ].TopAlg    = []
        self.config[ 'ApplicationMgr' ].OutStream = []
        if "HistogramPersistencySvc" in self.config.keys() :
            self.config[ 'HistogramPersistencySvc' ].OutputFile = ''
        self.config['MessageSvc'].Format    = '[Reader]% F%18W%S%7W%R%T %0W%M'
        self.evtMax = self.config[ 'ApplicationMgr' ].EvtMax

    def DumpEvent( self ) :
        tb = TBufferFile( TBuffer.kWrite )
        # print '----Reader dumping Buffer!!!'
        self.ts.dumpBuffer( tb )
        # print '\tBuffer Dumped, size : %i'%( tb.Length() )
        return tb

    def DoFirstEvent( self ) :
        # Do First Event ------------------------------------------------------
        # Check Termination Criteria
        startFirst = time.time()
        self.log.info('Reader : First Event')
        if self.nOut == self.evtMax :
            self.log.info('evtMax( %i ) reached'%(self.evtMax))
            self.lastEvent.set()
            return SUCCESS
        else :
            # Continue to read, dump and send event
            self.a.run(1)
            if not bool(self.evt['/Event']) :
                self.log.warning('No More Events! (So Far : %i)'%(self.nOut))
                self.lastEvent.set()
                return SUCCESS
            else :
                # Popluate TESSerializer list and send Event
                if self.app == "Gauss":
                   lst = self.evt.getHistoNames()
                else:
                  try :
                    lst = self.evt.getList()
                    if self.app == "DaVinci":
                      daqnode = self.evt.retrieveObject( '/Event/DAQ' ).registry()
                      setOwnership( daqnode, False )
                      self.evt.getList( daqnode, lst, daqnode.address().par() )
                  except :
                    self.log.critical('Reader could not acquire TES List!')
                    self.lastEvent.set()
                    return FAILURE
                self.log.info('Reader : TES List : %i items'%(len(lst)))
                for l in lst :
                    self.ts.addItem(l)
                self.currentEvent = self.getEventNumber( )
                tb = self.TS.Dump( )
                self.log.info('First Event Sent')
                self.evcom.send( (self.currentEvent, tb) )
                self.nOut += 1
                self.eventLoopSyncer.set()
                self.evt.clearStore( )
                self.firstEvTime = time.time()-startFirst
                return SUCCESS

    def Engine( self ) :
        # rename process
        import os
        import ctypes
        libc = ctypes.CDLL('libc.so.6')
        name = str(self.nodeType) + str(self.nodeID) + '\0'
        libc.prctl(15,name,0,0,0)


        startEngine = time.time()
        self.log.name = 'Reader'
        self.log.info('Reader Process starting')

        self.Initialize()

        # add the Histogram Collection Algorithm
        self.a.addAlgorithm( CollectHistograms(self) )

        self.log.info('Reader Beginning Distribution')
        sc = self.DoFirstEvent( )
        if sc.isSuccess() :
            self.log.info('Reader First Event OK')
        else :
            self.log.critical('Reader Failed on First Event')
            self.stat = FAILURE

        # Do All Others -------------------------------------------------------
        while True :
            # Check Termination Criteria
            if self.nOut == self.evtMax :
                self.log.info('evtMax( %i ) reached'%(self.evtMax))
                break
            # Check Health
            if not self.stat.isSuccess() :
                self.log.critical( 'Reader is Damaged!' )
                break
            # Continue to read, dump and send event
            t = time.time()
            self.a.run(1)
            self.rTime += (time.time()-t)
            if not bool(self.evt['/Event']) :
                self.log.warning('No More Events! (So Far : %i)'%(self.nOut))
                break
            self.currentEvent = self.getEventNumber( )
            tb = self.TS.Dump( )
            self.evcom.send( (self.currentEvent, tb) )
            # clean up
            self.nOut += 1
            self.eventLoopSyncer.set()
            self.evt.clearStore( )
        self.log.info('Setting <Last> Event')
        self.lastEvent.set()

        # Finalize
        self.log.info( 'Reader : Event Distribution complete.' )
        self.evcom.finalize()
        self.Finalize()
        self.tTime = time.time() - startEngine
        self.Report()

# =============================================================================
class Subworker(GMPComponent):
    def __init__( self, workerID, queues, events, params, subworkers  ) :
        GMPComponent.__init__(self,'Worker', workerID, queues, events, params, subworkers )
        # Identify the writer streams
        self.writerDict = self.IdentifyWriters( )
        # Identify the accept/veto checks for each event
        self.acceptAlgs, self.requireAlgs, self.vetoAlgs = self.getCheckAlgs()
        self.log.info("Subworker-%i Created OK"%(self.nodeID))
        self.eventOutput = True

    def Engine( self ) :
        # rename process
        import os
        import ctypes
        libc = ctypes.CDLL('libc.so.6')
        name = str(self.nodeType) + str(self.nodeID) + '\0'
        libc.prctl(15,name,0,0,0)

        self.initEvent.set()
        startEngine = time.time()
        msg = self.a.service('MessageSvc')
        msg.Format = '[' + self.log.name + '] % F%18W%S%7W%R%T %0W%M'

        self.log.name = "Worker-%i"%(self.nodeID)
        self.log.info("Subworker %i starting Engine"%(self.nodeID))
        self.filerecordsAgent = FileRecordsAgent(self)

        # populate the TESSerializer itemlist
        self.log.info('EVT WRITERS ON WORKER : %i'\
                       %( len(self.writerDict['events'])))

        nEventWriters = len( self.writerDict[ "events" ] )
        self.a.addAlgorithm( CollectHistograms(self) )

        # Begin processing
        Go = True
        while Go :
            packet = self.evcom.receive( )
            if packet : pass
            else      : continue
            if packet == 'FINISHED' : break
            evtNumber, tbin = packet    # unpack
            if self.cntr != None:

                self.cntr.setEventCounter( evtNumber )

            self.nIn += 1
            self.TS.Load( tbin )

            t = time.time()
            sc = self.a.executeEvent()
            if self.nIn == 1 :
                self.firstEvTime = time.time()-t
            else :
                self.rTime += (time.time()-t)
            if sc.isSuccess() :
                pass
            else :
            self.log.name = "Worker-%i"%(self.nodeID)
                self.log.warning('Did not Execute Event')
                self.evt.clearStore()
                continue
            if self.isEventPassed() :
                pass
            else :
                self.log.name = "Worker-%i"%(self.nodeID)
                self.log.warning( 'Event did not pass : %i'%(evtNumber) )
                self.evt.clearStore()
                continue
            if self.eventOutput :
                # It may be the case of generating Event Tags; hence
                #   no event output
                self.currentEvent = self.getEventNumber( )
                tb = self.TS.Dump( )
                self.evcom.send( (self.currentEvent, tb) )
                self.nOut += 1
            self.inc.fireIncident(gbl.Incident('Subworker','EndEvent'))
            self.eventLoopSyncer.set()
            self.evt.clearStore( )
        self.log.name = "Worker-%i"%(self.nodeID)
        self.log.info('Setting <Last> Event %s' %(self.nodeID))
        self.lastEvent.set()

        self.evcom.finalize()
        # Now send the FileRecords and stop/finalize the appMgr
        self.filerecordsAgent.SendFileRecords()
        self.tTime = time.time()-startEngine
        self.Finalize()
        self.Report()
        #self.finalEvent.set()

    def SetServices(self,a, evt, hvt, fsr, inc, pers, ts , cntr):
        self.a = a
        self.evt = evt
        self.hvt = hvt
        self.fsr = fsr
        #self.inc = inc
        self.inc = self.a.service('IncidentSvc','IIncidentSvc')
        self.pers = pers
        self.ts  = ts
        self.cntr = cntr
        self.TS  = TESSerializer( self.ts, self.evt,
                                  self.nodeType, self.nodeID, self.log )


    def getCheckAlgs( self ) :
        '''
        For some output writers, a check is performed to see if the event has
        executed certain algorithms.
        These reside in the AcceptAlgs property for those writers
        '''
        acc = []
        req = []
        vet = []
        for m in self.writerDict[ "events" ] :
            if hasattr(m.w, 'AcceptAlgs')  : acc += m.w.AcceptAlgs
            if hasattr(m.w, 'RequireAlgs') : req += m.w.RequireAlgs
            if hasattr(m.w, 'VetoAlgs')    : vet += m.w.VetoAlgs
        return (acc, req, vet)


    def checkExecutedPassed( self, algName ) :
        if  self.a.algorithm( algName )._ialg.isExecuted()\
        and self.a.algorithm( algName )._ialg.filterPassed() :
            return True
        else :
            return False

    def isEventPassed( self ) :
        '''
        Check the algorithm status for an event.
        Depending on output writer settings, the event
          may be declined based on various criteria.
        This is a transcript of the check that occurs in GaudiSvc::OutputStream
        '''
        passed = False

        self.log.debug('self.acceptAlgs is %s'%(str(self.acceptAlgs)))
        if self.acceptAlgs :
            for name in self.acceptAlgs :
                if self.checkExecutedPassed( name ) :
                    passed = True
                    break
        else :
            passed = True

        self.log.debug('self.requireAlgs is %s'%(str(self.requireAlgs)))
        for name in self.requireAlgs :
            if self.checkExecutedPassed( name ) :
                pass
            else :
                self.log.info('Evt declined (requireAlgs) : %s'%(name) )
                passed = False

        self.log.debug('self.vetoAlgs is %s'%(str(self.vetoAlgs)))
        for name in self.vetoAlgs :
            if self.checkExecutedPassed( name ) :
                pass
            else :
                self.log.info( 'Evt declined : (vetoAlgs) : %s'%(name) )
                passed = False
        return passed

# =============================================================================
class Worker( GMPComponent ) :
    def __init__( self, workerID, queues, events, params , subworkers ) :
        GMPComponent.__init__(self,'Worker', workerID, queues, events, params, subworkers )
        # Identify the writer streams
        self.writerDict = self.IdentifyWriters( )
        # Identify the accept/veto checks for each event
        self.acceptAlgs, self.requireAlgs, self.vetoAlgs = self.getCheckAlgs()
        self.log.name = "Worker-%i"%(self.nodeID)
        self.log.info("Worker-%i Created OK"%(self.nodeID))
        self.eventOutput = True

    def processConfiguration( self ) :

        # Worker :
        #   No input
        #   No output
        #   No Histos
        self.config[ 'EventSelector'  ].Input     = []
        self.config[ 'ApplicationMgr' ].OutStream = []
        if "HistogramPersistencySvc" in self.config.keys() :
            self.config[ 'HistogramPersistencySvc' ].OutputFile = ''
        formatHead = '[Worker-%i] '%(self.nodeID)
        self.config['MessageSvc'].Format = formatHead+'% F%18W%S%7W%R%T %0W%M'

        for key, lst in self.writerDict.iteritems() :
            self.log.info( 'Writer Type : %s\t : %i'%(key, len(lst)) )

        for m in self.writerDict[ "tuples" ] :
            # rename Tuple output file with an appendix
            # based on worker id, for merging later
            newName = m.getNewName( '.', '.w%i.'%(self.nodeID) )
            self.config[ m.key ].Output = newName

        # Suppress INFO Output for all but Worker-0
        #if self.nodeID == 0 :
        #    pass
        #else                :
        #    self.config[ 'MessageSvc' ].OutputLevel = ERROR

        if self.app == "Gauss":
          try:
            if "ToolSvc.EvtCounter" not in self.config:
                from Configurables import EvtCounter
                counter = EvtCounter()
            else:
                counter = self.config["ToolSvc.EvtCounter"]
            counter.UseIncident = False
          except:
            # ignore errors when trying to change the configuration of the EvtCounter
            self.log.warning('Cannot configure EvtCounter')

    def Engine( self ) :

        # rename process
        import os
        import ctypes
        libc = ctypes.CDLL('libc.so.6')
        name = str(self.nodeType) + str(self.nodeID) + '\0'
        libc.prctl(15,name,0,0,0)

        startEngine = time.time()
        self.log.info("Worker %i starting Engine"%(self.nodeID))
        self.Initialize()
        self.filerecordsAgent = FileRecordsAgent(self)

        # populate the TESSerializer itemlist
        self.log.info('EVT WRITERS ON WORKER : %i'\
                       %( len(self.writerDict['events'])))

        nEventWriters = len( self.writerDict[ "events" ] )
        if nEventWriters :
            itemList = set()
            optItemList = set()
            for m in self.writerDict[ "events" ] :
                for item in m.ItemList :
                    hsh = item.find( '#' )
                    if hsh != -1:
                      item = item[:hsh]
                    itemList.add( item )
                for item in m.OptItemList :
                    hsh = item.find( '#' )
                    if hsh != -1:
                      item = item[:hsh]
                    optItemList.add( item )
            # If an item is mandatory and optional, keep it only in the optional list
            itemList -= optItemList
            for item in sorted( itemList ):
                self.log.info( ' adding ItemList Item to ts : %s' % ( item ) )
                self.ts.addItem( item )
            for item in sorted( optItemList ):
                self.log.info( ' adding Optional Item to ts : %s' % ( item ) )
                self.ts.addOptItem( item )
        else :
            self.log.info( 'There is no Event Output for this app' )
            self.eventOutput = False

        # Begin processing
        Go = True
        while Go :
            packet = self.evcom.receive( )
            if packet : pass
            else      : continue
            if packet == 'FINISHED' : break
            evtNumber, tbin = packet    # unpack
            if self.cntr != None:
                self.cntr.setEventCounter( evtNumber )

            # subworkers are forked before the first event is processed
            # reader-thread for ConDB must be closed and reopened in each subworker
            # this is done by disconnect()
            if self.nIn == 0:

                self.log.info("Fork new subworkers and disconnect from CondDB")
                condDB = self.a.service('CondDBCnvSvc', gbl.ICondDBReader)
                condDB.disconnect()

                # Fork subworkers and share services
                for k in self.subworkers:
                   k.SetServices(self.a, self.evt, self.hvt, self.fsr, self.inc, self.pers, self.ts, self.cntr)
                   k.Start()
                   self.a.addAlgorithm( CollectHistograms(self) )
            self.nIn += 1
            self.TS.Load( tbin )

            t = time.time()
            sc = self.a.executeEvent()
            if self.nIn == 1 :
                self.firstEvTime = time.time()-t
            else :
                self.rTime += (time.time()-t)
            if sc.isSuccess() :
                pass
            else :
                self.log.warning('Did not Execute Event')
                self.evt.clearStore()
                continue
            if self.isEventPassed() :
                pass
            else :
                self.log.warning( 'Event did not pass : %i'%(evtNumber) )
                self.evt.clearStore()
                continue
            if self.eventOutput :
                # It may be the case of generating Event Tags; hence
                #   no event output
                self.currentEvent = self.getEventNumber( )
                tb = self.TS.Dump( )
                self.evcom.send( (self.currentEvent, tb) )
                self.nOut += 1
            self.inc.fireIncident(gbl.Incident('Worker','EndEvent'))
            self.eventLoopSyncer.set()
            self.evt.clearStore( )
        self.log.info('Setting <Last> Event')
        self.lastEvent.set()

        self.evcom.finalize()
        self.log.info( 'Worker-%i Finished Processing Events'%(self.nodeID) )
        # Now send the FileRecords and stop/finalize the appMgr
        self.filerecordsAgent.SendFileRecords()
        self.Finalize()
        self.tTime = time.time()-startEngine
        self.Report()

        for k in self.subworkers:
            self.log.info('Join subworkers')
            k.proc.join()

    def getCheckAlgs( self ) :
        '''
        For some output writers, a check is performed to see if the event has
        executed certain algorithms.
        These reside in the AcceptAlgs property for those writers
        '''
        acc = []
        req = []
        vet = []
        for m in self.writerDict[ "events" ] :
            if hasattr(m.w, 'AcceptAlgs')  : acc += m.w.AcceptAlgs
            if hasattr(m.w, 'RequireAlgs') : req += m.w.RequireAlgs
            if hasattr(m.w, 'VetoAlgs')    : vet += m.w.VetoAlgs
        return (acc, req, vet)


    def checkExecutedPassed( self, algName ) :
        if  self.a.algorithm( algName )._ialg.isExecuted()\
        and self.a.algorithm( algName )._ialg.filterPassed() :
            return True
        else :
            return False

    def isEventPassed( self ) :
        '''
        Check the algorithm status for an event.
        Depending on output writer settings, the event
          may be declined based on various criteria.
        This is a transcript of the check that occurs in GaudiSvc::OutputStream
        '''
        passed = False

        self.log.debug('self.acceptAlgs is %s'%(str(self.acceptAlgs)))
        if self.acceptAlgs :
            for name in self.acceptAlgs :
                if self.checkExecutedPassed( name ) :
                    passed = True
                    break
        else :
            passed = True

        self.log.debug('self.requireAlgs is %s'%(str(self.requireAlgs)))
        for name in self.requireAlgs :
            if self.checkExecutedPassed( name ) :
                pass
            else :
                self.log.info('Evt declined (requireAlgs) : %s'%(name) )
                passed = False

        self.log.debug('self.vetoAlgs is %s'%(str(self.vetoAlgs)))
        for name in self.vetoAlgs :
            if self.checkExecutedPassed( name ) :
                pass
            else :
                self.log.info( 'Evt declined : (vetoAlgs) : %s'%(name) )
                passed = False
        return passed

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

class Writer( GMPComponent ) :
    def __init__( self, queues, events, params, subworkers  ) :
        GMPComponent.__init__(self,'Writer', -2, queues, events, params, subworkers )
        # Identify the writer streams
        self.writerDict = self.IdentifyWriters( )
        # This keeps track of workers as they finish
        self.status = [False]*self.nWorkers
        self.log.name = "Writer--2"

    def processConfiguration( self ) :
        # Writer :
        #   No input
        #   No Algs
        self.config[ 'ApplicationMgr' ].TopAlg = []
        self.config[ 'EventSelector'  ].Input  = []

        self.config['MessageSvc'].Format = '[Writer] % F%18W%S%7W%R%T %0W%M'

        # Now process the output writers
        for key, lst in self.writerDict.iteritems() :
            self.log.info( 'Writer Type : %s\t : %i'%(key, len(lst)) )

        # Modify the name of the output file to reflect that it came
        #  from a parallel processing
        #
        # Event Writers
        for m in self.writerDict[ "events" ] :
            self.log.debug( 'Processing Event Writer : %s'%(m) )
            newName = m.getNewName( '.', '.p%i.'%self.nWorkers )
            self.config[ m.key ].Output = newName

        # Now, if there are no event writers, the FileRecords file
        #   will fail to open, as it only opens an UPDATE version
        #   of the existing Event Output File
        # So, if there are no event writers, edit the string of the
        #   FileRecord Writer

        # FileRecords Writers
        for m in self.writerDict[ "records" ] :
            self.log.debug( 'Processing FileRecords Writer: %s'%(m) )
            newName = m.getNewName( '.', '.p%i.'%self.nWorkers,
                                       extra=" OPT='RECREATE'" )
            self.config[ m.key ].Output = newName

        # same for histos
        hs = "HistogramPersistencySvc"
        n = None
        if hs in self.config.keys() :
            n = self.config[ hs ].OutputFile
        if n :
            newName=self.config[hs].OutputFile.replace('.',\
                                                       '.p%i.'%(self.nWorkers))
            self.config[ hs ].OutputFile = newName

    def Engine( self ) :
        # rename process
        import os
        import ctypes
        libc = ctypes.CDLL('libc.so.6')
        name = str(self.nodeType) + str(self.nodeID) + '\0'
        libc.prctl(15,name,0,0,0)

        startEngine = time.time()
        self.Initialize()
        self.histoAgent = HistoAgent( self )
        self.filerecordsAgent = FileRecordsAgent( self )

        # Begin processing
        Go      = True
        current = -1
        stopCriteria = self.nWorkers
        while Go :
            current = (current+1)%self.nWorkers
            packet = self.evcoms[current].receive( timeout=0.01 )
            if packet == None :
                continue
            if packet == 'FINISHED' :
                self.log.info('Writer got FINISHED flag : Worker %i'%(current))

                self.status[current] = True
                if all(self.status) :
                    self.log.info('FINISHED recd from all workers, break loop')
                    break
                continue
            # otherwise, continue as normal
            self.nIn += 1    # update central count (maybe needed by FSR store)
            evtNumber, tbin = packet    # unpack
            self.TS.Load( tbin )
            t  = time.time()
            self.a.executeEvent()
            self.rTime += ( time.time()-t )
            self.currentEvent = self.getEventNumber( )
            self.evt.clearStore( )
            self.eventLoopSyncer.set()
        self.log.name = "Writer--2"
        self.log.info('Setting <Last> Event')
        self.lastEvent.set()

        # finalisation steps
        [ e.finalize() for e in self.evcoms ]
        # Now do Histograms
        sc = self.histoAgent.Receive()
        sc = self.histoAgent.RebuildHistoStore()
        if sc.isSuccess() : self.log.info( 'Histo Store rebuilt ok' )
        else              : self.log.warning( 'Histo Store Error in Rebuild' )

        # Now do FileRecords
        sc = self.filerecordsAgent.Receive()
        self.filerecordsAgent.Rebuild()
        self.Finalize()
        #self.rTime = time.time()-startEngine
        self.Report()

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



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

class Coord( object ) :
    def __init__( self, nWorkers, config, log ) :

        self.log = log
        self.config = config
        # set up Logging
        self.log.name = 'GaudiPython-Parallel-Logger'
        self.log.info( 'GaudiPython Parallel Process Co-ordinator beginning' )

        if nWorkers == -1 :
            # The user has requested all available cpus in the machine
            self.nWorkers = cpu_count()
        else :
            self.nWorkers = nWorkers


        self.qs = self.SetupQueues( )    # a dictionary of queues (for Events)
        self.hq = JoinableQueue( )       # for Histogram data
        self.fq = JoinableQueue( )       # for FileRecords data

        # Make a Syncer for Initalise, Run, and Finalise
        self.sInit = Syncer( self.nWorkers, self.log,
                             limit=WAIT_INITIALISE,
                             step=STEP_INITIALISE    )
        self.sRun  = Syncer( self.nWorkers, self.log,
                             manyEvents=True,
                             limit=WAIT_SINGLE_EVENT,
                             step=STEP_EVENT,
                             firstEvent=WAIT_FIRST_EVENT )
        self.sFin  = Syncer( self.nWorkers, self.log,
                             limit=WAIT_FINALISE,
                             step=STEP_FINALISE )
        # and one final one for Histogram Transfer
        self.histSyncEvent = Event()

        # params are common to al subprocesses
        params = (self.nWorkers, self.histSyncEvent, self.config, self.log)

        self.subworkers = []
        # Declare SubProcesses!
        for i in range(1, self.nWorkers ) :
            sub = Subworker( i, self.getQueues(i), self.getSyncEvents(i), params, self.subworkers )
            self.subworkers.append( sub )
        self.reader= Reader(self.getQueues(-1), self.getSyncEvents(-1), params, self.subworkers)
        self.workers = []
        wk = Worker( 0, self.getQueues(0), self.getSyncEvents(0), params, self.subworkers )
        self.writer= Writer(self.getQueues(-2), self.getSyncEvents(-2), params, self.subworkers)

        self.system = []
        self.system.append(self.writer)
        self.system.append(wk)
        self.system.append(self.reader)

    def getSyncEvents( self, nodeID ) :
        init = self.sInit.d[nodeID].event
        run  = ( self.sRun.d[nodeID].event, self.sRun.d[nodeID].lastEvent )
        fin  = self.sFin.d[nodeID].event
        return ( init, run, fin )

    def getQueues( self, nodeID ) :
        eventQ = self.qs[ nodeID ]
        histQ  = self.hq
        fsrQ   = self.fq
        return ( eventQ, histQ, fsrQ )

    def Go( self ) :

        # Initialise
        self.log.name = 'GaudiPython-Parallel-Logger'
        self.log.info( 'INITIALISING SYSTEM' )

        # Start reader, writer and main worker
        for p in self.system :
            p.Start()

        sc = self.sInit.syncAll(step="Initialise")
        if sc == SUCCESS: pass
        else  : self.Terminate() ; return FAILURE

        # Run
        self.log.name = 'GaudiPython-Parallel-Logger'
        self.log.info( 'RUNNING SYSTEM' )
        sc = self.sRun.syncAll(step="Run")
        if sc == SUCCESS: pass
        else  : self.Terminate() ; return FAILURE

        # Finalise
        self.log.name = 'GaudiPython-Parallel-Logger'
        self.log.info( 'FINALISING SYSTEM' )
        sc = self.sFin.syncAll(step="Finalise")
        if sc == SUCCESS: pass
        else  : self.Terminate() ; return FAILURE

        # if we've got this far, finally report SUCCESS
        self.log.info( "Cleanly join all Processes" )
        self.Stop()
        self.log.info( "Report Total Success to Main.py" )
        return SUCCESS

    def Terminate( self ) :
        # Brutally kill sub-processes
        children = multiprocessing.active_children()
        for i in children:
            i.terminate()

        #self.writer.proc.terminate()
        #[ w.proc.terminate() for w in self.workers]
        #self.reader.proc.terminate()

    def Stop( self ) :
        # procs should be joined in reverse order to launch
        self.system.reverse()
        for s in self.system :
            s.proc.join()
        return SUCCESS

    def SetupQueues( self ) :
        # This method will set up the network of Queues needed
        # N Queues = nWorkers + 1
        # Each Worker has a Queue in, and a Queue out
        # Reader has Queue out only
        # Writer has nWorkers Queues in

        # one queue from Reader-Workers
        rwk = JoinableQueue()
        # one queue from each worker to writer
        workersWriter = [ JoinableQueue() for i in xrange(self.nWorkers) ]
        d = {}
        d[-1] = (None, rwk)              # Reader
        d[-2] = (workersWriter, None)    # Writer
        for i in xrange(self.nWorkers) : d[i] =  (rwk, workersWriter[i])
        return d

# ============================= EOF ===========================================