| # Gaudi Framework, version v22r2 |

Home |  | Generated: Tue May 10 2011

Public Member Functions | Public Attributes | Private Attributes

# GaudiMP::pTools::FileRecordsAgent Class Reference

List of all members.

## Public Member Functions
def | __init__
def | localCmp
def | SendFileRecords
def | Receive
def | Rebuild
def | MergeFSRobject
def | ProcessTimeSpanFSR
def | ProcessEventCountFSR
def | MergeLumiFSR
## Public Attributes
| fsr
| q
| log
| objectsIn
| objectsOut
## Private Attributes
| _gmpc

## Detailed Description

Definition at line 203 of file pTools.py.

## Constructor & Destructor Documentation

def GaudiMP::pTools::FileRecordsAgent::__init__ | ( |  | self,
|  |  | gmpComponent
| ) |  |

Definition at line 204 of file pTools.py.

    00205                                        :
    00206         self._gmpc = gmpComponent
    00207         self.fsr = self._gmpc.fsr
    00208         self.q   = self._gmpc.fq
    00209         self.log = self._gmpc.log
    00210         self.objectsIn  = []   # used for collecting FSR store objects
    00211         self.objectsOut = []

## Member Function Documentation

def GaudiMP::pTools::FileRecordsAgent::localCmp | ( |  | self,
|  |  | tupA,
|  |  | tupB
| ) |  |

Definition at line 212 of file pTools.py.

    00213                                      :
    00214         # sort tuples by a particular element
    00215         # for the sort() method
    00216         ind  = 0
    00217         valA = tupA[ind]
    00218         valB = tupB[ind]
    00219         if   valA<valB : return -1
    00220         elif valA>valB : return  1
    00221         else           : return  0
    00222

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

Definition at line 346 of file pTools.py.

    00347                                                      :
    00348         # Merge Non-Empty Keyed Container from Worker>0
    00349         if path == '/FileRecords/TimeSpanFSR' :
    00350             # TimeSpanFSR is a straightforward case
    00351             self.ProcessTimeSpanFSR( path, ob )
    00352         elif path == '/FileRecords/EventCountFSR' :
    00353             # Event Counter is also easy
    00354             self.ProcessEventCountFSR( path, ob )
    00355         # now other cases may not be so easy...
    00356         elif "KeyedContainer" in ob.__class__.__name__ :
    00357             # Keyed Container of LumiFSRs : extract and re-register
    00358             # self.ProcessLumiFSR( path, ob )
    00359             if "LumiFSR" in ob.__class__.__name__ :
    00360                 self.MergeLumiFSR( path, ob )
    00361             else:
    00362                 self.log.info("Skipping Merge of Keyed Container %s for %s"\
    00363                               %(ob.__class__.__name__,path))

def GaudiMP::pTools::FileRecordsAgent::MergeLumiFSR | ( |  | self,
|  |  | path,
|  |  | keyedC
| ) |  |

Definition at line 381 of file pTools.py.

    00382                                            :
    00383         from ROOT import string
    00384         # Fetch the first lumi
    00385         keyedContainer = self.fsr.retrieveObject(path)
    00386         # The LumiFSR KeyedContainers only have one object
    00387         assert keyedContainer.numberOfObjects() == 1
    00388         l = keyedContainer.containedObject(0)
    00389         baseLumi = LumiFSR( l )
    00390         # Now deal with the argument Non-empty Keyed Container of LumiFSRs
    00391         nCont = keyedC.numberOfObjects()
    00392         for i in xrange(nCont) :
    00393             obj = keyedC.containedObject(i)
    00394             nextLumi = LumiFSR( obj )
    00395             baseLumi.merge( nextLumi )
    00396         # Now Rebuild and ReRegister
    00397         newLumi = gbl.LHCb.LumiFSR()
    00398         for r in baseLumi.runs :
    00399             newLumi.addRunNumber( r )
    00400         for f in baseLumi.files :
    00401             newLumi.addFileID( string(f) )
    00402         for k in baseLumi.keys :
    00403             increment, integral = baseLumi.info[k]
    00404             newLumi.addInfo(k, increment, integral)
    00405         # clear existing Keyed Container
    00406         self.fsr[path].clear()
    00407         # Add newly merged lumiFSR
    00408         self.fsr[path].add(newLumi)
    00409         return SUCCESS
    00410
    00411 # =============================================================================

def GaudiMP::pTools::FileRecordsAgent::ProcessEventCountFSR | ( |  | self,
|  |  | path,
|  |  | ob
| ) |  |

Definition at line 377 of file pTools.py.

    00378                                                :
    00379         self.log.debug('Event Count Input Addition')
    00380         self.fsr[path].setInput( self.fsr[path].input()+ob.input() )

def GaudiMP::pTools::FileRecordsAgent::ProcessTimeSpanFSR | ( |  | self,
|  |  | path,
|  |  | ob
| ) |  |

Definition at line 364 of file pTools.py.

    00365                                              :
    00366         if ob.containedObjects().size() :
    00367             sz = ob.containedObjects().size()
    00368             for j in xrange(sz) :
    00369                 cob = ob.containedObjects()[j]
    00370                 self.log.debug('Adding TimeSpanFSR')
    00371                 # this is annoying: it has to be rebuilt, without a key & added
    00372                 tsfsr = gbl.LHCb.TimeSpanFSR()
    00373                 tsfsr.setEarliest( cob.earliest() )
    00374                 tsfsr.setLatest(   cob.latest(  ) )
    00375                 self.fsr[path].add(tsfsr)
    00376                 continue

def GaudiMP::pTools::FileRecordsAgent::Rebuild | ( |  | self ) |

Definition at line 302 of file pTools.py.

    00303                         :
    00304         # objects is a list of (path, serializedObject) tuples
    00305         for sourceNode, path, serialob in self.objectsIn :
    00306             self.log.debug('Working with %s'%(path))
    00307             ob = pickle.loads(serialob)
    00308             if hasattr( ob, 'update' ) :
    00309                 ob.update()
    00310             if hasattr( ob, 'numberOfObjects' ) :
    00311                 nCont = ob.numberOfObjects()
    00312                 self.log.debug( '\t %s has containedObjects : %i'%(type(ob).__name__,  nCont) )
    00313             if sourceNode == 0 :
    00314                 self.log.debug('Registering Object to : %s'%(path))
    00315                 self.fsr.registerObject( path, ob )
    00316             else :
    00317                 self.log.debug('Merging Object to : %s'%(path))
    00318                 self.MergeFSRobject( sourceNode, path, ob )
    00319         # As RecordStream has been split into Worker and Writer parts, the
    00320         # count for output is wrong... fix that here, as every event received
    00321         # by the writer is written (validation testing occurs on the worker)
    00322
    00323         self.log.info('FSR Store Rebuilt.  Correcting EventCountFSR')
    00324         if bool( self.fsr._idp ) :   # There might not be an FSR stream (Gauss)
    00325             ecount  = '/FileRecords/EventCountFSR'
    00326             if self.fsr[ecount] :
    00327                 self.fsr[ecount].setOutput( self._gmpc.nIn )
    00328                 self.log.info( 'Event Counter Output set : %s : %i'\
    00329                                %(ecount, self.fsr[ecount].output()) )
    00330             # Do some reporting
    00331             self.log.debug('FSR store reconstructed!')
    00332             lst = self.fsr.getHistoNames()
    00333             if lst :
    00334                 for l in lst :
    00335                     ob = self.fsr.retrieveObject(l)
    00336                     if hasattr( ob, 'configureDirectAccess' ) :
    00337                         ob.configureDirectAccess()
    00338                     if hasattr( ob, 'containedObjects' ) :
    00339                         # if ob.numberOfObjects() :
    00340                         self.log.debug('\t%s (cont. objects : %i)'\
    00341                                        %(l, ob.numberOfObjects()))
    00342                     else :
    00343                         self.log.debug('\t%s'%(l))
    00344         self.log.info('FSR Store fully rebuilt.')
    00345         return SUCCESS

def GaudiMP::pTools::FileRecordsAgent::Receive | ( |  | self ) |

Definition at line 282 of file pTools.py.

    00283                         :
    00284         # Receive contents of all Workers FileRecords Transient Stores
    00285         self.log.info('Receiving FSR store data...')
    00286         nc = self._gmpc.nWorkers
    00287         while nc > 0 :
    00288             objects = self.q.get( )
    00289             if objects == 'END_FSR' :
    00290                 nc -= 1
    00291                 continue
    00292                 if nc==0 :
    00293                     break
    00294             # but if it's regular objects...
    00295             for o in objects :
    00296                 self.objectsIn.append(o)
    00297         # Now sort it by which worker it came from
    00298         # an object is : (nodeID, path, pickledObject)
    00299         self.objectsIn.sort(cmp=self.localCmp)
    00300         self.log.info('All FSR data received')
    00301         return SUCCESS

def GaudiMP::pTools::FileRecordsAgent::SendFileRecords | ( |  | self ) |

Definition at line 223 of file pTools.py.

    00224                                 :
    00225         # send the FileRecords data as part of finalisation
    00226
    00227         # Take Care of FileRecords!
    00228         # There are two main things to consider here
    00229         # 1) The DataObjects in the FileRecords Transient Store
    00230         # 2) The fact that they are Keyed Containers, containing other objects
    00231         #
    00232         # The Lead Worker, nodeID=0, sends everything in the FSR store, as
    00233         #   a completeness guarantee,
    00234         #
    00235         # send in form ( nodeID, path, object)
    00236         self.log.info('Sending FileRecords...')
    00237         lst      = self.fsr.getHistoNames()
    00238
    00239         # Check Validity
    00240         if not lst :
    00241             self.log.info('No FileRecords Data to send to Writer.')
    00242             self.q.put( 'END_FSR' )
    00243             return SUCCESS
    00244
    00245         # no need to send the root node
    00246         if '/FileRecords' in lst : lst.remove('/FileRecords')
    00247
    00248         for l in lst :
    00249             o = self.fsr.retrieveObject( l )
    00250             if hasattr(o, "configureDirectAccess") :
    00251                 o.configureDirectAccess()
    00252             # lead worker sends everything, as completeness guarantee
    00253             if self._gmpc.nodeID == 0 :
    00254                 self.objectsOut.append( (0, l, pickle.dumps(o)) )
    00255             else :
    00256                 # only add the Event Counter
    00257                 # and non-Empty Keyed Containers (ignore empty ones)
    00258                 if l == '/FileRecords/EventCountFSR' :
    00259                     tup = (self._gmpc.nodeID, l, pickle.dumps(o))
    00260                     self.objectsOut.append( tup )
    00261                 else :
    00262                     # It's a Keyed Container
    00263                     assert "KeyedContainer" in o.__class__.__name__
    00264                     nObjects = o.numberOfObjects()
    00265                     if nObjects :
    00266                         self.log.debug("Keyed Container %s with %i objects"\
    00267                                        %(l, nObjects))
    00268                         tup = (self._gmpc.nodeID, l, pickle.dumps(o))
    00269                         self.objectsOut.append( tup )
    00270         self.log.debug('Done with FSR store, just to send to Writer.')
    00271
    00272         if self.objectsOut :
    00273             self.log.debug('%i FSR objects to Writer'%(len(self.objectsOut)))
    00274             for ob in self.objectsOut :
    00275                 self.log.debug('\t%s'%(ob[0]))
    00276             self.q.put( self.objectsOut )
    00277         else :
    00278             self.log.info('Valid FSR Store, but no data to send to Writer')
    00279         self.log.info('SendFSR complete')
    00280         self.q.put( 'END_FSR' )
    00281         return SUCCESS

## Member Data Documentation

GaudiMP::pTools::FileRecordsAgent::_gmpc`[private]`

Definition at line 204 of file pTools.py.

GaudiMP::pTools::FileRecordsAgent::fsr

Definition at line 204 of file pTools.py.

GaudiMP::pTools::FileRecordsAgent::log

Definition at line 204 of file pTools.py.

GaudiMP::pTools::FileRecordsAgent::objectsIn

Definition at line 204 of file pTools.py.

GaudiMP::pTools::FileRecordsAgent::objectsOut

Definition at line 204 of file pTools.py.

GaudiMP::pTools::FileRecordsAgent::q

Definition at line 204 of file pTools.py.

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

- /afs/.cern.ch/sw/Gaudi/releases/GAUDI/GAUDI_v22r2/GaudiMP/python/GaudiMP/pTools.py

All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Defines