def METHOD_NAME(
    script: Union[bytes, str, BinaryIO, TextIO],
    n_workers: int,
    data: Dict[str, TileableType] = None,
    gpu: Optional[bool] = None,
    command_argv: List[str] = None,
    retry_when_fail: bool = False,
    session: SessionType = None,
    run_kwargs: Dict[str, Any] = None,
    port: int = None,
):
    """
    Run PyTorch script in Mars cluster.
    Parameters
    ----------
    script: str or file-like object
        Script to run
    n_workers : int
        Number of PyTorch workers
    data : dict
        Variable name to data.
    gpu : bool
        Run PyTorch script on GPU
    command_argv : list
        Extra command args for script
    retry_when_fail : bool
        If True, retry when function failed.
    session
        Mars session, if not provided, will use default one.
    run_kwargs : dict
        Extra kwargs for `session.run`.
    port : int
        Port of PyTorch worker or ps, will automatically increase for the same worker
    Returns
    -------
    status
        return {'status': 'ok'} if succeeded, or error raised
    """
    if int(n_workers) <= 0:
        raise ValueError("n_workers should be at least 1")
    if hasattr(script, "read"):
        code = script.read()
    else:
        with open(os.path.abspath(script), "rb") as f:
            code = f.read()
    inputs = _extract_inputs(data)
    port = 29500 if port is None else port
    op = RunPyTorch(
        data=data,
        code=to_binary(code),
        world_size=int(n_workers),
        retry_when_fail=retry_when_fail,
        gpu=gpu,
        master_port=port,
        command_args=command_argv,
    )
    return op(inputs).execute(session=session, **(run_kwargs or {}))