3 from AthenaConfiguration.AllConfigFlags
import initConfigFlags, GetFileMD
4 from AthenaConfiguration.ComponentAccumulator
import ComponentAccumulator
5 from AthenaConfiguration.ComponentFactory
import CompFactory
6 from AthenaConfiguration.Enums
import Format, ProductionStep
7 from AthenaConfiguration.MainServicesConfig
import MainServicesCfg
9 from AthenaCommon.Logging
import log
as msg
11 import os, shutil, uuid
15 """Fill MP configuration flags from run arguments."""
16 if hasattr(runArgs,
"athenaMPWorkerTopDir"):
17 flags.MP.WorkerTopDir = runArgs.athenaMPWorkerTopDir
19 if hasattr(runArgs,
"athenaMPOutputReportFile"):
20 flags.MP.OutputReportFile = runArgs.athenaMPOutputReportFile
22 if hasattr(runArgs,
"athenaMPCollectSubprocessLogs"):
23 flags.MP.CollectSubprocessLogs = runArgs.athenaMPCollectSubprocessLogs
25 if hasattr(runArgs,
"athenaMPStrategy"):
26 flags.MP.Strategy = runArgs.athenaMPStrategy
28 if hasattr(runArgs,
"athenaMPReadEventOrders"):
29 flags.MP.ReadEventOrders = runArgs.athenaMPReadEventOrders
31 if hasattr(runArgs,
"athenaMPEventOrdersFile"):
32 flags.MP.EventOrdersFile = runArgs.athenaMPEventOrdersFile
34 if hasattr(runArgs,
"athenaMPEventsBeforeFork"):
35 flags.MP.EventsBeforeFork = runArgs.athenaMPEventsBeforeFork
37 if hasattr(runArgs,
"sharedWriter"):
38 flags.MP.UseSharedWriter = runArgs.sharedWriter
40 if hasattr(runArgs,
"sharedReader"):
41 flags.MP.UseSharedReader = runArgs.sharedReader
43 if hasattr(runArgs,
"parallelCompression"):
44 flags.MP.UseParallelCompression = runArgs.parallelCompression
46 if hasattr(runArgs,
"eventService"):
47 flags.MP.Strategy =
"EventService"
51 os.putenv(
'XRD_ENABLEFORKHANDLERS',
'1')
52 os.putenv(
'XRD_RUNFORKHANDLER',
'1')
57 mpevtloop = CompFactory.AthMpEvtLoopMgr(EventPrintoutInterval = flags.Exec.EventPrintoutInterval)
59 mpevtloop.NWorkers = flags.Concurrency.NumProcs
60 mpevtloop.Strategy = flags.MP.Strategy
61 mpevtloop.WorkerTopDir = flags.MP.WorkerTopDir
62 mpevtloop.OutputReportFile = flags.MP.OutputReportFile
63 mpevtloop.CollectSubprocessLogs = flags.MP.CollectSubprocessLogs
64 mpevtloop.PollingInterval = flags.MP.PollingInterval
65 mpevtloop.MemSamplingInterval = flags.MP.MemSamplingInterval
66 mpevtloop.IsPileup = flags.Common.ProductionStep
in [ProductionStep.Digitization, ProductionStep.PileUpPresampling, ProductionStep.FastChain]
and flags.Digitization.PileUp
67 mpevtloop.EventsBeforeFork = 0
if flags.MP.Strategy ==
'EventService' else flags.MP.EventsBeforeFork
70 filemgr = CompFactory.FileMgr(LogFile=
"FileManagerLog")
71 result.addService(filemgr)
75 if os.path.isfile(
'PoolFileCatalog.xml'):
76 shutil.copyfile(
'PoolFileCatalog.xml',
'PoolFileCatalog.xml.AthenaMP-saved')
81 msg.warning(
'Nonpositive ChunkSize (%i) caught, setting it to 1', chunk_size)
85 debug_worker = flags.Concurrency.DebugWorkers
86 event_range_channel = flags.MP.EventRangeChannel
87 use_shared_reader = flags.MP.UseSharedReader
88 use_shared_writer = flags.MP.UseSharedWriter
89 unique_id = f
"{str(os.getpid())}-{uuid.uuid4().hex}"
91 if flags.MP.Strategy ==
'SharedQueue' or flags.MP.Strategy ==
'RoundRobin':
93 AthenaSharedMemoryTool = CompFactory.AthenaSharedMemoryTool
95 if flags.Input.Format
is Format.BS:
96 evSel = CompFactory.EventSelectorByteStream(
"EventSelector")
98 from ByteStreamCnvSvc.ByteStreamConfig
import ByteStreamReadCfg
102 evSel = CompFactory.EventSelectorAthenaPool(
"EventSelector")
105 SharedMemoryName=f
"InputStream{unique_id}")
107 from AthenaPoolCnvSvc.PoolReadConfig
import PoolReadCfg
109 from AthenaPoolCnvSvc.PoolCommonConfig
import AthenaPoolCnvSvcCfg
113 SharedMemoryName=f
"EventStream{unique_id}")
114 result.addService(evSel)
116 if use_shared_writer:
117 if any((flags.Output.doWriteESD,
118 flags.Output.doWriteAOD,
119 flags.Output.doWriteDAOD,
120 flags.Output.doWriteRDO))
or flags.Output.HITSFileName!=
'':
121 AthenaSharedMemoryTool = CompFactory.AthenaSharedMemoryTool
123 SharedMemoryName=f
"OutputStream{unique_id}")
125 from AthenaPoolCnvSvc.PoolCommonConfig
import AthenaPoolCnvSvcCfg
128 queue_provider = CompFactory.SharedEvtQueueProvider(UseSharedReader=use_shared_reader,
129 IsPileup=mpevtloop.IsPileup,
130 EventsBeforeFork=mpevtloop.EventsBeforeFork,
131 ChunkSize=chunk_size)
132 if flags.Concurrency.NumThreads > 0:
133 if mpevtloop.IsPileup:
134 raise Exception(
'Running pileup digitization in mixed MP+MT currently not supported')
135 from AthenaConfiguration.MainServicesConfig
import AthenaMtesEventLoopMgrCfg
137 queue_consumer = CompFactory.SharedHiveEvtQueueConsumer(UseSharedWriter=use_shared_writer,
138 EventsBeforeFork=mpevtloop.EventsBeforeFork,
141 queue_consumer = CompFactory.SharedEvtQueueConsumer(UseSharedReader=use_shared_reader,
142 UseSharedWriter=use_shared_writer,
143 IsPileup=mpevtloop.IsPileup,
144 IsRoundRobin=(flags.MP.Strategy==
'RoundRobin'),
145 EventsBeforeFork=mpevtloop.EventsBeforeFork,
146 ReadEventOrders=flags.MP.ReadEventOrders,
147 EventOrdersFile=flags.MP.EventOrdersFile,
149 mpevtloop.Tools += [ queue_provider, queue_consumer ]
151 if use_shared_writer:
152 shared_writer = CompFactory.SharedWriterTool(MotherProcess=(mpevtloop.EventsBeforeFork>0),
153 IsPileup=mpevtloop.IsPileup,
155 mpevtloop.Tools += [ shared_writer ]
157 elif flags.MP.Strategy==
'EventService':
158 channelScatterer2Processor =
"AthenaMP_Scatterer2Processor"
159 channelProcessor2EvtSel =
"AthenaMP_Processor2EvtSel"
161 mpevtloop.Tools += [ CompFactory.EvtRangeScatterer(ProcessorChannel = channelScatterer2Processor,
162 EventRangeChannel = event_range_channel,
163 DoCaching=flags.MP.EvtRangeScattererCaching) ]
164 mpevtloop.Tools += [ CompFactory.EvtRangeProcessor(IsPileup=mpevtloop.IsPileup,
165 Channel2Scatterer = channelScatterer2Processor,
166 Channel2EvtSel = channelProcessor2EvtSel,
167 Debug=debug_worker) ]
169 from AthenaServices.OutputStreamSequencerSvcConfig
import OutputStreamSequencerSvcCfg
172 msg.warning(
"Unknown strategy %s. No MP tools will be configured", flags.MP.Strategy)
174 result.addService(mpevtloop, primary=
True)
180 if flags.MP.ChunkSize > 0:
181 chunk_size = flags.MP.ChunkSize
182 msg.info(
'Chunk size set to %i', chunk_size)
186 if flags.MP.UseSharedReader:
187 msg.info(
'Shared Reader in use, chunk_size set to default (%i)', chunk_size)
189 elif flags.MP.ChunkSize == -1:
190 if md.get(
'file_comp_alg',-1) == 2:
191 chunk_size = md.get(
'auto_flush',-1)
192 msg.info(
'Chunk size set to auto flush (%i)', chunk_size)
194 msg.info(
'LZMA algorithm not in use, chunk_size set to default (%i)', chunk_size)
196 elif flags.MP.ChunkSize == -2:
197 if md.get(
'file_comp_alg',-1)
in [1,2]:
198 chunk_size = md.get(
'auto_flush',-1)
199 msg.info(
'Chunk size set to auto flush (%i)', chunk_size)
201 msg.info(
'LZMA nor ZLIB in use, chunk_size set to default (%i)', chunk_size)
203 elif flags.MP.ChunkSize == -3:
204 if md.get(
'file_comp_alg',-1)
in [1,2,4]:
205 chunk_size = md.get(
'auto_flush',-1)
206 msg.info(
'Chunk size set to auto flush (%i)', chunk_size)
208 msg.info(
'LZMA, ZLIB nor LZ4 in use, chunk_size set to (%i)', chunk_size)
210 elif flags.MP.ChunkSize <= -4:
211 chunk_size = md.get(
'auto_flush',-1)
212 msg.info(
'Chunk size set to auto flush (%i)', chunk_size)
214 msg.warning(
'Invalid ChunkSize, Chunk Size set to default (%i)', chunk_size)
219 if __name__==
"__main__":
222 from AthenaConfiguration.TestDefaults
import defaultTestFiles
224 flags.Input.Files = defaultTestFiles.ESD
225 flags.Exec.MaxEvents=10
226 flags.Concurrency.NumProcs=2
229 from AthenaPoolCnvSvc.PoolReadConfig
import EventSelectorAthenaPoolCfg