Refactor input/output file setting.

Allows for input and output files to be set in options files after the
EverythingHandler has been created and configured in an earlier options
file.

Hlt/Hlt1Conf/options/hlt1_example.py

@@ -8,7 +8,7 @@
 # granted to it by virtue of its status as an Intergovernmental Organization  #
 # or submit itself to any jurisdiction.                                       #
 ###############################################################################
-from PyConf.environment import EverythingHandler
+from PyConf.environment import EverythingHandler, setupInputFromTestFileDB
 from PyConf.Algorithms import FTRawBankDecoder
 from RecoConf.hlt1_tracking import require_gec
 from Hlt1Conf.lines.track_mva import (one_track_mva_line, two_track_mva_line,
@@ -29,6 +29,6 @@ with FTRawBankDecoder.bind(DecodingVersion=ftdec_v), \
     for name, builder in builders.items():
         env.registerLine(name, builder())
 
-env.setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI')
+setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI')
 env.configure()
 # env.plotDataFlow()

Hlt/Hlt1Conf/tests/options/hlt1_read.py → Hlt/Hlt1Conf/tests/options/hlt1_dst_input.py

@@ -8,49 +8,8 @@
 # granted to it by virtue of its status as an Intergovernmental Organization  #
 # or submit itself to any jurisdiction.                                       #
 ###############################################################################
-"""Read a file created by the `hlt1_write.py` options.
+"""Read a DST file created by the `hlt1_dst_output.py` options."""
+from PyConf.environment import setupInputFromTestFileDB
 
-This options file must be run with a command line flag that specifies which
-type of file to read, MDF or DST. This can be done like:
-
-    gaudirun.py --option 'import sys; sys.argv.append("--dst")' hlt1_read.py
-
-Pass the `--mdf` flag instead to read an MDF file.
-"""
-import sys
-
-from PyConf.environment import EverythingHandler
-from PyConf.Algorithms import FTRawBankDecoder
-from RecoConf.hlt1_tracking import require_gec
-from Hlt1Conf.lines.track_mva import (one_track_mva_line, two_track_mva_line,
-                                      debug_two_track_mva_line)
-
-ftdec_v = 4
-env = EverythingHandler(
-    threadPoolSize=1, nEventSlots=1, evtMax=1000, debug=True)
-
-with FTRawBankDecoder.bind(DecodingVersion=ftdec_v), \
-     require_gec.bind(FTDecodingVersion=ftdec_v):
-
-    builders = {
-        'Hlt1TrackMVALine': one_track_mva_line,
-        'Hlt1TwoTrackMVALine': two_track_mva_line,
-        'Hlt1DebugTwoTrackMVALine': debug_two_track_mva_line,
-    }
-    for name, builder in builders.items():
-        env.registerLine(name, builder())
-
-read_mdf = '--mdf' in sys.argv
-read_dst = '--dst' in sys.argv
-assert read_mdf or read_dst and not (
-    read_mdf and read_dst), 'Must specify exactly one of --mdf or --dst'
-
-if read_mdf:
-    env.setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI',
-                                 ['test_hlt1_persistence_mdf_write.mdf'],
-                                 'MDF')
-elif read_dst:
-    env.setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI',
-                                 ['test_hlt1_persistence_dst_write.dst'],
-                                 'ROOT')
-env.configure()
+setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI',
+                         ['test_hlt1_persistence_dst_write.dst'], 'ROOT')

Hlt/Hlt1Conf/tests/options/hlt1_dst_output.py

+###############################################################################
+# (c) Copyright 2019 CERN for the benefit of the LHCb Collaboration           #
+#                                                                             #
+# This software is distributed under the terms of the GNU General Public      #
+# Licence version 3 (GPL Version 3), copied verbatim in the file "COPYING".   #
+#                                                                             #
+# In applying this licence, CERN does not waive the privileges and immunities #
+# granted to it by virtue of its status as an Intergovernmental Organization  #
+# or submit itself to any jurisdiction.                                       #
+###############################################################################
+"""Write an HLT1-filtered DST file."""
+from PyConf.environment import setupOutput
+
+setupOutput('test_hlt1_persistence_dst_write.dst', 'ROOT')

Hlt/Hlt1Conf/tests/options/hlt1_write.py → Hlt/Hlt1Conf/tests/options/hlt1_mdf_input.py

@@ -8,46 +8,8 @@
 # granted to it by virtue of its status as an Intergovernmental Organization  #
 # or submit itself to any jurisdiction.                                       #
 ###############################################################################
-"""Write an HLT1-filtered file.
+"""Read an MDF file created by the `hlt1_mdf_output.py` options."""
+from PyConf.environment import setupInputFromTestFileDB
 
-This options file must be run with a command line flag that specifies which
-type of file to read, MDF or DST. This can be done like:
-
-    gaudirun.py --option 'import sys; sys.argv.append("--dst")' hlt1_write.py
-
-Pass the `--mdf` flag instead to write an MDF file.
-"""
-import sys
-
-from PyConf.environment import EverythingHandler
-from PyConf.Algorithms import FTRawBankDecoder
-from RecoConf.hlt1_tracking import require_gec
-from Hlt1Conf.lines.track_mva import (one_track_mva_line, two_track_mva_line,
-                                      debug_two_track_mva_line)
-
-ftdec_v = 4
-env = EverythingHandler(
-    threadPoolSize=1, nEventSlots=1, evtMax=1000, debug=True)
-
-with FTRawBankDecoder.bind(DecodingVersion=ftdec_v), \
-     require_gec.bind(FTDecodingVersion=ftdec_v):
-
-    builders = {
-        'Hlt1TrackMVALine': one_track_mva_line,
-        'Hlt1TwoTrackMVALine': two_track_mva_line,
-        'Hlt1DebugTwoTrackMVALine': debug_two_track_mva_line,
-    }
-    for name, builder in builders.items():
-        env.registerLine(name, builder())
-
-write_mdf = '--mdf' in sys.argv
-write_dst = '--dst' in sys.argv
-assert write_mdf or write_dst and not (
-    write_mdf and write_dst), 'Must specify exactly one of --mdf or --dst'
-
-env.setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI')
-if write_mdf:
-    env.setupOutput('test_hlt1_persistence_mdf_write.mdf', 'MDF')
-elif write_dst:
-    env.setupOutput('test_hlt1_persistence_dst_write.dst', 'ROOT')
-env.configure()
+setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI',
+                         ['test_hlt1_persistence_mdf_write.mdf'], 'MDF')

Hlt/Hlt1Conf/tests/options/hlt1_mdf_output.py

+###############################################################################
+# (c) Copyright 2019 CERN for the benefit of the LHCb Collaboration           #
+#                                                                             #
+# This software is distributed under the terms of the GNU General Public      #
+# Licence version 3 (GPL Version 3), copied verbatim in the file "COPYING".   #
+#                                                                             #
+# In applying this licence, CERN does not waive the privileges and immunities #
+# granted to it by virtue of its status as an Intergovernmental Organization  #
+# or submit itself to any jurisdiction.                                       #
+###############################################################################
+"""Write an HLT1-filtered MDF file."""
+from PyConf.environment import setupOutput
+
+setupOutput('test_hlt1_persistence_mdf_write.mdf', 'MDF')

Hlt/Hlt1Conf/tests/qmtest/persistency.qms/dst_read.qmt

@@ -18,8 +18,8 @@ Run HLT1 on an HLT1-filtered DST file.
 </set></argument>
 <argument name="program"><text>gaudirun.py</text></argument>
 <argument name="args"><set>
-  <text>--option</text><text>import sys; sys.argv.append("--dst")</text>
-  <text>$HLT1CONFROOT/tests/options/hlt1_read.py</text>
+  <text>$HLT1CONFROOT/options/hlt1_example.py</text>
+  <text>$HLT1CONFROOT/tests/options/hlt1_dst_input.py</text>
 </set></argument>
 <argument name="use_temp_dir"><enumeral>true</enumeral></argument>
 <argument name="validator"><text>

Hlt/Hlt1Conf/tests/qmtest/persistency.qms/dst_write.qmt

@@ -15,8 +15,8 @@ Run HLT1 and save an DST file.
 <extension class="GaudiTest.GaudiExeTest" kind="test">
 <argument name="program"><text>gaudirun.py</text></argument>
 <argument name="args"><set>
-  <text>--option</text><text>import sys; sys.argv.append("--dst")</text>
-  <text>$HLT1CONFROOT/tests/options/hlt1_write.py</text>
+  <text>$HLT1CONFROOT/options/hlt1_example.py</text>
+  <text>$HLT1CONFROOT/tests/options/hlt1_dst_output.py</text>
 </set></argument>
 <argument name="use_temp_dir"><enumeral>true</enumeral></argument>
 <argument name="validator"><text>

Hlt/Hlt1Conf/tests/qmtest/persistency.qms/mdf_read.qmt

@@ -18,8 +18,8 @@ Run HLT1 on an HLT1-filtered MDF file.
 </set></argument>
 <argument name="program"><text>gaudirun.py</text></argument>
 <argument name="args"><set>
-  <text>--option</text><text>import sys; sys.argv.append("--mdf")</text>
-  <text>$HLT1CONFROOT/tests/options/hlt1_read.py</text>
+  <text>$HLT1CONFROOT/options/hlt1_example.py</text>
+  <text>$HLT1CONFROOT/tests/options/hlt1_mdf_input.py</text>
 </set></argument>
 <argument name="use_temp_dir"><enumeral>true</enumeral></argument>
 <argument name="validator"><text>

Hlt/Hlt1Conf/tests/qmtest/persistency.qms/mdf_write.qmt

@@ -15,8 +15,8 @@ Run HLT1 and save an MDF file.
 <extension class="GaudiTest.GaudiExeTest" kind="test">
 <argument name="program"><text>gaudirun.py</text></argument>
 <argument name="args"><set>
-  <text>--option</text><text>import sys; sys.argv.append("--mdf")</text>
-  <text>$HLT1CONFROOT/tests/options/hlt1_write.py</text>
+  <text>$HLT1CONFROOT/options/hlt1_example.py</text>
+  <text>$HLT1CONFROOT/tests/options/hlt1_mdf_output.py</text>
 </set></argument>
 <argument name="use_temp_dir"><enumeral>true</enumeral></argument>
 <argument name="validator"><text>

Hlt/Hlt2Conf/options/hlt2_example.py

@@ -16,7 +16,7 @@ Run like any other options file:
 """
 from __future__ import absolute_import, division, print_function
 
-from PyConf.environment import EverythingHandler
+from PyConf.environment import EverythingHandler, setupInput
 
 from PyConf.Algorithms import FTRawBankDecoder
 
@@ -47,7 +47,7 @@ raw_event_format = 4.3
 
 env = EverythingHandler(
     threadPoolSize=1, nEventSlots=1, evtMax=100, debug=False)
-env.setupInput(
+setupInput(
     input_files,
     dataType='Upgrade',
     DDDBTag='dddb-20171126',

Hlt/RecoConf/options/hlt1_reco_baseline.py

@@ -14,7 +14,7 @@ from RecoConf.hlt1_tracking import (require_gec, require_pvs, make_pvs,
                                     make_velokalman_fitted_tracks)
 
 from PyConf.Algorithms import FTRawBankDecoder
-from PyConf.environment import EverythingHandler
+from PyConf.environment import EverythingHandler, setupInputFromTestFileDB
 
 ftdec_v = 4
 
@@ -31,5 +31,5 @@ def hlt1_full_track_reco_line():
 env = EverythingHandler(
     threadPoolSize=1, nEventSlots=1, evtMax=1000, debug=True)
 env.registerLine('Hlt1_reco_baseline', algs=hlt1_full_track_reco_line())
-env.setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI')
+setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI')
 env.configure()

Hlt/RecoConf/options/hlt2_reco_baseline.py

@@ -9,7 +9,7 @@
 # or submit itself to any jurisdiction.                                       #
 ###############################################################################
 
-from PyConf.environment import EverythingHandler
+from PyConf.environment import EverythingHandler, setupInputFromTestFileDB
 from PyConf.Algorithms import FTRawBankDecoder
 
 from RecoConf.hlt1_tracking import (
@@ -34,5 +34,5 @@ def hlt2_full_track_reco_line():
 env = EverythingHandler(
     threadPoolSize=1, nEventSlots=1, evtMax=100, debug=True)
 env.registerLine("Hlt2_reco_baseline", algs=hlt2_full_track_reco_line())
-env.setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI')
+setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI')
 env.configure()

Hlt/RecoConf/options/hlt2_reco_full_geometry.py

@@ -9,7 +9,7 @@
 # or submit itself to any jurisdiction.                                       #
 ###############################################################################
 
-from PyConf.environment import EverythingHandler
+from PyConf.environment import EverythingHandler, setupInputFromTestFileDB
 from RecoConf.hlt1_tracking import (
     require_gec,
     require_pvs,
@@ -35,5 +35,5 @@ env = EverythingHandler(
     threadPoolSize=1, nEventSlots=1, evtMax=100, debug=True)
 env.registerLine(
     "Hlt2_reco_full_geometry", algs=hlt2_full_geometry_track_reco_line())
-env.setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI')
+setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI')
 env.configure()

Hlt/RecoConf/options/mc_matching_example.py

@@ -17,7 +17,7 @@ from RecoConf.hlt1_tracking import (
 )
 from PyConf.Algorithms import (FTRawBankDecoder)
 from RecoConf.mc_checking import (monitor_track_efficiency)
-from PyConf.environment import (EverythingHandler)
+from PyConf.environment import (EverythingHandler, setupInputFromTestFileDB)
 
 ftdec_v = 4
 
@@ -38,5 +38,5 @@ env = EverythingHandler(
     debug=True,
     HistoFile="MCMatching_Hlt1ForwardTracking.root")
 env.registerLine('MCMatching_Hlt1ForwardTracking', algs=mc_matching_line())
-env.setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI')
+setupInputFromTestFileDB('MiniBrunel_2018_MinBias_FTv4_DIGI')
 env.configure()

PyConf/python/PyConf/environment.py

@@ -21,7 +21,7 @@ from Configurables import (
     LHCb__DetDesc__ReserveDetDescForEvent as reserveIOV,
     LHCb__Tests__FakeEventTimeProducer as DummyEventTime,
 )
-from PyConf.Algorithms import LHCb__MDFWriter, InputCopyStream, OutputStream
+from Configurables import LHCb__MDFWriter, InputCopyStream, OutputStream
 from Gaudi.Configuration import ConfigurableUser, DEBUG
 from GaudiConf import IOHelper
 from PRConfig.TestFileDB import test_file_db
@@ -37,6 +37,15 @@ __all__ = [
 ]
 
 log = logging.getLogger(__name__)
+# Writer algorithm and configuration used for a given file type
+KNOWN_WRITERS = {
+    'MDF': (LHCb__MDFWriter, dict(BankLocation='/Event/DAQ/RawEvent')),
+    # FIXME(AP): InputCopyStream is broken because it uses incidents to
+    # try to ensure all data it writes originated from the same input
+    # file
+    # 'ROOT': (InputCopyStream, dict()),
+    'ROOT': (OutputStream, dict(ItemList=['/Event/DAQ/RawEvent#1'])),
+}
 
 
 def _is_node(arg):
@@ -66,6 +75,106 @@ def _output_writer(writer_cls, filename, **kwargs):
     return writer_cls(**kwargs)
 
 
+def setupInput(inputFiles, dataType, DDDBTag, CONDDBTag, Simulation,
+               inputFileType):
+    # FIXME(AP) we need to modify the ApplicationMgr and query the
+    # HiveWhiteBoard held by the EverythingHandler; the former modifies global
+    # state which is not ideal
+    whiteboard = setup_component('HiveWhiteBoard', instanceName='EventDataSvc')
+    if inputFileType != 'MDF' and whiteboard.EventSlots > 1:
+        raise ConfigurationError(
+            "only MDF files can run in multithreaded mode, please change number of eventslots to 1"
+        )
+    setup_component(
+        'ApplicationMgr',
+        packageName='Gaudi.Configuration',
+        EvtSel="EventSelector")
+    input_iohelper = IOHelper(inputFileType, None)
+    input_iohelper.setupServices()
+    evtSel = input_iohelper.inputFiles(inputFiles, clear=True)
+    inputs = []
+    for inp in evtSel.Input:
+        inputs.append(inp + " IgnoreChecksum='YES'")
+    evtSel.Input = inputs
+    evtSel.PrintFreq = 10000
+    setup_component('DDDBConf', Simulation=Simulation, DataType=dataType)
+    setup_component(
+        'CondDB', Upgrade=True, Tags={
+            'DDDB': DDDBTag,
+            'SIMCOND': CONDDBTag
+        })
+    setup_component('IODataManager', DisablePFNWarning=True)
+
+
+def setupInputFromTestFileDB(testFileDBkey, inputFiles=None, fileType=None):
+    """Run from files defined by a TestFileDB key.
+
+    Parameters
+    ----------
+    testFileDBkey : str
+    inputFiles : list of str
+        Overrides the input files defined by the TestFileDB entry.
+    fileType : str
+    """
+    if inputFiles is not None and fileType is None:
+        raise ValueError('Must specify fileType when inputFiles is specified')
+
+    qualifiers = test_file_db[testFileDBkey].qualifiers
+    dataType = qualifiers['DataType']
+    Simulation = qualifiers['Simulation']
+    fileType = fileType or 'ROOT' if qualifiers[
+        'Format'] != 'MDF' else qualifiers['Format']
+    CondDBTag = qualifiers['CondDB']
+    DDDBTag = qualifiers['DDDB']
+    if not inputFiles:
+        inputFiles = test_file_db[testFileDBkey].filenames
+    setupInput(inputFiles, dataType, DDDBTag, CondDBTag, Simulation, fileType)
+
+
+def setupOutput(filename, filetype):
+    """Configure the application to write out a file.
+
+    Only the raw event, under `/Event/DAQ/RawEvent`, is persisted.
+
+    Must be called after `EverythingHandler.configure` has been called.
+
+    Parameters
+    ----------
+    filename : str
+    filetype : str
+        One of the keys of `KNOWN_WRITERS`.
+    """
+    # FIXME(AP) we need to modify the HLTControlFlowMgr and query the
+    # HiveWhiteBoard held by the EverythingHandler; the former modifies global
+    # state which is not ideal
+    whiteboard = setup_component('HiveWhiteBoard', instanceName='EventDataSvc')
+    scheduler = setup_component('HLTControlFlowMgr')
+    assert whiteboard.EventSlots == 1, 'Cannot write output multithreaded'
+    assert scheduler.ThreadPoolSize == 1, 'Cannot write output multithreaded'
+
+    assert filetype in KNOWN_WRITERS, 'Output filetype not supported: {}'.format(
+        filename)
+    writer_cls, configuration = KNOWN_WRITERS[filetype]
+
+    writer = _output_writer(writer_cls, filename, **configuration)
+
+    # FIXME(AP) Hack to add the writer to the overall control flow
+    # This causes the writer to only run when the HLT decision is positive
+    for node_name, _, members, _ in scheduler.CompositeCFNodes:
+        if node_name == 'HLT':
+            assert len(members) == 1, 'setupOutput called twice'
+            members.append(writer.getFullName())
+            break
+    else:
+        raise ConfigurationError(
+            'Must called EverythingHandler.configure before setupOutput')
+
+    hiveDataBroker = setup_component('HiveDataBrokerSvc')
+    hiveDataBroker.DataProducers.append(writer.getFullName())
+
+    return writer
+
+
 class PythonLoggingConf(ConfigurableUser):
     """Takes care of configuring the python logging verbosity."""
     # Make sure we're applied before anything else by listing
@@ -147,10 +256,6 @@ class EverythingHandler(object):
         if debug:
             self._scheduler.OutputLevel = DEBUG
             self._hiveDataBroker.OutputLevel = DEBUG
-        # IOHelper instance used to configure input files
-        self._input_iohelper = None
-        # Output file writer algorithm instance
-        self._output_writer = None
         # Instantiate the configurable which will set the python
         # logging verbosity at the right time.
         PythonLoggingConf()
@@ -185,79 +290,6 @@ class EverythingHandler(object):
             Algorithm(
                 CallgrindProfile, StartFromEventN=start, StopAtEventN=stop))
 
-    def setupInput(self, inputFiles, dataType, DDDBTag, CONDDBTag, Simulation,
-                   inputFileType):
-        if inputFileType != 'MDF' and self._whiteboard.EventSlots > 1:
-            raise ConfigurationError(
-                "only MDF files can run in multithreaded mode, please change number of eventslots to 1"
-            )
-        self._appMgr.EvtSel = "EventSelector"
-        self._input_iohelper = IOHelper(inputFileType, None)
-        self._input_iohelper.setupServices()
-        evtSel = self._input_iohelper.inputFiles(inputFiles)
-        inputs = []
-        for inp in evtSel.Input:
-            inputs.append(inp + " IgnoreChecksum='YES'")
-        evtSel.Input = inputs
-        evtSel.PrintFreq = 10000
-        setup_component('DDDBConf', Simulation=Simulation, DataType=dataType)
-        setup_component(
-            'CondDB',
-            Upgrade=True,
-            Tags={
-                'DDDB': DDDBTag,
-                'SIMCOND': CONDDBTag
-            })
-        setup_component('IODataManager', DisablePFNWarning=True)
-
-    def setupInputFromTestFileDB(self,
-                                 testFileDBkey,
-                                 inputFiles=None,
-                                 fileType=None):
-        """Run from files defined by a TestFileDB key.
-
-        Parameters
-        ----------
-        testFileDBkey
-        """
-        if inputFiles is not None and fileType is None:
-            raise ValueError(
-                'Must specify fileType when inputFiles is specified')
-
-        # if you want to have another file but use the testfileDB qualifiers, then set inputFiles
-        qualifiers = test_file_db[testFileDBkey].qualifiers
-        dataType = qualifiers['DataType']
-        Simulation = qualifiers['Simulation']
-        fileType = fileType or 'ROOT' if qualifiers[
-            'Format'] != 'MDF' else qualifiers['Format']
-        CondDBTag = qualifiers['CondDB']
-        DDDBTag = qualifiers['DDDB']
-        if not inputFiles:
-            inputFiles = test_file_db[testFileDBkey].filenames
-        self.setupInput(inputFiles, dataType, DDDBTag, CondDBTag, Simulation,
-                        fileType)
-
-    def setupOutput(self, filename, filetype):
-        # The input IOHelper will set up services we need
-        assert self._input_iohelper is not None, 'Must call setupInput before setupOutput'
-        assert self._whiteboard.EventSlots == 1, 'Cannot write output multithreaded'
-        assert self._scheduler.ThreadPoolSize == 1, 'Cannot write output multithreaded'
-
-        writers = {
-            'MDF': (LHCb__MDFWriter, dict(BankLocation='/Event/DAQ/RawEvent')),
-            # FIXME(AP): InputCopyStream is broken because it uses incidents to
-            # try to ensure all data it writes originated from the same input
-            # file
-            # 'ROOT': (InputCopyStream, dict()),
-            'ROOT': (OutputStream, dict(ItemList=['/Event/DAQ/RawEvent#1'])),
-        }
-        assert filetype in writers, 'Output filetype not supported: {}'.format(
-            filename)
-        writer_cls, configuration = writers[filetype]
-
-        self._output_writer = _output_writer(writer_cls, filename,
-                                             **configuration)
-
     def _addAlg(self, alg):
         if not is_algorithm(alg):
             raise TypeError('{} is not of type Algorithm'.format(alg))
@@ -319,7 +351,7 @@ class EverythingHandler(object):
         the information it needs to schedule all algorithms.
         """
         if self._lines:  # if not, you probably set it up yoself?
-            self.setup_default_controlflow(self._lines, self._output_writer)
+            self.setup_default_controlflow(self._lines)
 
         configuration = dataflow_config()
         for alg in self._algs: