GaudiMP.pTools.FileRecordsAgent Class Reference

Public Member Functions
def	__init__ (self, gmpComponent)
def	localCmp (self, tupA, tupB)
def	SendFileRecords (self)
def	Receive (self)
def	Rebuild (self)
def	MergeFSRobject (self, sourceNode, path, ob)
def	ProcessTimeSpanFSR (self, path, ob)
def	ProcessEventCountFSR (self, path, ob)
def	MergeLumiFSR (self, path, keyedC)

Public Attributes
	fsr
	q
	log
	objectsIn
	objectsOut

Private Attributes
	_gmpc

Detailed Description

Definition at line 240 of file pTools.py.

Constructor & Destructor Documentation

__init__()

def GaudiMP.pTools.FileRecordsAgent.__init__	(		self,
			gmpComponent
	)

Definition at line 241 of file pTools.py.

    def __init__(self, gmpComponent):
        self._gmpc = gmpComponent
        self.fsr = self._gmpc.fsr
        self.q = self._gmpc.fq
        self.log = self._gmpc.log
        self.objectsIn = []  # used for collecting FSR store objects
        self.objectsOut = []
 

Member Function Documentation

localCmp()

def GaudiMP.pTools.FileRecordsAgent.localCmp	(		self,
			tupA,
			tupB
	)

Definition at line 249 of file pTools.py.

    def localCmp(self, tupA, tupB):
        # sort tuples by a particular element
        # for the sort() method
        ind = 0
        valA = tupA[ind]
        valB = tupB[ind]
        if valA < valB:
            return -1
        elif valA > valB:
            return 1
        else:
            return 0
 

MergeFSRobject()

def GaudiMP.pTools.FileRecordsAgent.MergeFSRobject	(		self,
			sourceNode,
			path,
			ob
	)

Definition at line 394 of file pTools.py.

    def MergeFSRobject(self, sourceNode, path, ob):
        # Merge Non-Empty Keyed Container from Worker>0
        if path == "/FileRecords/TimeSpanFSR":
            # TimeSpanFSR is a straightforward case
            self.ProcessTimeSpanFSR(path, ob)
        elif path == "/FileRecords/EventCountFSR":
            # Event Counter is also easy
            self.ProcessEventCountFSR(path, ob)
        # now other cases may not be so easy...
        elif (
            "KeyedContainer" in ob.__class__.__name__
            and "LumiFSR" in ob.__class__.__name__
        ):
            # Keyed Container of LumiFSRs : extract and re-register
            self.MergeLumiFSR(path, ob)
        else:
            self.log.info("Skipping Merge of %s at %s" % (ob.__class__.__name__, path))
 

MergeLumiFSR()

def GaudiMP.pTools.FileRecordsAgent.MergeLumiFSR	(		self,
			path,
			keyedC
	)

Definition at line 438 of file pTools.py.

    def MergeLumiFSR(self, path, keyedC):
        from ROOT import string
 
        # Fetch the first lumi
        keyedContainer = self.fsr.retrieveObject(path)
        # The LumiFSR KeyedContainers only have one object
        assert keyedContainer.numberOfObjects() == 1
        l = keyedContainer.containedObject(0)
        baseLumi = LumiFSR(l)
        # Now deal with the argument Non-empty Keyed Container of LumiFSRs
        nCont = keyedC.numberOfObjects()
        for i in range(nCont):
            obj = keyedC.containedObject(i)
            nextLumi = LumiFSR(obj)
            baseLumi.merge(nextLumi)
        # Now Rebuild and ReRegister
        newLumi = gbl.LHCb.LumiFSR()
        for r in baseLumi.runs:
            newLumi.addRunNumber(r)
        for f in baseLumi.files:
            newLumi.addFileID(string(f))
        for k in baseLumi.keys:
            increment, integral = baseLumi.info[k]
            newLumi.addInfo(k, increment, integral)
        # clear existing Keyed Container
        self.fsr[path].clear()
        # Add newly merged lumiFSR
        self.fsr[path].add(newLumi)
        return SUCCESS
 
 
# =============================================================================
 
 

ProcessEventCountFSR()

def GaudiMP.pTools.FileRecordsAgent.ProcessEventCountFSR	(		self,
			path,
			ob
	)

Definition at line 434 of file pTools.py.

    def ProcessEventCountFSR(self, path, ob):
        self.log.debug("Event Count Input Addition")
        self.fsr[path].setInput(self.fsr[path].input() + ob.input())
 

ProcessTimeSpanFSR()

def GaudiMP.pTools.FileRecordsAgent.ProcessTimeSpanFSR	(		self,
			path,
			ob
	)

Definition at line 412 of file pTools.py.

    def ProcessTimeSpanFSR(self, path, ob):
        ob2 = self.fsr.retrieveObject(path)
        if ob.containedObjects().size():
            sz = ob.containedObjects().size()
            cob = ob2.containedObjects()[0]
            min = cob.earliest()
            max = cob.latest()
            for j in range(sz):
                cob = ob.containedObjects()[j]
                self.log.debug("Adding TimeSpanFSR")
                if cob.earliest() < min:
                    min = cob.earliest()
                if cob.latest() > max:
                    max = cob.latest()
                # this is annoying: it has to be rebuilt, without a key & added
                continue
            tsfsr = gbl.LHCb.TimeSpanFSR()
            tsfsr.setEarliest(min)
            tsfsr.setLatest(max)
            self.fsr[path].clear()
            self.fsr[path].add(tsfsr)
 

Rebuild()

def GaudiMP.pTools.FileRecordsAgent.Rebuild

(

self

)

Definition at line 345 of file pTools.py.

    def Rebuild(self):
        # objects is a list of (path, serializedObject) tuples
        for sourceNode, path, serialob in self.objectsIn:
            self.log.debug("Working with %s" % (path))
            ob = pickle.loads(serialob)
            if hasattr(ob, "update"):
                ob.update()
            if hasattr(ob, "numberOfObjects"):
                nCont = ob.numberOfObjects()
                self.log.debug(
                    "\t %s has containedObjects : %i" % (type(ob).__name__, nCont)
                )
            if sourceNode == 0:
                self.log.debug("Registering Object to : %s" % (path))
                self.fsr.registerObject(path, ob)
            else:
                self.log.debug("Merging Object to : %s" % (path))
                self.MergeFSRobject(sourceNode, path, ob)
        # As RecordStream has been split into Worker and Writer parts, the
        # count for output is wrong... fix that here, as every event received
        # by the writer is written (validation testing occurs on the worker)
 
        self.log.info("FSR Store Rebuilt.  Correcting EventCountFSR")
        if bool(self.fsr._idp):  # There might not be an FSR stream (Gauss)
            ecount = "/FileRecords/EventCountFSR"
            if self.fsr[ecount]:
                self.fsr[ecount].setOutput(self._gmpc.nIn)
                self.log.info(
                    "Event Counter Output set : %s : %i"
                    % (ecount, self.fsr[ecount].output())
                )
            # Do some reporting
            self.log.debug("FSR store reconstructed!")
            lst = self.fsr.getHistoNames()
            if lst:
                for l in lst:
                    ob = self.fsr.retrieveObject(l)
                    if hasattr(ob, "configureDirectAccess"):
                        ob.configureDirectAccess()
                    if hasattr(ob, "containedObjects"):
                        # if ob.numberOfObjects() :
                        self.log.debug(
                            "\t%s (cont. objects : %i)" % (l, ob.numberOfObjects())
                        )
                    else:
                        self.log.debug("\t%s" % (l))
        self.log.info("FSR Store fully rebuilt.")
        return SUCCESS
 

Receive()

def GaudiMP.pTools.FileRecordsAgent.Receive

(

self

)

Definition at line 325 of file pTools.py.

    def Receive(self):
        # Receive contents of all Workers FileRecords Transient Stores
        self.log.info("Receiving FSR store data...")
        nc = self._gmpc.nWorkers
        while nc > 0:
            objects = self.q.get()
            if objects == "END_FSR":
                nc -= 1
                continue
                if nc == 0:
                    break
            # but if it's regular objects...
            for o in objects:
                self.objectsIn.append(o)
        # Now sort it by which worker it came from
        # an object is : (nodeID, path, pickledObject)
        self.objectsIn.sort(cmp=self.localCmp)
        self.log.info("All FSR data received")
        return SUCCESS
 

SendFileRecords()

def GaudiMP.pTools.FileRecordsAgent.SendFileRecords

(

self

)

Definition at line 262 of file pTools.py.

    def SendFileRecords(self):
        # send the FileRecords data as part of finalisation
 
        # Take Care of FileRecords!
        # There are two main things to consider here
        # 1) The DataObjects in the FileRecords Transient Store
        # 2) The fact that they are Keyed Containers, containing other objects
        #
        # The Lead Worker, nodeID=0, sends everything in the FSR store, as
        #   a completeness guarantee,
        #
        # send in form ( nodeID, path, object)
        self.log.info("Sending FileRecords...")
        lst = self.fsr.getHistoNames()
 
        # Check Validity
        if not lst:
            self.log.info("No FileRecords Data to send to Writer.")
            self.q.put("END_FSR")
            return SUCCESS
 
        # no need to send the root node
        if "/FileRecords" in lst:
            lst.remove("/FileRecords")
 
        for l in lst:
            o = self.fsr.retrieveObject(l)
            if hasattr(o, "configureDirectAccess"):
                o.configureDirectAccess()
            # lead worker sends everything, as completeness guarantee
            if self._gmpc.nodeID == 0:
                self.objectsOut.append((0, l, pickle.dumps(o)))
            else:
                # only add the Event Counter
                # and non-Empty Keyed Containers (ignore empty ones)
                if l == "/FileRecords/EventCountFSR":
                    tup = (self._gmpc.nodeID, l, pickle.dumps(o))
                    self.objectsOut.append(tup)
                elif "KeyedContainer" in o.__class__.__name__:
                    # It's a Keyed Container
                    nObjects = o.numberOfObjects()
                    if nObjects:
                        self.log.debug(
                            "Keyed Container %s with %i objects" % (l, nObjects)
                        )
                        tup = (self._gmpc.nodeID, l, pickle.dumps(o))
                        self.objectsOut.append(tup)
                else:
                    self.log.info("Ignoring %s in FSR" % o.__class__.__name__)
 
        self.log.debug("Done with FSR store, just to send to Writer.")
 
        if self.objectsOut:
            self.log.debug("%i FSR objects to Writer" % (len(self.objectsOut)))
            for ob in self.objectsOut:
                self.log.debug("\t%s" % (ob[0]))
            self.q.put(self.objectsOut)
        else:
            self.log.info("Valid FSR Store, but no data to send to Writer")
        self.log.info("SendFSR complete")
        self.q.put("END_FSR")
        return SUCCESS
 

Member Data Documentation

_gmpc

GaudiMP.pTools.FileRecordsAgent._gmpc

private

Definition at line 242 of file pTools.py.

fsr

GaudiMP.pTools.FileRecordsAgent.fsr

Definition at line 243 of file pTools.py.

log

GaudiMP.pTools.FileRecordsAgent.log

Definition at line 245 of file pTools.py.

objectsIn

GaudiMP.pTools.FileRecordsAgent.objectsIn

Definition at line 246 of file pTools.py.

objectsOut

GaudiMP.pTools.FileRecordsAgent.objectsOut

Definition at line 247 of file pTools.py.

q

GaudiMP.pTools.FileRecordsAgent.q

Definition at line 244 of file pTools.py.

---

The documentation for this class was generated from the following file:

• GaudiMP/python/GaudiMP/pTools.py