Method launch

opencompass/runners/local.py:69–196

Launch multiple tasks. Args: tasks (list[dict]): A list of task configs, usually generated by Partitioner. Returns: list[tuple[str, int]]: A list of (task name, exit code).

(self, tasks: List[Dict[str, Any]])

            logger.warning(f'Ignored argument in {self.__module__}: {k}={v}')

    def launch(self, tasks: List[Dict[str, Any]]) -> List[Tuple[str, int]]:
        """Launch multiple tasks.

        Args:
            tasks (list[dict]): A list of task configs, usually generated by
                Partitioner.

        Returns:
            list[tuple[str, int]]: A list of (task name, exit code).
        """

        status = []
        import torch

        if is_npu_available():
            visible_devices = 'ASCEND_RT_VISIBLE_DEVICES'
            device_nums = torch.npu.device_count()
        else:
            visible_devices = 'CUDA_VISIBLE_DEVICES'
            device_nums = torch.cuda.device_count()
        if visible_devices in os.environ:
            all_gpu_ids = [
                int(i)
                for i in re.findall(r'(?<!-)\d+', os.getenv(visible_devices))
            ]
        else:
            all_gpu_ids = list(range(device_nums))

        if self.debug:
            for task in tasks:
                task = TASKS.build(dict(cfg=task, type=self.task_cfg['type']))
                task_name = task.name
                num_gpus = task.num_gpus
                assert len(all_gpu_ids) >= num_gpus
                # get cmd
                mmengine.mkdir_or_exist('tmp/')
                import uuid
                uuid_str = str(uuid.uuid4())

                param_file = f'tmp/{uuid_str}_params.py'
                try:
                    task.cfg.dump(param_file)
                    # if use torchrun, restrict it behaves the same as non
                    # debug mode, otherwise, the torchrun will use all the
                    # available resources which might cause inconsistent
                    # behavior.
                    if len(all_gpu_ids) > num_gpus and num_gpus > 0:
                        get_logger().warning(f'Only use {num_gpus} GPUs for '
                                             f'total {len(all_gpu_ids)} '
                                             'available GPUs in debug mode.')
                    tmpl = get_command_template(all_gpu_ids[:num_gpus])
                    cmd = task.get_command(cfg_path=param_file, template=tmpl)
                    # run in subprocess if starts with torchrun etc.
                    if 'python3 ' in cmd or 'python ' in cmd:
                        # If it is an infer type task do not reload if
                        # the current model has already been loaded.
                        if 'infer' in self.task_cfg.type.lower():
                            # If a model instance already exists,