1 """Configuration for POOL file writing
3 Copyright (C) 2002-2024 CERN for the benefit of the ATLAS collaboration
6 from AthenaConfiguration.AccumulatorCache
import AccumulatorCache
9 """Helper function to override TreeAutoFlush from flags."""
10 if not flags.Output.TreeAutoFlush
or not isinstance(flags.Output.TreeAutoFlush, dict):
13 if stream
not in flags.Output.TreeAutoFlush:
16 override = flags.Output.TreeAutoFlush[stream]
17 if override
is not None:
18 logger.info(
'Overriding TreeAutoFlush value for stream "%s" from %d to %d', stream, value, override)
25 Helper to get all the streams from configuration flags
26 For each stream that's configured to be written out
27 we have two flags w/ the following convention:
28 + Output.{STREAM}FileName
29 + Output.doWrite{STREAM}
32 for key, value
in flags._flagdict.items():
33 if key.startswith(
"Output.")
and key.endswith(
"FileName")
and value.get():
34 stream = key.removeprefix(
"Output.").removesuffix(
"FileName")
35 if stream
not in [
"HIST"]:
42 """Return ComponentAccumulator configured to Write POOL files"""
45 from AthenaCommon.Logging
import logging
46 logger = logging.getLogger(
'PoolWriteCfg' )
50 PoolAttributes += [
"DEFAULT_SPLITLEVEL ='0'"]
53 PoolAttributes += [
"STREAM_MEMBER_WISE = '1'"]
56 PoolAttributes += [
"DEFAULT_BUFFERSIZE = '32000'"]
59 PoolAttributes += [
"ContainerName = 'TTree=POOLContainerForm(DataHeaderForm)'; CONTAINER_SPLITLEVEL = '0'"]
60 PoolAttributes += [
"TREE_BRANCH_OFFSETTAB_LEN ='100'"]
62 oneDHForm = flags.Output.OneDataHeaderForm
65 from AthenaPoolCnvSvc
import PoolAttributeHelper
as pah
70 "EVNT" : [2, 1, 500, 0, 0],
71 "EVNT_TR" : [2, 1, 1, 0, 0],
72 "HITS" : [2, 1, 10, 0, 0],
73 "RDO" : [2, 1, 10, 0, 0],
74 "ESD" : [2, 1, 10, 0, 0],
75 "AOD" : [2, 1, 100, 0, 0],
76 "DAOD_PHYSVAL" : [5, 5, 100, 0, 1],
77 "DAOD_PHYS" : [5, 5, 500, 0, 1],
78 "DAOD_PHYSLITE" : [5, 5, 500, 1, 1],
79 "DAOD_TRUTH3" : [5, 5, 500, 1, 1],
80 "D2AOD_PHYSLITE" : [5, 5, 500, 1, 1],
84 OutputMetadataContainers = []
91 fileName = getattr(flags.Output, f
"{stream}FileName")
94 compAlg, compLvl, autoFlush, splitLvl, dynSplitLvl = 2, 1, 10, 0, 0
95 if stream
in defaults:
96 compAlg, compLvl, autoFlush, splitLvl, dynSplitLvl = defaults[stream]
97 elif "DAOD" in stream:
98 compAlg, compLvl, autoFlush, splitLvl, dynSplitLvl = 5, 5, 100, 0, 1
99 elif "D2AOD" in stream:
100 compAlg, compLvl, autoFlush, splitLvl, dynSplitLvl = 5, 5, 500, 1, 1
109 isTemporaryStream = fileName.endswith(
'_000')
or fileName.startswith(
'tmp.')
or stream
in flags.Output.TemporaryStreams
110 if isTemporaryStream:
111 logger.info(f
"Stream {stream} is marked as temporary, overwriting the compression settings to 101")
112 compAlg, compLvl = (1, 1)
if isTemporaryStream
else (compAlg, compLvl)
118 logger.debug(f
"{fileName=} {stream=} {compAlg=} {compLvl=} {autoFlush=} {splitLvl=} {dynSplitLvl=}")
121 outputCollection =
"POOLContainer"
122 poolContainerPrefix =
"CollectionTree"
126 isAugmentation = flags.hasFlag(f
"Output.{stream}ParentStream")
127 if not isAugmentation:
129 PoolAttributes += [ pah.setFileCompAlg( fileName, compAlg ) ]
130 PoolAttributes += [ pah.setFileCompLvl( fileName, compLvl ) ]
134 PoolAttributes += [ pah.setMaxBufferSize( fileName,
"131072" ) ]
135 PoolAttributes += [ pah.setMinBufferEntries( fileName,
"10" ) ]
139 PoolAttributes += [ f
"DatabaseName = '{fileName}'; INDEX_MASTER = 'POOLContainer(DataHeader)'" ]
142 outputCollection += f
"_{stream}"
143 poolContainerPrefix += f
"_{stream}"
144 OutputMetadataContainers += [f
"MetaData_{stream}"]
147 PoolAttributes += [ pah.setTreeAutoFlush( fileName, poolContainerPrefix, autoFlush ) ]
148 PoolAttributes += [ pah.setTreeAutoFlush( fileName, outputCollection, autoFlush ) ]
149 PoolAttributes += [ pah.setTreeAutoFlush( fileName,
"POOLContainerForm", autoFlush ) ]
152 PoolAttributes += [ pah.setContainerSplitLevel( fileName, poolContainerPrefix, splitLvl ) ]
153 PoolAttributes += [ pah.setContainerSplitLevel( fileName,
"Aux.", splitLvl ) ]
154 PoolAttributes += [ pah.setContainerSplitLevel( fileName,
"Dyn.", dynSplitLvl ) ]
159 if "EVNT" in stream
or "RDO" in stream:
160 PoolAttributes += [ f
"DatabaseName = '{fileName}'; FILEFORWARD_COMPATIBILITY = '1'" ]
165 maxAutoFlush =
max(maxAutoFlush, autoFlush)
169 useParallelCompression = flags.MP.UseSharedWriter
and flags.MP.UseParallelCompression
170 if useParallelCompression:
172 requestedEvents = flags.Exec.MaxEvents
173 availableEvents = flags.Input.FileNentries - flags.Exec.SkipEvents
174 totalEntries = availableEvents
if requestedEvents == -1
else min( availableEvents, requestedEvents )
175 if ( totalEntries > 0 )
and ( maxAutoFlush > 0 )
and ( maxAutoFlush * flags.Concurrency.NumProcs > totalEntries ):
176 logger.info(
"Not enough events to process, disabling parallel compression for SharedWriter!" )
177 logger.info( f
"Processing {totalEntries} events in {flags.Concurrency.NumProcs} workers "
178 f
"and a maximum (across all outputs) AutoFlush of {maxAutoFlush}")
179 useParallelCompression =
False
181 from AthenaPoolCnvSvc.PoolCommonConfig
import AthenaPoolCnvSvcCfg
183 PoolAttributes=PoolAttributes,
184 ParallelCompression=useParallelCompression,
185 StorageTechnology=flags.Output.StorageTechnology.EventData,
186 OutputMetadataContainers=OutputMetadataContainers,
187 OneDataHeaderForm = oneDHForm)