runner.py 17 KB
Newer Older
1
2
3
4
5
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.

"""SuperBench Runner."""

6
import json
7
import random
8
from pathlib import Path
9
from pprint import pformat
10
from collections import defaultdict
11

12
import jsonlines
13
from natsort import natsorted
14
from joblib import Parallel, delayed
15
16
from omegaconf import ListConfig, OmegaConf

17
from superbench.common.utils import SuperBenchLogger, logger
18
from superbench.runner.ansible import AnsibleClient
19
from superbench.benchmarks import ReduceType, Reducer
20
from superbench.monitor import MonitorRecord
21
22
23
24


class SuperBenchRunner():
    """SuperBench runner class."""
25
    def __init__(self, sb_config, docker_config, ansible_config, sb_output_dir):
26
27
28
29
30
31
        """Initilize.

        Args:
            sb_config (DictConfig): SuperBench config object.
            docker_config (DictConfig): Docker config object.
            ansible_config (DictConfig): Ansible config object.
32
            sb_output_dir (str): SuperBench output directory.
33
34
35
36
        """
        self._sb_config = sb_config
        self._docker_config = docker_config
        self._ansible_config = ansible_config
37
38
        self._sb_output_dir = sb_output_dir
        self._output_path = Path(sb_output_dir).expanduser().resolve()
39
        self._ansible_client = AnsibleClient(ansible_config)
40
41

        self.__set_logger('sb-run.log')
42
        logger.info('Runner uses config: %s.', pformat(self._sb_config))
43
        logger.info('Runner writes to: %s.', str(self._output_path))
44

45
        self._sb_benchmarks = self._sb_config.superbench.benchmarks
46
        self.__validate_sb_config()
47
48
49
        self._sb_enabled_benchmarks = self.__get_enabled_benchmarks()
        logger.info('Runner will run: %s', self._sb_enabled_benchmarks)

50
51
52
53
54
55
    def __set_logger(self, filename):
        """Set logger and add file handler.

        Args:
            filename (str): Log file name.
        """
56
        SuperBenchLogger.add_handler(logger.logger, filename=str(self._output_path / filename))
57

Yifan Xiong's avatar
Yifan Xiong committed
58
    def __validate_sb_config(self):    # noqa: C901
59
60
61
62
63
64
        """Validate SuperBench config object.

        Raise:
            InvalidConfigError: If input config is invalid.
        """
        # TODO: add validation and defaulting
65
66
        if not self._sb_config.superbench.env:
            self._sb_config.superbench.env = {}
67
68
69
70
71
72
73
74
75
76
77
78
        for name in self._sb_benchmarks:
            if not self._sb_benchmarks[name].modes:
                self._sb_benchmarks[name].modes = []
            for idx, mode in enumerate(self._sb_benchmarks[name].modes):
                if mode.name == 'local':
                    if not mode.proc_num:
                        self._sb_benchmarks[name].modes[idx].proc_num = 1
                    if not mode.prefix:
                        self._sb_benchmarks[name].modes[idx].prefix = ''
                elif mode.name == 'torch.distributed':
                    if not mode.proc_num:
                        self._sb_benchmarks[name].modes[idx].proc_num = 8
Yifan Xiong's avatar
Yifan Xiong committed
79
80
81
82
83
84
85
86
87
88
89
90
                elif mode.name == 'mpi':
                    if not mode.mca:
                        self._sb_benchmarks[name].modes[idx].mca = {
                            'pml': 'ob1',
                            'btl': '^openib',
                            'btl_tcp_if_exclude': 'lo,docker0',
                            'coll_hcoll_enable': 0,
                        }
                    if not mode.env:
                        self._sb_benchmarks[name].modes[idx].env = {}
                    for key in ['PATH', 'LD_LIBRARY_PATH', 'SB_MICRO_PATH']:
                        self._sb_benchmarks[name].modes[idx].env.setdefault(key, None)
91

92
93
94
95
96
97
98
99
100
101
102
103
104
    def __get_enabled_benchmarks(self):
        """Get enabled benchmarks list.

        Return:
            list: List of benchmarks which will be executed.
        """
        if self._sb_config.superbench.enable:
            if isinstance(self._sb_config.superbench.enable, str):
                return [self._sb_config.superbench.enable]
            elif isinstance(self._sb_config.superbench.enable, (list, ListConfig)):
                return list(self._sb_config.superbench.enable)
        return [k for k, v in self._sb_benchmarks.items() if v.enable]

105
    def __get_mode_command(self, benchmark_name, mode):
106
107
108
        """Get runner command for given mode.

        Args:
109
            benchmark_name (str): Benchmark name.
110
111
112
113
114
            mode (DictConfig): Runner mode.

        Return:
            str: Runner command.
        """
115
116
117
118
        exec_command = ('sb exec --output-dir {output_dir} -c sb.config.yaml -C superbench.enable={name}').format(
            name=benchmark_name,
            output_dir=self._sb_output_dir,
        )
119
120
121
122
123
124
        mode_command = exec_command
        if mode.name == 'local':
            mode_command = '{prefix} {command}'.format(
                prefix=mode.prefix.format(proc_rank=mode.proc_rank, proc_num=mode.proc_num),
                command=exec_command,
            )
125
            mode_command = f'PROC_RANK={mode.proc_rank} {mode_command.strip()}'
126
        elif mode.name == 'torch.distributed':
127
128
            # TODO: replace with torch.distributed.run in v1.9
            # TODO: only supports node_num=1 and node_num=all currently
129
130
            torch_dist_params = '' if mode.node_num == 1 else \
                '--nnodes=$NNODES --node_rank=$NODE_RANK --master_addr=$MASTER_ADDR --master_port=$MASTER_PORT '
131
            mode_command = (
132
133
134
135
                f'python3 -m torch.distributed.launch'
                f' --use_env --no_python --nproc_per_node={mode.proc_num} {torch_dist_params}{exec_command}'
                f' superbench.benchmarks.{benchmark_name}.parameters.distributed_impl=ddp'
                f' superbench.benchmarks.{benchmark_name}.parameters.distributed_backend=nccl'
136
            )
Yifan Xiong's avatar
Yifan Xiong committed
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
        elif mode.name == 'mpi':
            mode_command = (
                'mpirun '    # use default OpenMPI in image
                '-tag-output '    # tag mpi output with [jobid,rank]<stdout/stderr> prefix
                '-allow-run-as-root '    # allow mpirun to run when executed by root user
                '-hostfile hostfile '    # use prepared hostfile
                '-map-by ppr:{proc_num}:node '    # launch {proc_num} processes on each node
                '-bind-to numa '    # bind processes to numa
                '{mca_list} {env_list} {command}'
            ).format(
                proc_num=mode.proc_num,
                mca_list=' '.join(f'-mca {k} {v}' for k, v in mode.mca.items()),
                env_list=' '.join(f'-x {k}={v}' if v else f'-x {k}' for k, v in mode.env.items()),
                command=exec_command,
            )
        else:
            logger.warning('Unknown mode %s.', mode.name)
154
        return mode_command.strip()
155

156
157
158
159
160
    def deploy(self):    # pragma: no cover
        """Deploy SuperBench environment."""
        logger.info('Preparing SuperBench environment.')
        extravars = {
            'ssh_port': random.randint(1 << 14, (1 << 15) - 1),
161
            'output_dir': str(self._output_path),
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
            'docker_image': self._docker_config.image,
        }
        if bool(self._docker_config.username) and bool(self._docker_config.password):
            extravars.update(
                {
                    'docker_registry': self._docker_config.registry,
                    'docker_username': self._docker_config.username,
                    'docker_password': self._docker_config.password,
                }
            )
        self._ansible_client.run(self._ansible_client.get_playbook_config('deploy.yaml', extravars=extravars))

    def check_env(self):    # pragma: no cover
        """Check SuperBench environment."""
        logger.info('Checking SuperBench environment.')
177
        OmegaConf.save(config=self._sb_config, f=str(self._output_path / 'sb.config.yaml'))
178
        self._ansible_client.run(
179
180
181
            self._ansible_client.get_playbook_config(
                'check_env.yaml',
                extravars={
182
                    'output_dir': str(self._output_path),
183
184
185
                    'env': '\n'.join(f'{k}={v}' for k, v in self._sb_config.superbench.env.items()),
                }
            )
186
187
        )

188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
    def fetch_results(self):    # pragma: no cover
        """Fetch benchmark results on all nodes."""
        try:
            (self._output_path / 'nodes').mkdir(mode=0o755, parents=True, exist_ok=True)
        except Exception:
            logger.exception('Failed to create directory %s.', str(self._output_path / 'nodes'))
            raise
        self._ansible_client.run(
            self._ansible_client.get_playbook_config(
                'fetch_results.yaml',
                extravars={
                    'sb_output_dir': self._sb_output_dir,
                    'absolute_output_dir': str(self._output_path),
                }
            )
        )

205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
    def __create_results_summary(self):    # pragma: no cover
        """Create the result summary file of all nodes."""
        all_results = list()
        for node_path in (self._output_path / 'nodes').glob('*'):
            if not node_path.is_dir():
                continue
            results_summary = self.__create_single_node_summary(node_path)
            results_summary['node'] = node_path.name
            all_results.append(results_summary)

        with (self._output_path / 'results-summary.jsonl').open(mode='w') as f:
            for result in all_results:
                json.dump(result, f)
                f.write('\n')

220
    def __create_single_node_summary(self, node_path):    # pragma: no cover # noqa: C901
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
        """Create the result summary file of single node.

        Args:
            node_path (Path): The Path instance of node directory.

        Returns:
            dict: Result summary of single node.
        """
        results_summary = dict()
        reduce_ops = dict()
        file_list = [Path(f) for f in natsorted([str(f) for f in node_path.glob('**/results.json')])]
        for results_file in file_list:
            with results_file.open() as f:
                try:
                    results = json.load(f)
                except ValueError:
                    logger.error('Invalid JSON file: {}'.format(results_file))
                    continue

                for result in results:
241
242
243
244
245
                    try:
                        benchmark_name = result['name']
                    except Exception:
                        logger.error('Invalid content in JSON file: {}'.format(results_file))
                        continue
246
247
248
249
250
251
252
253
254
255
256
257
258
259
                    if results_file.parts[-3].endswith('_models'):
                        benchmark_name = '{}/{}'.format(results_file.parts[-3], result['name'])
                    if benchmark_name not in results_summary:
                        results_summary[benchmark_name] = defaultdict(list)
                    for metric in result['result']:
                        metric_name = '{}/{}'.format(benchmark_name, metric)
                        if metric_name not in reduce_ops:
                            reduce_ops[metric_name] = result['reduce_op'][metric]
                        elif reduce_ops[metric_name] != result['reduce_op'][metric]:
                            logger.error('Inconsistent reduce type for metric: {}'.format(metric_name))
                            continue

                        results_summary[benchmark_name][metric].append(result['result'][metric])

260
261
262
        results_summary = self.__merge_benchmark_metrics(results_summary, reduce_ops)
        monitor_summary = self.__merge_monitor_metrics(node_path)
        results_summary = {**results_summary, **monitor_summary}
263
264
265
266
267
        with (node_path / 'results-summary.json').open(mode='w') as f:
            json.dump(results_summary, f, indent=2)

        return results_summary

268
    def __merge_benchmark_metrics(self, results_summary, reduce_ops):
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
        """Merge metrics of all benchmarks in one node.

        Args:
            results_summary (dict): Summarized result of one node.
            reduce_ops (dict): The reduce type of each metric.

        Returns:
            dict: Flattened result with metric as key.
        """
        metrics_summary = dict()
        for benchmark_name in results_summary:
            for metric in results_summary[benchmark_name]:
                metric_name = '{}/{}'.format(benchmark_name, metric)
                if metric_name not in reduce_ops or (
                    reduce_ops[metric_name] is not None and reduce_ops[metric_name] not in ReduceType.get_values()
                ):
                    logger.error('Unknown reduce type for metric: {}'.format(metric_name))
                    continue

                if reduce_ops[metric_name] is not None:
                    reduce_func = Reducer.get_reduce_func(ReduceType(reduce_ops[metric_name]))
                    values = [reduce_func(list(result)) for result in zip(*results_summary[benchmark_name][metric])]
                    for run_count in range(len(values)):
                        if len(values) > 1:
                            metric_name = '{}/{}/{}'.format(benchmark_name, run_count, metric)
                        else:
                            metric_name = '{}/{}'.format(benchmark_name, metric)
                        metrics_summary[metric_name] = values[run_count]
                else:
                    for rank in range(len(results_summary[benchmark_name][metric])):
                        for run_count in range(len(results_summary[benchmark_name][metric][rank])):
                            if len(results_summary[benchmark_name][metric][rank]) > 1:
                                metric_name = '{}/{}/{}:{}'.format(benchmark_name, run_count, metric, rank)
                            else:
                                metric_name = '{}/{}:{}'.format(benchmark_name, metric, rank)
                            metrics_summary[metric_name] = results_summary[benchmark_name][metric][rank][run_count]

        return metrics_summary

308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
    def __merge_monitor_metrics(self, node_path):
        """Merge and summarize monitor metrics of one node.

        Args:
            node_path (Path): The Path instance of node directory.

        Returns:
            dict: Flattened result with metric as key.
        """
        metrics_summary = dict()
        all_samples = list()
        file_list = list(node_path.glob('**/monitor.jsonl'))
        for results_file in file_list:
            try:
                with jsonlines.open(results_file) as reader:
                    all_samples = list(reader)
            except BaseException as e:
                logger.error('Invalid Jsonline file: {}, error message: {}'.format(results_file, str(e)))
                continue
        all_samples = sorted(all_samples, key=lambda k: k.get('time', '0'))
        metrics_dict = dict()
        for sample in all_samples:
            for metric, value in sample.items():
                if metric not in metrics_dict:
                    metrics_dict[metric] = list()
                metrics_dict[metric].append(value)

        for metric, values in metrics_dict.items():
            for pattern, reduce_type in MonitorRecord.reduce_ops.items():
                if pattern in metric:
                    reduce_func = Reducer.get_reduce_func(reduce_type)
                    metrics_summary[metric] = reduce_func(values)
                    continue

        return metrics_summary

344
345
346
347
348
349
350
351
352
353
354
355
356
    def _run_proc(self, benchmark_name, mode, vars):
        """Run the process.

        Args:
            benchmark_name (str): Benchmark name.
            mode (DictConfig): Runner mode.
            vars (dict): Process variables.

        Returns:
            int: Process return code.
        """
        mode.update(vars)
        logger.info('Runner is going to run %s in %s mode, proc rank %d.', benchmark_name, mode.name, mode.proc_rank)
Yifan Xiong's avatar
Yifan Xiong committed
357
358
359
360
361
        ansible_runner_config = self._ansible_client.get_shell_config(
            (
                'docker exec sb-workspace bash -c '
                "'set -o allexport && source sb.env && set +o allexport && {command}'"
            ).format(command=self.__get_mode_command(benchmark_name, mode))
362
        )
Yifan Xiong's avatar
Yifan Xiong committed
363
364
365
        if mode.name == 'mpi':
            ansible_runner_config = self._ansible_client.update_mpi_config(ansible_runner_config)
        rc = self._ansible_client.run(ansible_runner_config, sudo=True)
366
367
        return rc

368
    def run(self):
369
370
371
372
373
374
        """Run the SuperBench benchmarks distributedly."""
        self.check_env()
        for benchmark_name in self._sb_benchmarks:
            if benchmark_name not in self._sb_enabled_benchmarks:
                continue
            benchmark_config = self._sb_benchmarks[benchmark_name]
375
376
377
378
379
380
            for mode in benchmark_config.modes:
                if mode.name == 'local':
                    Parallel(n_jobs=mode.proc_num if mode.parallel else 1)(
                        delayed(self._run_proc)(benchmark_name, mode, {
                            'proc_rank': proc_rank
                        }) for proc_rank in range(mode.proc_num)
381
                    )
Yifan Xiong's avatar
Yifan Xiong committed
382
                elif mode.name == 'torch.distributed' or mode.name == 'mpi':
383
                    self._run_proc(benchmark_name, mode, {'proc_rank': 0})
Yifan Xiong's avatar
Yifan Xiong committed
384
385
                else:
                    logger.warning('Unknown mode %s.', mode.name)
386
            self.fetch_results()
387
388

        self.__create_results_summary()