


# This code is adapted from OpenAI's release
# https://github.com/openai/human-eval/blob/master/human_eval/execution.py

import contextlib
import faulthandler
import io
import multiprocessing
import os
import platform
import signal
import tempfile




def        snake_case_					(  snake_case      ,							snake_case      ,							snake_case      ,							snake_case					)  ->    Union[str, Any]:
       lowercase__:       Tuple        =       multiprocessing.Manager()
       lowercase__:       Any        =       manager.list()

       lowercase__:       List[Any]        =       multiprocessing.Process(target=snake_case      ,							args=(check_program, result, timeout)					)
       p.start()
       p.join(timeout=timeout + 1					)
       if p.is_alive():
              p.kill()

       if not result:
              result.append('timed out'					)

       return {
           "task_id": task_id,
           "passed": result[0] == "passed",
           "result": result[0],
           "completion_id": completion_id,
       }




def        snake_case_					(  snake_case      ,							snake_case      ,							snake_case					)  ->    Optional[Any]:
       with create_tempdir():
              # These system calls are needed when cleaning up tempdir.
              import os
              import shutil

              lowercase__:       Union[str, Any]        =       shutil.rmtree
              lowercase__:       List[Any]        =       os.rmdir
              lowercase__:       Optional[Any]        =       os.chdir

              # Disable functionalities that can make destructive changes to the test.
              reliability_guard()

              # Run program.
              try:
                     lowercase__:       str        =       {}
                     with swallow_io():
                            with time_limit(snake_case					):
                                   exec(snake_case      ,							snake_case					)
                     result.append('passed'					)
              except TimeoutException:
                     result.append('timed out'					)
              except BaseException as e:
                     result.append(f'failed: {e}'					)

              # Needed for cleaning up.
              lowercase__:       Optional[Any]        =       rmtree
              lowercase__:       Tuple        =       rmdir
              lowercase__:       Union[str, Any]        =       chdir




@contextlib.contextmanager
def        snake_case_					(  snake_case					)  ->    int:
       def signal_handler(snake_case      ,							snake_case					):
              raise TimeoutException('Timed out!'					)

       signal.setitimer(signal.ITIMER_REAL      ,							snake_case					)
       signal.signal(signal.SIGALRM      ,							snake_case					)
       try:
              yield
       finally:
              signal.setitimer(signal.ITIMER_REAL      ,							0					)




@contextlib.contextmanager
def        snake_case_					(  )  ->    Optional[Any]:
       lowercase__:       Any        =       WriteOnlyStringIO()
       with contextlib.redirect_stdout(snake_case					):
              with contextlib.redirect_stderr(snake_case					):
                     with redirect_stdin(snake_case					):
                            yield




@contextlib.contextmanager
def        snake_case_					(  )  ->    Dict:
       with tempfile.TemporaryDirectory() as dirname:
              with chdir(snake_case					):
                     yield dirname





class   __a      (   __UpperCamelCase				):
 pass





class   __a      (   io.StringIO				):
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			*lowerCAmelCase__    ,			**lowerCAmelCase__			) ->    Optional[Any]:




        '''simple docstring'''
        raise OSError
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			*lowerCAmelCase__    ,			**lowerCAmelCase__			) ->    int:




        '''simple docstring'''
        raise OSError
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			*lowerCAmelCase__    ,			**lowerCAmelCase__			) ->    str:




        '''simple docstring'''
        raise OSError
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			*lowerCAmelCase__    ,			**lowerCAmelCase__			) ->    Union[str, Any]:




        '''simple docstring'''
        return False





class   __a      (   contextlib._RedirectStream				):  # type: ignore
 __lowercase			:							str       					=			'stdin'





@contextlib.contextmanager
def        snake_case_					(  snake_case					)  ->    str:
       if root == ".":
              yield
              return
       lowercase__:       Optional[Any]        =       os.getcwd()
       os.chdir(snake_case					)
       try:
              yield
       except BaseException as exc:
              raise exc
       finally:
              os.chdir(snake_case					)




def        snake_case_					(  snake_case=None					)  ->    int:

       if maximum_memory_bytes is not None:
              import resource

              resource.setrlimit(resource.RLIMIT_AS      ,							(maximum_memory_bytes, maximum_memory_bytes)					)
              resource.setrlimit(resource.RLIMIT_DATA      ,							(maximum_memory_bytes, maximum_memory_bytes)					)
              if not platform.uname().system == "Darwin":
                     resource.setrlimit(resource.RLIMIT_STACK      ,							(maximum_memory_bytes, maximum_memory_bytes)					)

       faulthandler.disable()

       import builtins

       lowercase__:       str        =       None
       lowercase__:       Dict        =       None

       import os

       lowercase__:       str        =       '1'

       lowercase__:       Tuple        =       None
       lowercase__:       Optional[Any]        =       None
       lowercase__:       Dict        =       None
       lowercase__:       List[Any]        =       None
       lowercase__:       str        =       None
       lowercase__:       str        =       None
       lowercase__:       Optional[int]        =       None
       lowercase__:       Optional[Any]        =       None
       lowercase__:       Optional[int]        =       None
       lowercase__:       List[str]        =       None
       lowercase__:       Union[str, Any]        =       None
       lowercase__:       Dict        =       None
       lowercase__:       List[str]        =       None
       lowercase__:       Optional[Any]        =       None
       lowercase__:       List[str]        =       None
       lowercase__:       Optional[int]        =       None
       lowercase__:       Tuple        =       None
       lowercase__:       Dict        =       None
       lowercase__:       List[str]        =       None
       lowercase__:       Union[str, Any]        =       None
       lowercase__:       Union[str, Any]        =       None
       lowercase__:       Optional[int]        =       None
       lowercase__:       Optional[int]        =       None
       lowercase__:       str        =       None
       lowercase__:       Any        =       None
       lowercase__:       List[Any]        =       None
       lowercase__:       int        =       None

       import shutil

       lowercase__:       Optional[int]        =       None
       lowercase__:       Any        =       None
       lowercase__:       int        =       None

       import subprocess

       lowercase__:       str        =       None  # type: ignore

       lowercase__:       List[str]        =       None

       import sys

       lowercase__:       int        =       None
       lowercase__:       Tuple        =       None
       lowercase__:       Optional[Any]        =       None
       lowercase__:       Any        =       None
       lowercase__:       str        =       None






from typing import Dict

from transformers import EvalPrediction, HfArgumentParser, TrainingArguments, is_torch_available
from transformers.testing_utils import (
    TestCasePlus,
    execute_subprocess_async,
    get_torch_dist_unique_port,
    require_torch_multi_gpu,
    require_torch_neuroncore,
)
from transformers.training_args import ParallelMode
from transformers.utils import logging


__lowerCAmelCase			=							logging.get_logger(__name__)


if is_torch_available():
 import torch
 from torch import nn
 from torch.utils.data import Dataset

 from transformers import Trainer





 class   __a      (   __UpperCamelCase				):
  def __init__(       self    ,			lowerCAmelCase__ = 101			) ->    Any:




         '''simple docstring'''
         lowercase__:       Any        =       length
  def __len__(       self			) ->    List[Any]:




         '''simple docstring'''
         return self.length
  def __getitem__(       self    ,			lowerCAmelCase__			) ->    int:




         '''simple docstring'''
         return i





 class   __a      :
  def __call__(       self    ,			lowerCAmelCase__			) ->    List[str]:




         '''simple docstring'''
         return {"input_ids": torch.tensor(lowerCAmelCase__			), "labels": torch.tensor(lowerCAmelCase__			)}





 class   __a      (   nn.Module				):
  def __init__(       self			) ->    Tuple:




         '''simple docstring'''
         super().__init__()
         # Add some (unused) params otherwise DDP will complain.
         lowercase__:       List[str]        =       nn.Linear(120    ,			80			)
  def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__=None			) ->    int:




         '''simple docstring'''
         if labels is not None:
                return torch.tensor(0.0    ,			device=input_ids.device			), input_ids
         else:
                return input_ids





class   __a      (   __UpperCamelCase				):
 @require_torch_neuroncore
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Tuple:




        '''simple docstring'''
        lowercase__:       int        =       F'--nproc_per_node=2\n            --master_port={get_torch_dist_unique_port()}\n            {self.test_file_dir}/test_trainer_distributed.py\n        '.split()
        lowercase__:       Tuple        =       self.get_auto_remove_tmp_dir()
        lowercase__:       Optional[int]        =       F'--output_dir {output_dir}'.split()
        lowercase__:       int        =       ['torchrun'] + distributed_args + args
        execute_subprocess_async(lowerCAmelCase__    ,			env=self.get_env()			)
        # successful return here == success - any errors would have caused an error in the sub-call





class   __a      (   __UpperCamelCase				):
 @require_torch_multi_gpu
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Any:




        '''simple docstring'''
        lowercase__:       List[str]        =       F'--nproc_per_node={torch.cuda.device_count()}\n            --master_port={get_torch_dist_unique_port()}\n            {self.test_file_dir}/test_trainer_distributed.py\n        '.split()
        lowercase__:       Tuple        =       self.get_auto_remove_tmp_dir()
        lowercase__:       List[str]        =       F'--output_dir {output_dir}'.split()
        lowercase__:       int        =       ['torchrun'] + distributed_args + args
        execute_subprocess_async(lowerCAmelCase__    ,			env=self.get_env()			)
        # successful return here == success - any errors would have caused an error in the sub-call


if __name__ == "__main__":
 # The script below is meant to be run under torch.distributed, on a machine with multiple GPUs:
 #
 # PYTHONPATH="src" python -m torch.distributed.run --nproc_per_node 2 --output_dir output_dir ./tests/test_trainer_distributed.py

 __lowerCAmelCase			=							HfArgumentParser((TrainingArguments,))
 __lowerCAmelCase			=							parser.parse_args_into_dataclasses()[0]

 logger.warning(
     F'''Process rank: {training_args.local_rank}, device: {training_args.device}, n_gpu: {training_args.n_gpu}, '''
     F'''distributed training: {training_args.parallel_mode != ParallelMode.NOT_DISTRIBUTED}'''
 )

 # Essentially, what we want to verify in the distributed case is that we get all samples back,
 # in the right order. (this is crucial for prediction for instance)
 for dataset_length in [1_01, 40, 7]:
  __lowerCAmelCase			=							DummyDataset(dataset_length)




  def        snake_case_					(  snake_case					)  ->    Dict:
         lowercase__:       str        =       list(range(len(snake_case					)					)					)
         lowercase__:       Tuple        =       p.predictions.tolist() == sequential and p.label_ids.tolist() == sequential
         if not success and training_args.local_rank == 0:
                logger.warning(
                    'Predictions and/or labels do not match expected results:\n  - predictions: '
                    f'{p.predictions.tolist()}\n  - labels: {p.label_ids.tolist()}\n  - expected: {sequential}'					)
         return {"success": success}

  __lowerCAmelCase			=							Trainer(
      model=DummyModel(),
      args=training_args,
      data_collator=DummyDataCollator(),
      eval_dataset=dataset,
      compute_metrics=compute_metrics,
  )
  __lowerCAmelCase			=							trainer.evaluate()
  logger.info(metrics)
  if metrics["eval_success"] is not True:
   logger.error(metrics)
   exit(1)

  __lowerCAmelCase			=							trainer.predict(dataset)
  logger.info(p.metrics)
  if p.metrics["test_success"] is not True:
   logger.error(p.metrics)
   exit(1)

  __lowerCAmelCase			=							2

  __lowerCAmelCase			=							trainer.evaluate()
  logger.info(metrics)
  if metrics["eval_success"] is not True:
   logger.error(metrics)
   exit(1)

  __lowerCAmelCase			=							trainer.predict(dataset)
  logger.info(p.metrics)
  if p.metrics["test_success"] is not True:
   logger.error(p.metrics)
   exit(1)




  __lowerCAmelCase			=							None


