@argh_named("control")
    @arg_verbose
    @arg_full_match
    @arg_flag("-n", "--no-deps", help="do not run dependency tasks")
    @arg_flag("-i", "--ignore-missing",
              help="do not fail in case no matching operations are found")
    @arg_quiet
    @arg_output_dir
    @arg_flag("-t", "--clock-tasks", dest="show_times",
              help="show timeline of execution for each tasks")
    @argh.arg("-j", "--jobs", metavar="N", type=int,
              help="max concurrent tasks (default: unlimited)")
    @argh.arg('pattern', type=str, help='config search pattern')
    @arg_host_access_method
    @argh.arg('operation', type=str, help='operation to execute')
    @expects_obj
    def handle_control(self, arg):
        confman = self.get_confman(arg.root_dir, reset_cache=False)
        manager = self.get_manager(confman)
        self.collect_all(manager)

        # collect all possible control operations
        all_configs = list(confman.find_config(".", all_configs=True))
        all_ops = []
        provider = {}
        for conf_node, conf in all_configs:
            plugin = conf.get_plugin()
            if not plugin:
                # skip pluginless configs
                self.log.debug("skipping pluginless: %s:%s", conf_node.name,
                               conf.name)
                continue
            elif conf_node.get_tree_property("template", False):
                # skip template nodes
                self.log.debug("skipping template node: %s:%s", conf_node.name,
                               conf.name)
                continue

            for op in plugin.iter_control_operations(conf_node, conf):
                all_ops.append(op)
                for feature in op["provides"]:
                    ops = provider.setdefault(feature, [])
                    ops.append(op)

        handled = set()

        def add_all_required_ops(op):
            key = (op["node"].name, op["config"].name, op["name"])
            if key in handled:
                return
            handled.add(key)

            node = op["node"]
            conf = op["config"]
            tasks[(node.name, conf.name, op["name"])] = op

            reqs = [(True, req) for req in op["requires"]]
            reqs.extend((False, req) for req in op["optional_requires"])

            for must_have, feature in reqs:
                try:
                    provider_ops = provider[feature]
                except KeyError:
                    if not must_have:
                        # this feature is optional, missing provider is ok
                        continue

                    raise errors.OperationError(
                        "%s/%s operation %r depends on feature %r, "
                        "which is not provided by any config" % (
                        node.name, conf.name, arg.operation, feature))

                depends = op.setdefault("depends", [])
                for dep_op in provider_ops:
                    depends.append(dep_op)
                    add_all_required_ops(dep_op)

        # select user-specified ops and their dependencies from the full list
        tasks = {}
        comparison = core.ConfigMatch(arg.pattern, full_match=arg.full_match)
        for op in all_ops:
            node = op["node"]
            conf = op["config"]
            # control op name, node name, config name, all must match
            if ((arg.operation != op["name"])
                or not comparison.match_node(node.name)
                or not comparison.match_config(conf.name)):
                continue

            op["run"] = True  # only explicit targets are marked for running
            add_all_required_ops(op)

        if not tasks:
            if arg.ignore_missing:
                self.log.info("no matching operations found: --ignore-missing specified, ok!")
                return
            else:
                raise errors.UserError("no matching operations found")

        if arg.no_deps:
            # filter out the implicit dependency tasks
            for op in all_ops:
                depends = op.get("depends", [])
                for dep_op in depends[:]:
                    if not dep_op.get("run"):
                        depends.remove(dep_op)

        # assign tasks
        runner = work.Runner(max_jobs=arg.jobs)
        logger = self.log.info if arg.verbose else self.log.debug
        for op_id, op in tasks.items():
            run = op.get("run") or (not arg.no_deps)
            op["run"] = run
            if not run:
                continue

            plugin = op["plugin"]
            logger("scheduled to run: %s/%s [%s]", op["node"].name,
                   op["config"].name, op["name"])
            task = ControlTask(op, arg.extras, verbose=arg.verbose,
                               quiet=arg.quiet, output_dir=arg.output_dir,
                               method=arg.method, color=arg.color)
            runner.add_task(task)

        # execute tasks
        runner.run_all()

        # collect results
        results = [task.op.get("result") for task in runner.stopped]
        failed = [r for r in results if r]
        skipped_count = sum(1 for op in tasks.values() if not op["run"])
        ran_count = len(tasks) - skipped_count
        assert len(results) == ran_count

        # add task times to report
        for i, task in enumerate(runner.stopped):
            task_name = "%s/%s" % (task.op["node"].name,
                                   task.op["config"].name)
            self.task_times.add_task(i, task_name, task.op["start_time"],
                                     task.op["stop_time"])

        if arg.verbose:
            for task in runner.stopped:
                res = task.op["result"]
                if res:
                    self.log.error("FAILED: %s/%s [%s]: %r",
                                   task.op["node"].name,
                                   task.op["config"].name, task.op["name"],
                                   task.op["result"])

        self.log.debug("all tasks finished: %r", results)
        if failed:
            raise errors.ControlError(
                "[%d/%d] control tasks failed (%d skipped)" % (
                    len(failed), ran_count, skipped_count))
        else:
            self.log.info(
                "all [%d] control tasks finished successfully (%d skipped)",
                    ran_count, skipped_count)
config control operation
dataset/ETHPy150Open ohmu/poni/poni/tool.py/Tool.handle_control