


#
# This a `torch.distributed` diagnostics script that checks that all GPUs in the cluster (one or
# many nodes) can talk to each other via nccl and allocate gpu memory.
#
# To run first adjust the number of processes and nodes:
#
# python -m torch.distributed.run --nproc_per_node 2 --nnodes 1 torch-distributed-gpu-test.py
#
# You may need to add --master_addr $MASTER_ADDR --master_port $MASTER_PORT if using a custom addr:port
#
# You can also use the rdzv API: --rdzv_endpoint $MASTER_ADDR:$MASTER_PORT --rdzv_backend c10d
#
# use torch.distributed.launch instead of torch.distributed.run for torch < 1.9
#
# If you get a hanging in `barrier` calls you have some network issues, you may try to debug this with:
#
# NCCL_DEBUG=INFO python -m torch.distributed.run --nproc_per_node 2 --nnodes 1 torch-distributed-gpu-test.py
#
# which should tell you what's going on behind the scenes.
#
#
# This script can be run via `srun` in the SLURM environment as well. Here is a SLURM script that
# runs on 2 nodes of 4 gpus per node:
#
# #SBATCH --job-name=test-nodes        # name
# #SBATCH --nodes=2                    # nodes
# #SBATCH --ntasks-per-node=1          # crucial - only 1 task per dist per node!
# #SBATCH --cpus-per-task=10           # number of cores per tasks
# #SBATCH --gres=gpu:4                 # number of gpus
# #SBATCH --time 0:05:00               # maximum execution time (HH:MM:SS)
# #SBATCH --output=%x-%j.out           # output file name
#
# GPUS_PER_NODE=4
# MASTER_ADDR=$(scontrol show hostnames $SLURM_JOB_NODELIST | head -n 1)
# MASTER_PORT=6000
#
# srun --jobid $SLURM_JOBID bash -c 'python -m torch.distributed.run \
# --nproc_per_node $GPUS_PER_NODE --nnodes $SLURM_NNODES --node_rank $SLURM_PROCID \
# --master_addr $MASTER_ADDR --master_port $MASTER_PORT \
# torch-distributed-gpu-test.py'
#

import fcntl
import os
import socket

import torch
import torch.distributed as dist


def 				lowerCamelCase__	(      *a	:   List[str]				)    ->	Dict:





			"""simple docstring"""
			with open(__A	,  "r"				) as fh:
						fcntl.flock(__A	,  fcntl.LOCK_EX				)
						try:
									print(*__A				)
						finally:
									fcntl.flock(__A	,  fcntl.LOCK_UN				)


snake_case__			 =						int(os.environ['''LOCAL_RANK'''])
torch.cuda.set_device(local_rank)
snake_case__			 =						torch.device('''cuda''', local_rank)
snake_case__			 =						socket.gethostname()

snake_case__			 =						f'''[{hostname}-{local_rank}]'''

try:
		# test distributed
		dist.init_process_group('''nccl''')
		dist.all_reduce(torch.ones(1).to(device), op=dist.ReduceOp.SUM)
		dist.barrier()

		# test cuda is available and can allocate memory
		torch.cuda.is_available()
		torch.ones(1).cuda(local_rank)

		# global rank
		snake_case__			 =						dist.get_rank()
		snake_case__			 =						dist.get_world_size()

		printflock(f'''{gpu} is OK (global rank: {rank}/{world_size})''')

		dist.barrier()
		if rank == 0:
				printflock(f'''pt={torch.__version__}, cuda={torch.version.cuda}, nccl={torch.cuda.nccl.version()}''')

except Exception:
		printflock(f'''{gpu} is broken''')
		raise






import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionTextToImagePipeline
from diffusers.utils.testing_utils import nightly, require_torch_gpu, torch_device


snake_case__			 =						False


class 					lowerCAmelCase_ (       unittest.TestCase):
						pass






@nightly
@require_torch_gpu
class 					lowerCAmelCase_ (       unittest.TestCase):




						def  _snake_case					(	self						:   Tuple						)      ->Optional[int]:






									"""simple docstring"""






									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()




						def  _snake_case					(	self						:   Dict						)      ->Any:






									"""simple docstring"""






									a__ :Union[str, Any]						  =    VersatileDiffusionTextToImagePipeline.from_pretrained("shi-labs/versatile-diffusion"						)
									# remove text_unet
									pipe.remove_unused_weights()
									pipe.to(__A						)
									pipe.set_progress_bar_config(disable=__A						)

									a__ :List[Any]						  =    "A painting of a squirrel eating a burger "
									a__ :Optional[Any]						  =    torch.manual_seed(0						)
									a__ :List[Any]						  =    pipe(
									    prompt=__A   ,     generator=__A   ,     guidance_scale=7.5   ,     num_inference_steps=2   ,     output_type="numpy"						).images

									with tempfile.TemporaryDirectory() as tmpdirname:
												pipe.save_pretrained(__A						)
												a__ :List[Any]						  =    VersatileDiffusionTextToImagePipeline.from_pretrained(__A						)
									pipe.to(__A						)
									pipe.set_progress_bar_config(disable=__A						)

									a__ :Optional[int]						  =    generator.manual_seed(0						)
									a__ :List[Any]						  =    pipe(
									    prompt=__A   ,     generator=__A   ,     guidance_scale=7.5   ,     num_inference_steps=2   ,     output_type="numpy"						).images

									assert np.abs(image - new_image						).sum() < 1E-5, "Models don't have the same forward pass"






						def  _snake_case					(	self						:   Optional[Any]						)      ->List[Any]:






									"""simple docstring"""






									a__ :Tuple						  =    VersatileDiffusionTextToImagePipeline.from_pretrained(
									    "shi-labs/versatile-diffusion"   ,     torch_dtype=torch.floataa						)
									pipe.to(__A						)
									pipe.set_progress_bar_config(disable=__A						)

									a__ :Tuple						  =    "A painting of a squirrel eating a burger "
									a__ :Tuple						  =    torch.manual_seed(0						)
									a__ :Optional[Any]						  =    pipe(
									    prompt=__A   ,     generator=__A   ,     guidance_scale=7.5   ,     num_inference_steps=50   ,     output_type="numpy"						).images

									a__ :Tuple						  =    image[0, 253:256, 253:256, -1]

									assert image.shape == (1, 512, 512, 3)
									a__ :Tuple						  =    np.array([0.3_367, 0.3_169, 0.2_656, 0.3_870, 0.4_790, 0.3_796, 0.4_009, 0.4_878, 0.4_778]						)

									assert np.abs(image_slice.flatten() - expected_slice						).max() < 1E-2


