Diff Coverage

Diff: origin/master...HEAD, staged and unstaged changes

Source File Diff Coverage (%) Missing Lines
hyper_parallel/trainer/base.py 62.7% 295-296,299-304,306-310,599,1013-1014,1027,1032,1035,1037,1039-1040
hyper_parallel/trainer/callbacks/__init__.py 100%  
hyper_parallel/trainer/callbacks/profiler_callback.py 89.7% 80,107,196,200,202,206,216-217,219-220,230,234,240,242,244-245,270,289,336-337
hyper_parallel/trainer/config/trainer.py 100%  
hyper_parallel/trainer/config/training.py 100%  
hyper_parallel/trainer/runtime/memory_profiler.py 96.3% 78,154-156,185,191,205
hyper_parallel/trainer/text_trainer.py 33.3% 55-56,59-61,64-65,68-70,72-77,325-327,329-338,341,343-346
hyper_parallel/trainer/vlm_trainer.py 34.0% 53-54,57-59,62-63,66-68,70-74,294,296-297,299-300,302-308,311,313-315,317-318
hyper_parallel/trainer/base.py
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
            #   - consume the ``DistributedSetup`` created above;
            #   - set ``model`` and the resolved checkpoint ``model_config``;
            #   - return a materialized, weight-loaded, already-parallelized model;
            #   - never call a second trainer-side ``build_parallelize_model``.
            self._build_model()
            self._build_loss()

            # Build trainer-owned data components after the model finalizes parameters and sharding.
            self._build_model_assets()
            self._build_data_transform()
            self._build_dataset()
            self._build_collate_fn()
            self._build_dataloader()
            self._compute_train_iters()

            self._build_optimizer()
            self.model_integration.attach_optimizer(self.optimizer)
            self._build_lr_scheduler()
            self._build_training_context()
            self._init_callbacks()
        except BaseException:
            self.abort_profilers()
            raise
595
596
597
598
599
600
601
602
603
        self.tqdm_callback = TqdmCallback(self)
        self.logging_callback = LoggingCallback(self)
        self.evaluate_callback = EvaluateCallback(self)
        self.garbage_collection_callback = GarbageCollectionCallback(self)
        self.profiler_callback = ProfilerCallback(self)
        self._callbacks = [
            self.environ_meter_callback,
            self.logging_callback,
            self.tqdm_callback,
1009
1010
1011
1012
1013
1014
1015
1016
1017
            try:
                start_epoch = self.state.epoch
                for epoch in range(start_epoch, self.train_epochs):
                    if epoch != start_epoch:
                        self.train_dataloader.set_epoch(epoch)
                        self.data_iterator = HyperIter(
                            self.train_dataloader, use_background_prefetcher=config.dataloader.use_background_prefetcher
                        )
                    self.state.epoch = epoch
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
                    for _ in range(start_step, train_steps):
                        try:
                            self.train_step(self.data_iterator)
                        except StopIteration:
                            logger.info(
                                "epoch:%s Dataloader finished with drop_last %s",
                                epoch,
                                config.dataloader.drop_last,
                            )
                            break

                    self.on_epoch_end()
                    self.state.epoch = epoch + 1

                    print_device_mem_info(f"VRAM usage after epoch {epoch + 1}")

                    if config.dataloader.use_background_prefetcher:
                        self.data_iterator.stop()

                memory_profiler.stop()
                self.on_train_end()
            finally:
hyper_parallel/trainer/callbacks/profiler_callback.py
76
77
78
79
80
81
82
83
84
            return

        self._device_type, self._profiler_module = self._resolve_profiler_module()
        if self.config.mstx and self._device_type != "npu":
            raise ValueError("profiler.mstx is supported only when training on Ascend NPU")

        output_root = Path(self.config.output_path or "./output").expanduser()
        self.output_path = output_root / "profile" / f"rank_{rank}"
        logger.info("Profile save path: %s", self.output_path)
103
104
105
106
107
108
109
110
111
            **kwargs: Additional callback context supplied by the trainer.
        """
        del state, kwargs
        if self.profiler is None or not self.start_on_init:
            return
        self._start()
        self._record_metadata()

    def on_step_begin(
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
            "mstx",
        )
        for name in boolean_fields:
            if not isinstance(getattr(self.config, name), bool):
                raise TypeError(f"profiler.{name} must be a bool")
        for name in ("start_step", "stop_step"):
            value = getattr(self.config, name)
            if isinstance(value, bool) or not isinstance(value, int):
                raise TypeError(f"profiler.{name} must be an int")
        if self.config.output_path is not None and not isinstance(self.config.output_path, str):
            raise TypeError("profiler.output_path must be a string or null")
        ranks = self.config.rank_ids
        if ranks is not None:
            if not isinstance(ranks, list):
                raise TypeError("profiler.rank_ids must be a list or null")
            if any(isinstance(rank, bool) or not isinstance(rank, int) or not 0 <= rank < self.trainer.world_size
                   for rank in ranks):
                raise ValueError(f"profiler.rank_ids must contain only integers in [0, {self.trainer.world_size})")
212
213
214
215
216
217
218
219
220
221
222
223
224
        """Normalize the inclusive profiling window to valid positive steps."""
        start_step = self.config.start_step
        stop_step = self.config.stop_step
        if start_step < 1:
            logger.warning("profiler.start_step must be greater than 0; reset it to 1")
            start_step = 1
        if stop_step < 1:
            logger.warning("profiler.stop_step must be greater than 0; reset it to 10")
            stop_step = 10
        if start_step > stop_step:
            logger.warning(
                "profiler.stop_step must be greater than or equal to profiler.start_step; reset both to 1 and 10"
            )
226
227
228
229
230
231
232
233
234
235
236
237
238
        return start_step, stop_step

    def _normalize_start_on_init(self) -> bool:
        if self.start_step != 1 and self.config.start_on_init:
            logger.warning(
                "profiler.start_step and profiler.start_on_init cannot take effect simultaneously; "
                "reset profiler.start_on_init to false"
            )
            return False
        return self.config.start_on_init

    def _normalize_level(self) -> int:
        level = self.config.level
236
237
238
239
240
241
242
243
244
245
246
247
248
249

    def _normalize_level(self) -> int:
        level = self.config.level
        if level is None:
            return 0
        if isinstance(level, bool) or not isinstance(level, int):
            raise TypeError("profiler.level must be an int or null")
        if level not in (0, 1, 2):
            logger.warning("Invalid profiler.level %s; reset it to 0", level)
            return 0
        return level

    def _get_schedule(self) -> Any:
        if self.start_on_init:
266
267
268
269
270
271
272
273
274
        pipeline_ranks: list[int] = []
        if self.config.pipeline_stage_leaders:
            pipeline_stages = max(1, self.trainer.config.accelerator.pp_size)
            if self.trainer.world_size % pipeline_stages != 0:
                raise ValueError("device count must be divisible by pipeline stage count")
            devices_per_stage = self.trainer.world_size // pipeline_stages
            pipeline_ranks = [stage * devices_per_stage for stage in range(pipeline_stages)]

        if not ranks and not pipeline_ranks:
285
286
287
288
289
290
291
292
293
            self._torch_npu = torch_npu
            return device_type, torch_npu.profiler
        if device_type in ("cpu", "cuda"):
            return device_type, torch.profiler
        raise RuntimeError(f"profiler does not support device type {device_type!r}")

    def _get_activities(self) -> list[Any]:
        activities = [self._profiler_module.ProfilerActivity.CPU]
        if self._device_type == "cuda":
332
333
334
335
336
337
338
339
340
341
        }
        try:
            self.profiler.add_metadata_json("distributed_args", json.dumps(metadata))
            self._metadata_recorded = True
        except AttributeError as error:
            logger.warning("Profiler failed to record distributed args: %s", error)

    def _close_mstx_range(self) -> None:
        if self._mstx_range_id is None:
            return
hyper_parallel/trainer/runtime/memory_profiler.py
74
75
76
77
78
79
80
81
82
            dp_rank: Rank in the data-parallel group.
        """
        self._discard_previous_session()
        if config is None:
            return

        self._validate(config)
        self.enable = config.enable
        self.mem_info = config.mem_info
150
151
152
153
154
155
156
157
158
159
160
        self.enable = False
        self.mem_info = False
        try:
            self._stop_history()
        except Exception:  # pylint: disable=broad-exception-caught
            logger.exception("Failed to stop allocator memory history while aborting training.")
            return

        self._deactivate_session()

    def _discard_previous_session(self) -> None:
181
182
183
184
185
186
187
188
189

    @staticmethod
    def _validate_bool(value: bool, name: str) -> None:
        if not isinstance(value, bool):
            raise TypeError(f"memory.{name} must be a bool")

    @staticmethod
    def _validate_steps(start_step: int, end_step: int) -> None:
        for name, value in (("start_step", start_step), ("end_step", end_step)):
187
188
189
190
191
192
193
194
195
    @staticmethod
    def _validate_steps(start_step: int, end_step: int) -> None:
        for name, value in (("start_step", start_step), ("end_step", end_step)):
            if isinstance(value, bool) or not isinstance(value, int):
                raise TypeError(f"memory.{name} must be an int")
        if start_step < 0:
            raise ValueError("memory.start_step must be non-negative")
        if end_step < start_step:
            raise ValueError("memory.end_step must be greater than or equal to memory.start_step")
201
202
203
204
205
206
207
208
209

    @staticmethod
    def _validate_dump_ranks(dump_ranks: list[int]) -> None:
        if not isinstance(dump_ranks, list):
            raise ValueError("memory.dump_ranks must be a list of non-negative integers")
        if any(
                isinstance(rank, bool) or not isinstance(rank, int) or rank < 0
                for rank in dump_ranks
        ):
hyper_parallel/trainer/text_trainer.py
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
                global_rank=self.base.global_rank,
                tp_rank=self.base.mesh.tp_rank,
                dp_rank=self.base.mesh.dp_rank,
            )
            self.base._build_model()
            self.base._build_loss()

            # datasets
            self._build_model_assets()
            self._build_data_transform()
            self.base._build_dataset()

            # dataloader
            self._build_collate_fn()
            self.base._build_dataloader()

            # get_batch
            self._build_get_batch()
            self.base.attach_model_integration_data_pipeline()
            self.base._compute_train_iters()

            self.base._build_optimizer()
            self.base.model_integration.attach_optimizer(self.base.optimizer)
            self.base._build_lr_scheduler()
            self.base._build_training_context()
            self.base._build_graph_compiler()
            self.base._init_callbacks()
        except BaseException:
            self.base.abort_profilers()
            raise
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
            )

            # Checkpoint resume restores state.global_step, state.epoch, and the DataLoader cursor.
            for epoch in range(self.base.state.epoch, self.base.train_epochs):
                train_dataloader = self.base.train_dataloader
                if hasattr(train_dataloader, "set_epoch"):
                    train_dataloader.set_epoch(epoch)

                self.base.state.epoch = epoch
                self.on_epoch_begin()
                data_iterator = iter(train_dataloader) if train_dataloader is not None else None
                start_step = self.base.state.global_step - epoch * self.base.train_steps
                train_steps = min(self.base.train_steps, self.base.train_iters - epoch * self.base.train_steps)
                for _ in range(start_step, train_steps):
                    try:
                        self.train_step(data_iterator)
                    except StopIteration:
                        logger.info(
                            "epoch:%s Dataloader finished with drop_last %s", epoch, config.dataloader.drop_last
                        )
                        break

                self.on_epoch_end()
                self.base.state.epoch = epoch + 1
                print_device_mem_info(f"VRAM usage after epoch {epoch + 1}")
                if self.base.state.global_step >= self.base.train_iters:
                    break

            memory_profiler.stop()
            self.on_train_end()
hyper_parallel/trainer/vlm_trainer.py
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
                global_rank=self.base.global_rank,
                tp_rank=self.base.mesh.tp_rank,
                dp_rank=self.base.mesh.dp_rank,
            )
            self.base._build_model()
            self.base._build_loss()

            # datasets
            self._build_model_assets()
            self._build_data_transform()
            self.base._build_dataset()

            # dataloader
            self._build_collate_fn()
            self.base._build_dataloader()

            # get_batch
            self._build_get_batch()
            self.base.attach_model_integration_data_pipeline()
            self.base._compute_train_iters()

            self.base._build_optimizer()
            self.base.model_integration.attach_optimizer(self.base.optimizer)
            self.base._build_lr_scheduler()
            self.base._build_training_context()
            self.base._init_callbacks()
        except BaseException:
            self.base.abort_profilers()
            raise
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
            )

            # Checkpoint resume restores state.global_step, state.epoch, and the DataLoader cursor.
            for epoch in range(self.base.state.epoch, self.base.train_epochs):
                train_dataloader = self.base.train_dataloader

                if hasattr(train_dataloader, "set_epoch"):
                    train_dataloader.set_epoch(epoch)

                self.on_epoch_begin()
                data_iterator = iter(train_dataloader) if train_dataloader is not None else None

                start_step = self.base.state.global_step - epoch * self.base.train_steps
                train_steps = min(self.base.train_steps, self.base.train_iters - epoch * self.base.train_steps)
                for _ in range(start_step, train_steps):
                    try:
                        self.train_step(data_iterator)
                    except StopIteration:
                        logger.info(
                            "epoch:%s Dataloader finished with drop_last %s", epoch, config.dataloader.drop_last
                        )
                        break

                self.on_epoch_end()
                self.base.state.epoch = epoch + 1
                print_device_mem_info(f"VRAM usage after epoch {epoch + 1}")

                if self.base.state.global_step >= self.base.train_iters:
                    break

            memory_profiler.stop()
            self.on_train_end()