Test full async with NCCL communication.
(self)
| 1777 | |
| 1778 | class FullAsyncNCCLTest(BaseTrainerCase): |
| 1779 | def test_trainer(self): |
| 1780 | """Test full async with NCCL communication.""" |
| 1781 | self.config.cluster.node_num = 2 |
| 1782 | self.config.cluster.gpu_per_node = 2 |
| 1783 | self.config.algorithm.algorithm_type = "grpo" |
| 1784 | self.config.algorithm.advantage_fn = "grpo" |
| 1785 | self.config.algorithm.repeat_times = 4 |
| 1786 | self.config.buffer.batch_size = 4 |
| 1787 | self.config.buffer.explorer_input.taskset = get_unittest_dataset_config("gsm8k") |
| 1788 | self.config.explorer.rollout_model.tensor_parallel_size = 2 |
| 1789 | self.config.explorer.rollout_model.engine_num = 1 |
| 1790 | self.config.synchronizer.sync_interval = 1 |
| 1791 | self.config.synchronizer.sync_method = SyncMethod.NCCL |
| 1792 | self.config.synchronizer.sync_style = SyncStyle.FULLY_ASYNC |
| 1793 | self.config.check_and_update() |
| 1794 | both(self.config) |
| 1795 | parser = TensorBoardParser(os.path.join(self.config.monitor.cache_dir, "tensorboard")) |
| 1796 | rollout_metrics = parser.metric_list("rollout") |
| 1797 | self.assertGreater(len(rollout_metrics), 0) |
| 1798 | pipeline_metrics = parser.metric_list("experience_pipeline") |
| 1799 | self.assertGreater(len(pipeline_metrics), 0) |
| 1800 | self.assertEqual(parser.metric_max_step(rollout_metrics[0]), 8) |
| 1801 | # check model version |
| 1802 | model_versions = parser.metric_values("rollout/model_version") |
| 1803 | self.assertEqual(model_versions[0], 0) |
| 1804 | self.assertGreater(model_versions[-1], 0) |
nothing calls this directly
no test coverage detected