| The Gaudi Framework v32r1 (f65d50dc)

Public Member Functions | Public Attributes | List of all members
GaudiMP.GMPBase.Worker Class Reference
Inheritance diagram for GaudiMP.GMPBase.Worker:
[legend]
Collaboration diagram for GaudiMP.GMPBase.Worker:
[legend]

## Public Member Functions
def | __init__ (self, workerID, queues, events, params, subworkers)
def | processConfiguration (self)
def | Engine (self)
def | getCheckAlgs (self)
def | checkExecutedPassed (self, algName)
def | isEventPassed (self)
Public Member Functions inherited from GaudiMP.GMPBase.GMPComponent
def | __init__ (self, nodeType, nodeID, queues, events, params, subworkers)
def | Start (self)
def | SetupGaudiPython (self)
def | StartGaudiPython (self)
def | LoadTES (self, tbufferfile)
def | getEventNumber (self)
def | IdentifyWriters (self)
def | dumpHistograms (self)
def | Initialize (self)
def | Finalize (self)
def | Report (self)

## Public Attributes
| writerDict
| vetoAlgs
| eventOutput
| app
| filerecordsAgent
| nIn
| firstEvTime
| currentEvent
| tTime
Public Attributes inherited from GaudiMP.GMPBase.GMPComponent
| nodeType
| finalEvent
| lastEvent
| log
| subworkers
| nodeID
| msgFormat
| currentEvent
| queues
| num
| app
| evcom
| evcoms
| hq
| fq
| nIn
| nOut
| stat
| iTime
| rTime
| fTime
| firstEvTime
| tTime
| proc
| a
| evt
| hvt
| fsr
| inc
| pers
| ts
| TS
| cntr

## Detailed Description

Definition at line 976 of file GMPBase.py.

## Constructor & Destructor Documentation

## ◆ __init__()

def GaudiMP.GMPBase.Worker.__init__ | ( |  | self,
|  |  | workerID,
|  |  | queues,
|  |  | events,
|  |  | params,
|  |  | subworkers
| ) |  |

Definition at line 977 of file GMPBase.py.

977 def __init__(self, workerID, queues, events, params, subworkers):
978 GMPComponent.__init__(self, 'Worker', workerID, queues, events, params,
979 subworkers)
980 # Identify the writer streams
981 self.writerDict = self.IdentifyWriters()
982 # Identify the accept/veto checks for each event
983 self.acceptAlgs, self.requireAlgs, self.vetoAlgs = self.getCheckAlgs()
984 self.log.name = "Worker-%i" % (self.nodeID)
985 self.log.info("Worker-%i Created OK" % (self.nodeID))
986 self.eventOutput = True
987

## Member Function Documentation

## ◆ checkExecutedPassed()

def GaudiMP.GMPBase.Worker.checkExecutedPassed | ( |  | self,
|  |  | algName
| ) |  |

Definition at line 1167 of file GMPBase.py.

1167 def checkExecutedPassed(self, algName):
1168 if self.a.algorithm(algName)._ialg.isExecuted()\
1169 and self.a.algorithm(algName)._ialg.filterPassed():
1170 return True
1171 else:
1172 return False
1173

## ◆ Engine()

def GaudiMP.GMPBase.Worker.Engine | ( |  | self | ) |

Reimplemented from GaudiMP.GMPBase.GMPComponent.

Definition at line 1029 of file GMPBase.py.

1029 def Engine(self):
1030
1031 # rename process
1032 import os
1033 import ctypes
1034 libc = ctypes.CDLL('libc.so.6')
1035 name = str(self.nodeType) + str(self.nodeID) + '\0'
1036 libc.prctl(15, name, 0, 0, 0)
1037
1038 startEngine = time.time()
1039 self.log.info("Worker %i starting Engine" % (self.nodeID))
1040 self.Initialize()
1041 self.filerecordsAgent = FileRecordsAgent(self)
1042
1043 # populate the TESSerializer itemlist
1044 self.log.info(
1045 'EVT WRITERS ON WORKER : %i' % (len(self.writerDict['events'])))
1046
1047 nEventWriters = len(self.writerDict["events"])
1048 if nEventWriters:
1049 itemList = set()
1050 optItemList = set()
1051 for m in self.writerDict["events"]:
1052 for item in m.ItemList:
1053 hsh = item.find('#')
1054 if hsh != -1:
1055 item = item[:hsh]
1056 itemList.add(item)
1057 for item in m.OptItemList:
1058 hsh = item.find('#')
1059 if hsh != -1:
1060 item = item[:hsh]
1061 optItemList.add(item)
1062 # If an item is mandatory and optional, keep it only in the optional list
1063 itemList -= optItemList
1064 for item in sorted(itemList):
1065 self.log.info(' adding ItemList Item to ts : %s' % (item))
1066 self.ts.addItem(item)
1067 for item in sorted(optItemList):
1068 self.log.info(' adding Optional Item to ts : %s' % (item))
1069 self.ts.addOptItem(item)
1070 else:
1071 self.log.info('There is no Event Output for this app')
1072 self.eventOutput = False
1073
1074 # Begin processing
1075 Go = True
1076 while Go:
1077 packet = self.evcom.receive()
1078 if packet:
1079 pass
1080 else:
1081 continue
1082 if packet == 'FINISHED':
1083 break
1084 evtNumber, tbin = packet # unpack
1085 if self.cntr != None:
1086 self.cntr.setEventCounter(evtNumber)
1087
1088 # subworkers are forked before the first event is processed
1089 # reader-thread for ConDB must be closed and reopened in each subworker
1090 # this is done by disconnect()
1091 if self.nIn == 0:
1092
1093 self.log.info("Fork new subworkers and disconnect from CondDB")
1094 condDB = self.a.service('CondDBCnvSvc', gbl.ICondDBReader)
1095 condDB.disconnect()
1096
1097 # Fork subworkers and share services
1098 for k in self.subworkers:
1099 k.SetServices(self.a, self.evt, self.hvt, self.fsr,
1100 self.inc, self.pers, self.ts, self.cntr)
1101 k.Start()
1102 self.a.addAlgorithm(CollectHistograms(self))
1103 self.nIn += 1
1104 self.TS.Load(tbin)
1105
1106 t = time.time()
1107 sc = self.a.executeEvent()
1108 if self.nIn == 1:
1109 self.firstEvTime = time.time() - t
1110 else:
1111 self.rTime += (time.time() - t)
1112 if sc.isSuccess():
1113 pass
1114 else:
1115 self.log.warning('Did not Execute Event')
1116 self.evt.clearStore()
1117 continue
1118 if self.isEventPassed():
1119 pass
1120 else:
1121 self.log.warning('Event did not pass : %i' % (evtNumber))
1122 self.evt.clearStore()
1123 continue
1124 if self.eventOutput:
1125 # It may be the case of generating Event Tags; hence
1126 # no event output
1127 self.currentEvent = self.getEventNumber()
1128 tb = self.TS.Dump()
1129 self.evcom.send((self.currentEvent, tb))
1130 self.nOut += 1
1131 self.inc.fireIncident(gbl.Incident('Worker', 'EndEvent'))
1132 self.eventLoopSyncer.set()
1133 self.evt.clearStore()
1134 self.log.info('Setting <Last> Event')
1135 self.lastEvent.set()
1136
1137 self.evcom.finalize()
1138 self.log.info('Worker-%i Finished Processing Events' % (self.nodeID))
1139 # Now send the FileRecords and stop/finalize the appMgr
1140 self.filerecordsAgent.SendFileRecords()
1141 self.Finalize()
1142 self.tTime = time.time() - startEngine
1143 self.Report()
1144
1145 for k in self.subworkers:
1146 self.log.info('Join subworkers')
1147 k.proc.join()
1148

## ◆ getCheckAlgs()

def GaudiMP.GMPBase.Worker.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

Definition at line 1149 of file GMPBase.py.

1149 def getCheckAlgs(self):
1150 '''
1151 For some output writers, a check is performed to see if the event has
1152 executed certain algorithms.
1153 These reside in the AcceptAlgs property for those writers
1154 '''
1155 acc = []
1156 req = []
1157 vet = []
1158 for m in self.writerDict["events"]:
1159 if hasattr(m.w, 'AcceptAlgs'):
1160 acc += m.w.AcceptAlgs
1161 if hasattr(m.w, 'RequireAlgs'):
1162 req += m.w.RequireAlgs
1163 if hasattr(m.w, 'VetoAlgs'):
1164 vet += m.w.VetoAlgs
1165 return (acc, req, vet)
1166

## ◆ isEventPassed()

def GaudiMP.GMPBase.Worker.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

Definition at line 1174 of file GMPBase.py.

1174 def isEventPassed(self):
1175 '''
1176 Check the algorithm status for an event.
1177 Depending on output writer settings, the event
1178 may be declined based on various criteria.
1179 This is a transcript of the check that occurs in GaudiSvc::OutputStream
1180 '''
1181 passed = False
1182
1183 self.log.debug('self.acceptAlgs is %s' % (str(self.acceptAlgs)))
1184 if self.acceptAlgs:
1185 for name in self.acceptAlgs:
1186 if self.checkExecutedPassed(name):
1187 passed = True
1188 break
1189 else:
1190 passed = True
1191
1192 self.log.debug('self.requireAlgs is %s' % (str(self.requireAlgs)))
1193 for name in self.requireAlgs:
1194 if self.checkExecutedPassed(name):
1195 pass
1196 else:
1197 self.log.info('Evt declined (requireAlgs) : %s' % (name))
1198 passed = False
1199
1200 self.log.debug('self.vetoAlgs is %s' % (str(self.vetoAlgs)))
1201 for name in self.vetoAlgs:
1202 if self.checkExecutedPassed(name):
1203 pass
1204 else:
1205 self.log.info('Evt declined : (vetoAlgs) : %s' % (name))
1206 passed = False
1207 return passed
1208
1209
1210 # =============================================================================
1211
1212

## ◆ processConfiguration()

def GaudiMP.GMPBase.Worker.processConfiguration | ( |  | self | ) |

Reimplemented from GaudiMP.GMPBase.GMPComponent.

Definition at line 988 of file GMPBase.py.

988 def processConfiguration(self):
989
990 # Worker :
991 # No input
992 # No output
993 # No Histos
994 self.config['EventSelector'].Input = []
995 self.config['ApplicationMgr'].OutStream = []
996 if "HistogramPersistencySvc" in self.config.keys():
997 self.config['HistogramPersistencySvc'].OutputFile = ''
998 formatHead = '[Worker-%i]' % (self.nodeID)
999 self.config['MessageSvc'].Format = '%-13s ' % formatHead + \
1000 self.msgFormat
1001
1002 for key, lst in self.writerDict.iteritems():
1003 self.log.info('Writer Type : %s\t : %i' % (key, len(lst)))
1004
1005 for m in self.writerDict["tuples"]:
1006 # rename Tuple output file with an appendix
1007 # based on worker id, for merging later
1008 newName = m.getNewName('.', '.w%i.' % (self.nodeID))
1009 self.config[m.key].Output = newName
1010
1011 # Suppress INFO Output for all but Worker-0
1012 # if self.nodeID == 0 :
1013 # pass
1014 # else :
1015 # self.config[ 'MessageSvc' ].OutputLevel = ERROR
1016
1017 if self.app == "Gauss":
1018 try:
1019 if "ToolSvc.EvtCounter" not in self.config:
1020 from Configurables import EvtCounter
1021 counter = EvtCounter()
1022 else:
1023 counter = self.config["ToolSvc.EvtCounter"]
1024 counter.UseIncident = False
1025 except:
1026 # ignore errors when trying to change the configuration of the EvtCounter
1027 self.log.warning('Cannot configure EvtCounter')
1028

## Member Data Documentation

## ◆ app

GaudiMP.GMPBase.Worker.app

Definition at line 1017 of file GMPBase.py.

## ◆ currentEvent

GaudiMP.GMPBase.Worker.currentEvent

Definition at line 1127 of file GMPBase.py.

## ◆ eventOutput

GaudiMP.GMPBase.Worker.eventOutput

Definition at line 986 of file GMPBase.py.

## ◆ filerecordsAgent

GaudiMP.GMPBase.Worker.filerecordsAgent

Definition at line 1041 of file GMPBase.py.

## ◆ firstEvTime

GaudiMP.GMPBase.Worker.firstEvTime

Definition at line 1109 of file GMPBase.py.

## ◆ nIn

GaudiMP.GMPBase.Worker.nIn

Definition at line 1091 of file GMPBase.py.

## ◆ tTime

GaudiMP.GMPBase.Worker.tTime

Definition at line 1142 of file GMPBase.py.

## ◆ vetoAlgs

GaudiMP.GMPBase.Worker.vetoAlgs

Definition at line 983 of file GMPBase.py.

## ◆ writerDict

GaudiMP.GMPBase.Worker.writerDict

Definition at line 981 of file GMPBase.py.

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

- GaudiMP/python/GaudiMP/GMPBase.py