The ComponentAccumulator is a container for storing configuration of components. It may contain configuration of multiple components that form a consistent set (i.e. an algorithm & tools/services it needs for execution). The job configuration is built by merging several ComponentAccumulators.
... topCA.merge(alg1CA) topCA.merge(alg2CA) ...
During the merging it may occur that ComponentAccumulators (here
alg1CA) contain the configuration of the same component.
The ComponentAccumulator has functionality to unify their settings. So you don’t need to worry about adding a service/tool twice. On the contrary, ComponentAccumulator instances are supposed to be as much as possible self-contained, so it is recommended to add services/tools needed for an algorithm to work to the ComponentAccumulator and not to rely on the user to add what is necessary.
The settings unification process is called deduplication and is applied to every component in the merged ComponentAccumulators. It works as follows:
Methods that configure pieces of the job instantiate a ComponentAccumulator and add to it services, algorithms, etc. as needed.
They can call other configuration methods (to obtain the configuration of components they depend on) and merge the result with their own ComponentAccumulator.
As parameter they take a configuration flags container and potentially other
kwargs (positional and keyword arguments).
A typical implementation looks like this:
from AthenaConfiguration.ComponentAccumulator import ComponentAccumulator from AthenaConfiguration.ComponentFactory import CompFactory def MyAlgoCfg(flags, name="MyAlgo", **kwargs): acc = ComponentAccumulator() # Call the config method of a service that we need: from ARequiredSvcPack.ARequiredSvcPackConfig import SuperServiceCfg # We get an accumulator containing possibly other components that our SuperService # depends upon and the service itself svcAcc = SuperServiceCfg(flags) # SuperService is the primary component that SuperServiceConfig configures # Get it, so we can attach it to the ServiceHandle of an algorithm kwargs.setdefault("svcHandle", svcAcc.getPrimary()) # Merge its accumulator with our accumulator to absorb all required dependencies acc.merge(svcAcc) # NOTE: A shorthand for the above three lines is (more details below) # kwargs.setdefault("svcHandle", acc.getPrimaryAndMerge(SuperServiceCfg(flags))) # Set additional properties kwargs.setdefault("isData", not flags.Input.isMC) # instantiate algorithm configuration object setting its properties acc.addEventAlgo(CompFactory.MyAlgo(name, **kwargs)) # Return our accumulator (containing SuperService and its dependencies) return acc
AlgTools are special because they can’t exist without a parent. There is no meaningful way for accumulating them elsewhere.
However, configuration methods configuring a private
AlgTool with all its dependencies yet without a parent is still a valid use-case.
To allow this, the ComponentAccumulator class has methods
A function configuring an
AlgTool returns an instance of ComponentAccumulator that has the tool attached via
setPrivateTool and contains all the auxiliary components (services, conditions Algorithm etc.) that the tool needs to work.
The caller then obtains the private tool via the
popPrivateTools method and assigns it to the
PrivateToolHandle of the parent and merges the returned ComponentAccumulator with its own ComponentAccumulator.
This works for a single
AlgTool as well as for lists of
AlgTools that are typically assigned to a
Merging a ComponentAccumulator that has still private tools attached (e.g.
popPrivateTools was never called) will raise an exception complaining about a dangling private tool.
The ComponentAccumulator provides a handy shortcut method
getPrimaryAndMerge, see the next section) that does aforementioned two operations at once.
The example below illustrates how this works is as follows:
def PrivateToolCfg(flags): acc = ComponentAccumulator() ... # merge/add dependencies acc.addService(CompFactory.SomeService(...)) # it is recommended not to give a custom name to the private tool acc.setPrivateTools(CompFactory.ToolA(settingA=True)) return acc def AlgorithmCfg(flags): acc = ComponentAccumulator() tool = acc.popToolsAndMerge(PrivateToolCfg(flags)) # or longer alternative toolAcc = PrivateToolCfg(flags) tool = toolAcc.popPrivateTools() acc.merge(toolAcc) acc.addEventAlgo(CompFactory.Alg("MyAlg", myTool=tool, settingX=0.5)) return acc
When the ComponentAccumulator is a result of merging of several smaller components it may be useful to designate a component that is a primary concern for a configuration function. This way client code does not need to discover the component by name.
It may also be that the primary component may change depending on the flags. Also in this situation it is convenient to shield the client code by specifying the primary component. Such example is shown below:
def ToolCfg(flags): acc = ComponentAccumulator() # this adds some public tools acc.merge(OtherTooolsCfg(flags)) # not a primary component acc.addPublicTool(CompFactory.ToolX("X", setting=...)) # configure different tools depending on the flag value # here we will designate the primary component if flags.addA: acc.addPublicTool(CompFactory.ToolA("ToolA", settingX=...), primary=True) else: acc.addPublicTool(CompFactory.ToolA("ToolB", settingY=...), primary=True) return acc def ConsumerCfg(flags): acc = ComponentAccumulator() # instead of code like this: # toolAcc.getPublicTool("ToolA" if flags.flagA else "ToolB") we can do tool = acc.getPrimaryAndMerge(ToolCfg(flags)) # no need to agree on the name of the tool # configured by toolCfg # it possible to go the above in steps: # tempAcc = ToolCfg(falgs) # tool = tempAcc.getPrimary() # acc.merge(tempAcc) # however the shortcut method getPrimaryAndMerge is provided for your convenience acc.addEventAlgo(CompFactor.MyAlg("MyAlg", toolX=tool)) return acc
Configuration methods that are called many times may profit from caching their result:
from AthenaConfiguration.AccumulatorCache import AccumulatorCache @AccumulatorCache def MyAlgoCfg(flags, name="MyAlgo", **kwargs): ...
This will cache the result of
MyAlgoCfg (similar to Python’s
lru_cache) if the
function is called with the same flags (and other parameters) multiple times. The decorator
has a few (mostly experts) options that are documented in
AccumulatorCache. It can also print the cache hit/miss statistics via:
from AthenaConfiguration.AccumulatorCache import AccumulatorDecorator AccumulatorDecorator.printStats()
Do not blindly apply this decorator. Only use it for methods that are known to be hot spots.
The ComponentAccumulator has the following methods to add components to it:
othertop sequence are added to the destination sequence if the second argument is provided. Else the sequence structure is merged.
sequenceis added as a subsequence of the sequence with that name. Handy methods to create various types of sequences (parallel/serial with AND/OR logic) are defined in CFElements.
sequenceNameargument is provided algorithm is added to this sequence.
createis set the service is added to the set of services forcibly created by Athena early in the job even without any client requiring it.
setPrivateTools(tool or list of tools)
AlgTool(or list of private
AlgTools) to the accumulator. They need to be removed before merging.
For the explanation of the
primary option see above.
DeduplicationFailure or plain
TypeError are raised in case of misuse of these methods.
The ComponentAccumulator can be queried with these methods:
AlgToolor list of
AlgTools previously attached to the accumulator.
Additional methods are available for the use in top level scripts for running the configuration contained in the ComponentAccumulator.
openand closed after invocation of
Content of the configuration can be printed with:
printConfig(withDetails=False, summariseProps=False, onlyComponents = , printDefaults=False, printComponentsOnly=False)
Various flags define level of details that will be emitted from this function. The meaning should be obvious.
It is advised that in the file defining the function that generates configuration also the self-test is defined. If the configuration contains algorithms a short test athena job can be setup there. Configurations without the algorithms can still be tested in this way to some extent.
# assume this is the content of MyAlgConfig.py def MyAlgCfg(flags): acc = ComponentAccumulator() ... return acc if __name__ == "__main__": from AthenaCommon.Configurable import Configurable Configurable.configurableRun3Behavior = 1 # import the flags and set them from AthenaConfiguration.AllConfigFlags import ConfigFlags ... ConfigFlags.Exec.MaxEvents = 3 ... # use one of the predefined files from AthenaConfiguration.TestDefaults import defaultTestFiles ConfigFlags.Input.Files = defaultTestFiles.RAW # lock the flags ConfigFlags.lock() # create basic infrastructure from AthenaConfiguration.MainServicesConfig import MainServicesCfg acc = MainServicesCfg(ConfigFlags) # add the algorithm to the configuration acc.merge(MyAlgCfg(ConfigFlags)) # debug printout acc.printConfig(withDetails=True, summariseProps=True) # run the job status = acc.run() # report the execution status (0 ok, else error) import sys sys.exit(not status.isSuccess())
See the TileRawChannelMakerConfig example.
This script it then runable with the command:
python -m PackageName.MyAlgConfig
If the test does not take long time (it should not) it can be added as a package unit test in
atlas_add_test( MyAlgConfig_test SCRIPT python -m PackageName.MyAlgConfig PROPERTIES TIMEOUT 300 POST_EXEC_SCRIPT nopost.sh)
then this test is always run as part of gitlab CI and can/should be
In an identical way the main driver scripts can be constructed.
Useful functionality, to interpret command line options as flags
ConfigFlags.fillFromArgs() is used (see more).
For examples see: