DM-17038: Refactoring of CmdLineFwk class #8
Conversation
andy-slac
force-pushed
the
tickets/DM-17038
branch
3 times, most recently
from
073ce59
to
3a2d871
Compare
| self.taskFactory = taskFactory | ||
|
|
||
| def execute(self, taskClass, config, quantum): | ||
| """Execute super-task on a single Quantum. |
There was a problem hiding this comment.
Now PipelineTask, not super-task.
| # if hasattr(dataRef, "dataId"): | ||
| # lsst.log.MDC("LABEL", str(dataRef.dataId)) | ||
| # elif isinstance(dataRef, (list, tuple)): | ||
| # lsst.log.MDC("LABEL", str([ref.dataId for ref in dataRef if hasattr(ref, "dataId")])) |
There was a problem hiding this comment.
Did you intend to comment all of this out?
There was a problem hiding this comment.
I was intentionally delaying it 🙂. I'll comment it out now and make sure it works (that was just a copy from CmdLineTask, I guess we have to match that).
| Instance of ``taskClass`` type. | ||
| """ | ||
| # call task factory for that | ||
| return self.taskFactory.makeTask(taskClass, config, None, self.butler) |
There was a problem hiding this comment.
If you want any custom logging, I imagine you'd need to pass that information here through the TaskFactory into the Task constructor (maybe just via **kwarg forwarding?).
Side note: I assume the TaskFactory is here just to handle the complexity of getting InitInput datasets from Butler, since taskClass is defined here to really be a PipelineTask class, not the string name of one (which is good - the fewer places that need to worry about tasks-as-strings, the better). Does that mean we can factor that logic out of the TaskFactory ABC so it's just a function call that classes like this can use?
There was a problem hiding this comment.
Not sure what do you mean by custom logging and what info do you want to pass to a task constructor, do you have some examples?
We can certainly factor out implementation of the makeTask from TaskFactory - it does not need that dependency on a TaskLoader so it may be beneficial to be able to use that code without instantiating TaskLoader.
There was a problem hiding this comment.
I was going to say that Task.__init__ takes a lsst.log.Log to be used for all of the logging it and its subtasks do, so if you want to customize that you'd need to do it here.
But looking closer, it seems it just uses that to set the Log instance's name (which we probably don't want to change here), and I presume that means customization (e.g. making all log messages include the data ID) would happen via some other API that doesn't work directly on the Log instance.
| return self.taskFactory.makeTask(taskClass, config, None, self.butler) | ||
|
|
||
| def updateQuantumInputs(self, quantum): | ||
| """Update qunatum with extra information. |
| Some methods may require input DataRefs to have non-None | ||
| ``dataset_id``, but in case of intermediate dataset it may not be | ||
| filled during QuantumGraph construction. This method will retrieve | ||
| missing info from registry. |
There was a problem hiding this comment.
DataRefs -> DatasetRefs
Is the concern about needing dataset_id about executors or PipelineTasks? I don't think PipelineTasks will care, as butler.get will do a find itself if dataset_id is None.
There was a problem hiding this comment.
I found that registry.addQuantum() needs it, so it is Executors concern mostly. Not sure if anything else would ever need it, I just think that it's safer to add it here for consistency with pre-existing datasets that have dataset_id.
| # TODO: see if Pipeline and software versions are already written | ||
| # to butler and associated with Run, check for consistency if they | ||
| # are, and if so skip writing TaskInitOutputs (because those should | ||
| # also only be done once). If not, write them. |
There was a problem hiding this comment.
This TODO represents one way we may want to approach this one (and probably the most likely one), but I think we should think about possible alternatives and high-level use cases before treating it as a straightforward TODO item. Comparing configs and versions and disabling that with --clobber-config has been very messy in Gen2, and I'd like to find a better way.
| # copy all collected refs to output collection | ||
| collection = butler.run.collection | ||
| registry = butler.registry | ||
| registry.associate(collection, list(id2ref.values())) |
There was a problem hiding this comment.
At least after DM-16227, list here should be unnecessary; any iterable would be fine.
There was a problem hiding this comment.
👍 Will change it once that branch is merged.
|
|
||
| Implementation of this method depends on particular execution model | ||
| and it has to be provided by a subclass. Execution model determines | ||
| what happens here, it can be either acual running of the task or, |
There was a problem hiding this comment.
typo: acual. Also, semicolon is better after "here" instead of comma.
| for example, generation of the scripts for delayed batch execution. | ||
|
|
||
| Any exception raised in this method will be propagated to the caller | ||
| of `execute` method. |
There was a problem hiding this comment.
I think the exception propagation behavior should either be left to subclasses or controllable some other way; I think it's quite likely an activator might want to just warn and/or remember failures and proceed with executing any Quanta that did not depend on them.
There was a problem hiding this comment.
I think this is consistent with what is written in docstring - sub-classes are free to raise exceptions or do anything else according their own policy, same applies to activators. Error handling strategy is not very well defined at this point, we may will to think about it more.
| args : `argparse.Namespace` | ||
| Parsed command line | ||
| """ | ||
| # If output collections are given then use them to override | ||
| # butler-configured ones. |
There was a problem hiding this comment.
There can only be one output collection, right?
There was a problem hiding this comment.
I don't know 🤔 There is one default output collection, but I think we also support per-dataset type collections too. I'll change the comment to make it more explicit.
There was a problem hiding this comment.
I suppose I shouldn't have phrased this as a question - I don't think Butler can handle multiple output collections in the same run.
This new class implements small piece of cmdLineFwk funcionality which is responsible for execution of single Quantum. Adds a bunch of customization points so that new activators can extend default behavior. No unit test yet for this class, it needs butler and registry mocks, wouls be easier to make that when we have some simple actual butler for unit tests.
andy-slac
force-pushed
the
tickets/DM-17038
branch
from
3a2d871
to
6408e8f
Compare
One new class QuantumGraphExecutor which implements most common pre-execution operations but defers actual execution to a sub-class. MPGraphExecutor implements multiprocess-based execution on top of that.
andy-slac
force-pushed
the
tickets/DM-17038
branch
from
6408e8f
to
33512a5
Compare
|
@TallJimbo, one more commit to look at: 29e87d4. I hope this is last refactoring on this ticket 🙂. It splits initialization steps of QuantumGraphExecutor into a separate non-abstract class. |
| """Initialization of registry for QuantumGraph execution. | ||
|
|
||
| This class encapsulates all necessary operations that have to be performed | ||
| on butler and registry to prepare them for QuanrumGraph execution. |
|
|
||
| Parameters | ||
| ---------- | ||
| graph : `~lsst.pip.base.QuantumGraph` |
| Data butler instance. | ||
| """ | ||
| def __init__(self, butler): | ||
| self.butler = butler |
There was a problem hiding this comment.
Given that it would be nice to run at least some of these operations on a Pipeline in the future as well as a QuantumGraph, what do you think about passing the QuantumGraph (either instead of or in addition to the Butler) at construction instead of to each of the methods? That would let us have a different class in the future with (mostly) the same method interfaces to handle Pipelines.
There was a problem hiding this comment.
Hmm, some of that interface cannot even work with Pipeline, so I'm not sure how we can make it work. We could probably define some polymorphic interface for that but if it's going to work differently for graph and pipeline it will not be truly polymorphic. I'd say that having separate interface for Pipeline (maybe even in this same class) is probably better and less confusing.
andy-slac
force-pushed
the
tickets/DM-17038
branch
from
29e87d4
to
abefd7f
Compare
CmdLineFwk class was split into small set of (mostly) reusable components. QuantumGraph execution is now handled by new QuantumGraphExecutor class which can be customized for particular execution model.
multiprocessing-based execution model is implemented on top of that inMPGraphExecutorclass.Remaining code in
CmdLineFwkclass was slightly refactored to to make structure less confusing.