| The Gaudi Framework v32r1 (f65d50dc)

Public Member Functions | Public Attributes | Private Attributes | List of all members
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 209 of file pTools.py.

## Constructor & Destructor Documentation

## ◆ __init__()

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

Definition at line 210 of file pTools.py.

210 def __init__(self, gmpComponent):
211 self._gmpc = gmpComponent
212 self.fsr = self._gmpc.fsr
213 self.q = self._gmpc.fq
214 self.log = self._gmpc.log
215 self.objectsIn = [] # used for collecting FSR store objects
216 self.objectsOut = []
217

## Member Function Documentation

## ◆ localCmp()

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

Definition at line 218 of file pTools.py.

218 def localCmp(self, tupA, tupB):
219 # sort tuples by a particular element
220 # for the sort() method
221 ind = 0
222 valA = tupA[ind]
223 valB = tupB[ind]
224 if valA < valB:
225 return -1
226 elif valA > valB:
227 return 1
228 else:
229 return 0
230

## ◆ MergeFSRobject()

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

Definition at line 356 of file pTools.py.

356 def MergeFSRobject(self, sourceNode, path, ob):
357 # Merge Non-Empty Keyed Container from Worker>0
358 if path == '/FileRecords/TimeSpanFSR':
359 # TimeSpanFSR is a straightforward case
360 self.ProcessTimeSpanFSR(path, ob)
361 elif path == '/FileRecords/EventCountFSR':
362 # Event Counter is also easy
363 self.ProcessEventCountFSR(path, ob)
364 # now other cases may not be so easy...
365 elif "KeyedContainer" in ob.__class__.__name__:
366 # Keyed Container of LumiFSRs : extract and re-register
367 # self.ProcessLumiFSR( path, ob )
368 if "LumiFSR" in ob.__class__.__name__:
369 self.MergeLumiFSR(path, ob)
370 else:
371 self.log.info("Skipping Merge of Keyed Container %s for %s" %
372 (ob.__class__.__name__, path))
373

## ◆ MergeLumiFSR()

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

Definition at line 400 of file pTools.py.

400 def MergeLumiFSR(self, path, keyedC):
401 from ROOT import string
402 # Fetch the first lumi
403 keyedContainer = self.fsr.retrieveObject(path)
404 # The LumiFSR KeyedContainers only have one object
405 assert keyedContainer.numberOfObjects() == 1
406 l = keyedContainer.containedObject(0)
407 baseLumi = LumiFSR(l)
408 # Now deal with the argument Non-empty Keyed Container of LumiFSRs
409 nCont = keyedC.numberOfObjects()
410 for i in xrange(nCont):
411 obj = keyedC.containedObject(i)
412 nextLumi = LumiFSR(obj)
413 baseLumi.merge(nextLumi)
414 # Now Rebuild and ReRegister
415 newLumi = gbl.LHCb.LumiFSR()
416 for r in baseLumi.runs:
417 newLumi.addRunNumber(r)
418 for f in baseLumi.files:
419 newLumi.addFileID(string(f))
420 for k in baseLumi.keys:
421 increment, integral = baseLumi.info[k]
422 newLumi.addInfo(k, increment, integral)
423 # clear existing Keyed Container
424 self.fsr[path].clear()
425 # Add newly merged lumiFSR
426 self.fsr[path].add(newLumi)
427 return SUCCESS
428
429
430 # =============================================================================
431
432
GaudiPython.Persistency.add
def add(instance)
Definition: Persistency.py:39
compareOutputFiles.LumiFSR
def LumiFSR(lumi)
Definition: compareOutputFiles.py:322

## ◆ ProcessEventCountFSR()

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

Definition at line 396 of file pTools.py.

396 def ProcessEventCountFSR(self, path, ob):
397 self.log.debug('Event Count Input Addition')
398 self.fsr[path].setInput(self.fsr[path].input() + ob.input())
399

## ◆ ProcessTimeSpanFSR()

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

Definition at line 374 of file pTools.py.

374 def ProcessTimeSpanFSR(self, path, ob):
375 ob2 = self.fsr.retrieveObject(path)
376 if ob.containedObjects().size():
377 sz = ob.containedObjects().size()
378 cob = ob2.containedObjects()[0]
379 min = cob.earliest()
380 max = cob.latest()
381 for j in xrange(sz):
382 cob = ob.containedObjects()[j]
383 self.log.debug('Adding TimeSpanFSR')
384 if cob.earliest() < min:
385 min = cob.earliest()
386 if cob.latest() > max:
387 max = cob.latest()
388 # this is annoying: it has to be rebuilt, without a key & added
389 continue
390 tsfsr = gbl.LHCb.TimeSpanFSR()
391 tsfsr.setEarliest(min)
392 tsfsr.setLatest(max)
393 self.fsr[path].clear()
394 self.fsr[path].add(tsfsr)
395
details::size
constexpr auto size(const T &, Args &&...) noexcept
Definition: AnyDataWrapper.h:14
GaudiPython.Persistency.add
def add(instance)
Definition: Persistency.py:39

## ◆ Rebuild()

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

Definition at line 311 of file pTools.py.

311 def Rebuild(self):
312 # objects is a list of (path, serializedObject) tuples
313 for sourceNode, path, serialob in self.objectsIn:
314 self.log.debug('Working with %s' % (path))
315 ob = pickle.loads(serialob)
316 if hasattr(ob, 'update'):
317 ob.update()
318 if hasattr(ob, 'numberOfObjects'):
319 nCont = ob.numberOfObjects()
320 self.log.debug('\t %s has containedObjects : %i' %
321 (type(ob).__name__, nCont))
322 if sourceNode == 0:
323 self.log.debug('Registering Object to : %s' % (path))
324 self.fsr.registerObject(path, ob)
325 else:
326 self.log.debug('Merging Object to : %s' % (path))
327 self.MergeFSRobject(sourceNode, path, ob)
328 # As RecordStream has been split into Worker and Writer parts, the
329 # count for output is wrong... fix that here, as every event received
330 # by the writer is written (validation testing occurs on the worker)
331
332 self.log.info('FSR Store Rebuilt. Correcting EventCountFSR')
333 if bool(self.fsr._idp): # There might not be an FSR stream (Gauss)
334 ecount = '/FileRecords/EventCountFSR'
335 if self.fsr[ecount]:
336 self.fsr[ecount].setOutput(self._gmpc.nIn)
337 self.log.info('Event Counter Output set : %s : %i' %
338 (ecount, self.fsr[ecount].output()))
339 # Do some reporting
340 self.log.debug('FSR store reconstructed!')
341 lst = self.fsr.getHistoNames()
342 if lst:
343 for l in lst:
344 ob = self.fsr.retrieveObject(l)
345 if hasattr(ob, 'configureDirectAccess'):
346 ob.configureDirectAccess()
347 if hasattr(ob, 'containedObjects'):
348 # if ob.numberOfObjects() :
349 self.log.debug('\t%s (cont. objects : %i)' %
350 (l, ob.numberOfObjects()))
351 else:
352 self.log.debug('\t%s' % (l))
353 self.log.info('FSR Store fully rebuilt.')
354 return SUCCESS
355
gaudirun.output
output
Definition: gaudirun.py:484
gaudirun.type
type
Definition: gaudirun.py:142

## ◆ Receive()

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

Definition at line 291 of file pTools.py.

291 def Receive(self):
292 # Receive contents of all Workers FileRecords Transient Stores
293 self.log.info('Receiving FSR store data...')
294 nc = self._gmpc.nWorkers
295 while nc > 0:
296 objects = self.q.get()
297 if objects == 'END_FSR':
298 nc -= 1
299 continue
300 if nc == 0:
301 break
302 # but if it's regular objects...
303 for o in objects:
304 self.objectsIn.append(o)
305 # Now sort it by which worker it came from
306 # an object is : (nodeID, path, pickledObject)
307 self.objectsIn.sort(cmp=self.localCmp)
308 self.log.info('All FSR data received')
309 return SUCCESS
310
Gaudi::Functional::details::get
auto get(const Handle &handle, const Algo &, const EventContext &) -> decltype(details::deref(handle.get()))
Definition: FunctionalDetails.h:332

## ◆ SendFileRecords()

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

Definition at line 231 of file pTools.py.

231 def SendFileRecords(self):
232 # send the FileRecords data as part of finalisation
233
234 # Take Care of FileRecords!
235 # There are two main things to consider here
236 # 1) The DataObjects in the FileRecords Transient Store
237 # 2) The fact that they are Keyed Containers, containing other objects
238 #
239 # The Lead Worker, nodeID=0, sends everything in the FSR store, as
240 # a completeness guarantee,
241 #
242 # send in form ( nodeID, path, object)
243 self.log.info('Sending FileRecords...')
244 lst = self.fsr.getHistoNames()
245
246 # Check Validity
247 if not lst:
248 self.log.info('No FileRecords Data to send to Writer.')
249 self.q.put('END_FSR')
250 return SUCCESS
251
252 # no need to send the root node
253 if '/FileRecords' in lst:
254 lst.remove('/FileRecords')
255
256 for l in lst:
257 o = self.fsr.retrieveObject(l)
258 if hasattr(o, "configureDirectAccess"):
259 o.configureDirectAccess()
260 # lead worker sends everything, as completeness guarantee
261 if self._gmpc.nodeID == 0:
262 self.objectsOut.append((0, l, pickle.dumps(o)))
263 else:
264 # only add the Event Counter
265 # and non-Empty Keyed Containers (ignore empty ones)
266 if l == '/FileRecords/EventCountFSR':
267 tup = (self._gmpc.nodeID, l, pickle.dumps(o))
268 self.objectsOut.append(tup)
269 else:
270 # It's a Keyed Container
271 assert "KeyedContainer" in o.__class__.__name__
272 nObjects = o.numberOfObjects()
273 if nObjects:
274 self.log.debug("Keyed Container %s with %i objects" %
275 (l, nObjects))
276 tup = (self._gmpc.nodeID, l, pickle.dumps(o))
277 self.objectsOut.append(tup)
278 self.log.debug('Done with FSR store, just to send to Writer.')
279
280 if self.objectsOut:
281 self.log.debug('%i FSR objects to Writer' % (len(self.objectsOut)))
282 for ob in self.objectsOut:
283 self.log.debug('\t%s' % (ob[0]))
284 self.q.put(self.objectsOut)
285 else:
286 self.log.info('Valid FSR Store, but no data to send to Writer')
287 self.log.info('SendFSR complete')
288 self.q.put('END_FSR')
289 return SUCCESS
290
Gaudi::Functional::details::put
Out1 * put(const DataObjectHandle< Out1 > &out_handle, Out2 &&out)
Definition: FunctionalDetails.h:121

## Member Data Documentation

## ◆ _gmpc

GaudiMP.pTools.FileRecordsAgent._gmpc
| private

Definition at line 211 of file pTools.py.

## ◆ fsr

GaudiMP.pTools.FileRecordsAgent.fsr

Definition at line 212 of file pTools.py.

## ◆ log

GaudiMP.pTools.FileRecordsAgent.log

Definition at line 214 of file pTools.py.

## ◆ objectsIn

GaudiMP.pTools.FileRecordsAgent.objectsIn

Definition at line 215 of file pTools.py.

## ◆ objectsOut

GaudiMP.pTools.FileRecordsAgent.objectsOut

Definition at line 216 of file pTools.py.

## ◆ q

GaudiMP.pTools.FileRecordsAgent.q

Definition at line 213 of file pTools.py.

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

- GaudiMP/python/GaudiMP/pTools.py