(self,
model: str = None,
preprocessor: Union[Preprocessor, List[Preprocessor]] = None,
auto_collate=True,
**kwargs)
| 460 | """ |
| 461 | |
| 462 | def __init__(self, |
| 463 | model: str = None, |
| 464 | preprocessor: Union[Preprocessor, List[Preprocessor]] = None, |
| 465 | auto_collate=True, |
| 466 | **kwargs): |
| 467 | # DistributedPipeline uses classmethod to initialize model |
| 468 | # without calling super().__init__ method |
| 469 | self.preprocessor = preprocessor |
| 470 | self._model_prepare = False |
| 471 | self._model_prepare_lock = Lock() |
| 472 | self._auto_collate = auto_collate |
| 473 | |
| 474 | if os.path.exists(model): |
| 475 | self.model_dir = model |
| 476 | else: |
| 477 | self.model_dir = snapshot_download(model) |
| 478 | self.cfg = read_config(self.model_dir) |
| 479 | self.world_size = self._get_world_size(self.cfg) |
| 480 | self.model_pool = None |
| 481 | self.device_name = 'cpu' |
| 482 | self.device = create_device(self.device_name) |
| 483 | self.has_multiple_models = False |
| 484 | self.framework = self.cfg.framework |
| 485 | torch.multiprocessing.set_start_method('spawn', force=True) |
| 486 | |
| 487 | ranks = list(range(self.world_size)) |
| 488 | self.model_pool = Pool(self.world_size) |
| 489 | |
| 490 | if 'master_ip' not in kwargs: |
| 491 | kwargs['master_ip'] = '127.0.0.1' |
| 492 | master_port = int(kwargs['master_port'] |
| 493 | ) if 'master_port' in kwargs else random.randint( |
| 494 | 29500, 39500) |
| 495 | from modelscope.utils.torch_utils import _find_free_port, _is_free_port |
| 496 | if not _is_free_port(master_port): |
| 497 | master_port = _find_free_port() |
| 498 | kwargs['master_port'] = str(master_port) |
| 499 | # TODO: Pass ip and port to megatron_util for initialization |
| 500 | os.environ['MASTER_ADDR'] = kwargs['master_ip'] |
| 501 | os.environ['MASTER_PORT'] = kwargs['master_port'] |
| 502 | |
| 503 | self.model_pool.map( |
| 504 | partial( |
| 505 | self.__class__._instantiate_one, |
| 506 | model_dir=self.model_dir, |
| 507 | **self.cfg.model, |
| 508 | **kwargs), ranks) |
| 509 | self.models = [] |
| 510 | |
| 511 | def __del__(self): |
| 512 | if hasattr(self, 'model_pool') and self.model_pool is not None: |
nothing calls this directly
no test coverage detected