The Gaudi Framework  v41r0 (bdbe7744)
Loading...
Searching...
No Matches
GaudiMP.pTools.FileRecordsAgent Class Reference

Public Member Functions

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

Public Attributes

 fsr = self._gmpc.fsr
 q = self._gmpc.fq
 log = self._gmpc.log
list objectsIn = []
list objectsOut = []

Protected Attributes

 _gmpc = gmpComponent

Detailed Description

Definition at line 236 of file pTools.py.

Constructor & Destructor Documentation

◆ __init__()

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

Definition at line 237 of file pTools.py.

237 def __init__(self, gmpComponent):
238 self._gmpc = gmpComponent
239 self.fsr = self._gmpc.fsr
240 self.q = self._gmpc.fq
241 self.log = self._gmpc.log
242 self.objectsIn = [] # used for collecting FSR store objects
243 self.objectsOut = []
244

Member Function Documentation

◆ localCmp()

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

Definition at line 245 of file pTools.py.

245 def localCmp(self, tupA, tupB):
246 # sort tuples by a particular element
247 # for the sort() method
248 ind = 0
249 valA = tupA[ind]
250 valB = tupB[ind]
251 if valA < valB:
252 return -1
253 elif valA > valB:
254 return 1
255 else:
256 return 0
257

◆ MergeFSRobject()

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

Definition at line 385 of file pTools.py.

385 def MergeFSRobject(self, sourceNode, path, ob):
386 # Merge Non-Empty Keyed Container from Worker>0
387 if path == "/FileRecords/TimeSpanFSR":
388 # TimeSpanFSR is a straightforward case
389 self.ProcessTimeSpanFSR(path, ob)
390 elif path == "/FileRecords/EventCountFSR":
391 # Event Counter is also easy
392 self.ProcessEventCountFSR(path, ob)
393 # now other cases may not be so easy...
394 elif (
395 "KeyedContainer" in ob.__class__.__name__
396 and "LumiFSR" in ob.__class__.__name__
397 ):
398 # Keyed Container of LumiFSRs : extract and re-register
399 self.MergeLumiFSR(path, ob)
400 else:
401 self.log.info(f"Skipping Merge of {ob.__class__.__name__} at {path}")
402

◆ MergeLumiFSR()

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

Definition at line 429 of file pTools.py.

429 def MergeLumiFSR(self, path, keyedC):
430 from ROOT import string
431
432 # Fetch the first lumi
433 keyedContainer = self.fsr.retrieveObject(path)
434 # The LumiFSR KeyedContainers only have one object
435 assert keyedContainer.numberOfObjects() == 1
436 l = keyedContainer.containedObject(0)
437 baseLumi = LumiFSR(l)
438 # Now deal with the argument Non-empty Keyed Container of LumiFSRs
439 nCont = keyedC.numberOfObjects()
440 for i in range(nCont):
441 obj = keyedC.containedObject(i)
442 nextLumi = LumiFSR(obj)
443 baseLumi.merge(nextLumi)
444 # Now Rebuild and ReRegister
445 newLumi = gbl.LHCb.LumiFSR()
446 for r in baseLumi.runs:
447 newLumi.addRunNumber(r)
448 for f in baseLumi.files:
449 newLumi.addFileID(string(f))
450 for k in baseLumi.keys:
451 increment, integral = baseLumi.info[k]
452 newLumi.addInfo(k, increment, integral)
453 # clear existing Keyed Container
454 self.fsr[path].clear()
455 # Add newly merged lumiFSR
456 self.fsr[path].add(newLumi)
457 return SUCCESS
458
459
460# =============================================================================
461
462
string
STL class.

◆ ProcessEventCountFSR()

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

Definition at line 425 of file pTools.py.

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

◆ ProcessTimeSpanFSR()

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

Definition at line 403 of file pTools.py.

403 def ProcessTimeSpanFSR(self, path, ob):
404 ob2 = self.fsr.retrieveObject(path)
405 if ob.containedObjects().size():
406 sz = ob.containedObjects().size()
407 cob = ob2.containedObjects()[0]
408 min = cob.earliest()
409 max = cob.latest()
410 for j in range(sz):
411 cob = ob.containedObjects()[j]
412 self.log.debug("Adding TimeSpanFSR")
413 if cob.earliest() < min:
414 min = cob.earliest()
415 if cob.latest() > max:
416 max = cob.latest()
417 # this is annoying: it has to be rebuilt, without a key & added
418 continue
419 tsfsr = gbl.LHCb.TimeSpanFSR()
420 tsfsr.setEarliest(min)
421 tsfsr.setLatest(max)
422 self.fsr[path].clear()
423 self.fsr[path].add(tsfsr)
424

◆ Rebuild()

GaudiMP.pTools.FileRecordsAgent.Rebuild ( self)

Definition at line 339 of file pTools.py.

339 def Rebuild(self):
340 # objects is a list of (path, serializedObject) tuples
341 for sourceNode, path, serialob in self.objectsIn:
342 self.log.debug(f"Working with {path}")
343 ob = pickle.loads(serialob)
344 if hasattr(ob, "update"):
345 ob.update()
346 if hasattr(ob, "numberOfObjects"):
347 nCont = ob.numberOfObjects()
348 self.log.debug(f"\t {type(ob).__name__} has containedObjects : {nCont}")
349 if sourceNode == 0:
350 self.log.debug(f"Registering Object to : {path}")
351 self.fsr.registerObject(path, ob)
352 else:
353 self.log.debug(f"Merging Object to : {path}")
354 self.MergeFSRobject(sourceNode, path, ob)
355 # As RecordStream has been split into Worker and Writer parts, the
356 # count for output is wrong... fix that here, as every event received
357 # by the writer is written (validation testing occurs on the worker)
358
359 self.log.info("FSR Store Rebuilt. Correcting EventCountFSR")
360 if bool(self.fsr._idp): # There might not be an FSR stream (Gauss)
361 ecount = "/FileRecords/EventCountFSR"
362 if self.fsr[ecount]:
363 self.fsr[ecount].setOutput(self._gmpc.nIn)
364 self.log.info(
365 f"Event Counter Output set : {ecount} : {self.fsr[ecount].output()}"
366 )
367 # Do some reporting
368 self.log.debug("FSR store reconstructed!")
369 lst = self.fsr.getHistoNames()
370 if lst:
371 for l in lst:
372 ob = self.fsr.retrieveObject(l)
373 if hasattr(ob, "configureDirectAccess"):
374 ob.configureDirectAccess()
375 if hasattr(ob, "containedObjects"):
376 # if ob.numberOfObjects() :
377 self.log.debug(
378 f"\t{l} (cont. objects : {ob.numberOfObjects()})"
379 )
380 else:
381 self.log.debug(f"\t{l}")
382 self.log.info("FSR Store fully rebuilt.")
383 return SUCCESS
384

◆ Receive()

GaudiMP.pTools.FileRecordsAgent.Receive ( self)

Definition at line 319 of file pTools.py.

319 def Receive(self):
320 # Receive contents of all Workers FileRecords Transient Stores
321 self.log.info("Receiving FSR store data...")
322 nc = self._gmpc.nWorkers
323 while nc > 0:
324 objects = self.q.get()
325 if objects == "END_FSR":
326 nc -= 1
327 continue
328 if nc == 0:
329 break
330 # but if it's regular objects...
331 for o in objects:
332 self.objectsIn.append(o)
333 # Now sort it by which worker it came from
334 # an object is : (nodeID, path, pickledObject)
335 self.objectsIn.sort(cmp=self.localCmp)
336 self.log.info("All FSR data received")
337 return SUCCESS
338

◆ SendFileRecords()

GaudiMP.pTools.FileRecordsAgent.SendFileRecords ( self)

Definition at line 258 of file pTools.py.

258 def SendFileRecords(self):
259 # send the FileRecords data as part of finalisation
260
261 # Take Care of FileRecords!
262 # There are two main things to consider here
263 # 1) The DataObjects in the FileRecords Transient Store
264 # 2) The fact that they are Keyed Containers, containing other objects
265 #
266 # The Lead Worker, nodeID=0, sends everything in the FSR store, as
267 # a completeness guarantee,
268 #
269 # send in form ( nodeID, path, object)
270 self.log.info("Sending FileRecords...")
271 lst = self.fsr.getHistoNames()
272
273 # Check Validity
274 if not lst:
275 self.log.info("No FileRecords Data to send to Writer.")
276 self.q.put("END_FSR")
277 return SUCCESS
278
279 # no need to send the root node
280 if "/FileRecords" in lst:
281 lst.remove("/FileRecords")
282
283 for l in lst:
284 o = self.fsr.retrieveObject(l)
285 if hasattr(o, "configureDirectAccess"):
286 o.configureDirectAccess()
287 # lead worker sends everything, as completeness guarantee
288 if self._gmpc.nodeID == 0:
289 self.objectsOut.append((0, l, pickle.dumps(o)))
290 else:
291 # only add the Event Counter
292 # and non-Empty Keyed Containers (ignore empty ones)
293 if l == "/FileRecords/EventCountFSR":
294 tup = (self._gmpc.nodeID, l, pickle.dumps(o))
295 self.objectsOut.append(tup)
296 elif "KeyedContainer" in o.__class__.__name__:
297 # It's a Keyed Container
298 nObjects = o.numberOfObjects()
299 if nObjects:
300 self.log.debug(f"Keyed Container {l} with {nObjects} objects")
301 tup = (self._gmpc.nodeID, l, pickle.dumps(o))
302 self.objectsOut.append(tup)
303 else:
304 self.log.info(f"Ignoring {o.__class__.__name__} in FSR")
305
306 self.log.debug("Done with FSR store, just to send to Writer.")
307
308 if self.objectsOut:
309 self.log.debug(f"{len(self.objectsOut)} FSR objects to Writer")
310 for ob in self.objectsOut:
311 self.log.debug(f"\t{ob[0]}")
312 self.q.put(self.objectsOut)
313 else:
314 self.log.info("Valid FSR Store, but no data to send to Writer")
315 self.log.info("SendFSR complete")
316 self.q.put("END_FSR")
317 return SUCCESS
318

Member Data Documentation

◆ _gmpc

GaudiMP.pTools.FileRecordsAgent._gmpc = gmpComponent
protected

Definition at line 238 of file pTools.py.

◆ fsr

GaudiMP.pTools.FileRecordsAgent.fsr = self._gmpc.fsr

Definition at line 239 of file pTools.py.

◆ log

GaudiMP.pTools.FileRecordsAgent.log = self._gmpc.log

Definition at line 241 of file pTools.py.

◆ objectsIn

list GaudiMP.pTools.FileRecordsAgent.objectsIn = []

Definition at line 242 of file pTools.py.

◆ objectsOut

GaudiMP.pTools.FileRecordsAgent.objectsOut = []

Definition at line 243 of file pTools.py.

◆ q

GaudiMP.pTools.FileRecordsAgent.q = self._gmpc.fq

Definition at line 240 of file pTools.py.

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