





import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized, parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
      from sagemaker import Session, TrainingJobAnalytics
      from sagemaker.huggingface import HuggingFace




@pytest.mark.skipif(
    literal_eval(os.getenv('''TEST_SAGEMAKER'''							,		'''False'''       )       ) is not True							,		reason='''Skipping test because should only be run when releasing minor transformers version'''							,		)
@pytest.mark.usefixtures('''sm_env'''       )
@parameterized_class(
    [
        {
            '''framework''': '''pytorch''',
            '''script''': '''run_glue_model_parallelism.py''',
            '''model_name_or_path''': '''roberta-large''',
            '''instance_type''': '''ml.p3dn.24xlarge''',
            '''results''': {'''train_runtime''': 1600, '''eval_accuracy''': 0.3, '''eval_loss''': 1.2},
        },
        {
            '''framework''': '''pytorch''',
            '''script''': '''run_glue.py''',
            '''model_name_or_path''': '''roberta-large''',
            '''instance_type''': '''ml.p3dn.24xlarge''',
            '''results''': {'''train_runtime''': 1600, '''eval_accuracy''': 0.3, '''eval_loss''': 1.2},
        },
    ]       )
class 			UpperCAmelCase					(      unittest.TestCase       ):




      def     lowercase__   (      self      :			Dict  )    ->				Optional[Any]:
            if self.framework == "pytorch":
                  subprocess.run(
                      f"cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py".split()      ,      encoding="""utf-8"""      ,      check=__snake_case      ,      )
            assert hasattr(self      ,      """env"""  )




      def     lowercase__   (      self      :			Optional[int]      ,      __snake_case      :			Tuple  )    ->				Any:
            # configuration for running training on smdistributed Model Parallel
            _lowerCAmelCase												=							{
                """enabled""": True,
                """processes_per_host""": 8,
            }
            _lowerCAmelCase												=							{
                """enabled""": True,
                """parameters""": {
                    """microbatches""": 4,
                    """placement_strategy""": """spread""",
                    """pipeline""": """interleaved""",
                    """optimize""": """speed""",
                    """partitions""": 4,
                    """ddp""": True,
                },
            }

            _lowerCAmelCase												=							{"""smdistributed""": {"""modelparallel""": smp_options}, """mpi""": mpi_options}

            _lowerCAmelCase												=							"""trainer""" if self.script == """run_glue.py""" else """smtrainer"""
            # creates estimator
            return HuggingFace(
                entry_point=self.script      ,      source_dir=self.env.test_path      ,      role=self.env.role      ,      image_uri=self.env.image_uri      ,      base_job_name=f"{self.env.base_job_name}-{instance_count}-smp-{name_extension}"      ,      instance_count=__snake_case      ,      instance_type=self.instance_type      ,      debugger_hook_config=__snake_case      ,      hyperparameters={
                    **self.env.hyperparameters,
                    """model_name_or_path""": self.model_name_or_path,
                    """max_steps""": 5_00,
                }      ,      metric_definitions=self.env.metric_definitions      ,      distribution=__snake_case      ,      py_version="""py36"""      ,      )




      def     lowercase__   (      self      :			int      ,      __snake_case      :			Optional[int]  )    ->				Union[str, Any]:
            TrainingJobAnalytics(__snake_case  ).export_csv(f"{self.env.test_path}/{job_name}_metrics.csv"  )




      @parameterized.expand([(1,)]  )
      def     lowercase__   (      self      :			str      ,      __snake_case      :			str  )    ->				List[Any]:
            # create estimator
            _lowerCAmelCase												=							self.create_estimator(__snake_case  )

            # run training
            estimator.fit()

            # result dataframe
            _lowerCAmelCase												=							TrainingJobAnalytics(estimator.latest_training_job.name  ).dataframe()

            # extract kpis
            _lowerCAmelCase												=							list(result_metrics_df[result_metrics_df.metric_name == """eval_accuracy"""]["""value"""]  )
            _lowerCAmelCase												=							list(result_metrics_df[result_metrics_df.metric_name == """eval_loss"""]["""value"""]  )
            # get train time from SageMaker job, this includes starting, preprocessing, stopping
            _lowerCAmelCase												=							(
                Session().describe_training_job(estimator.latest_training_job.name  ).get("""TrainingTimeInSeconds"""      ,      99_99_99  )
            )

            # assert kpis
            assert train_runtime <= self.results["train_runtime"]
            assert all(t >= self.results["""eval_accuracy"""] for t in eval_accuracy  )
            assert all(t <= self.results["""eval_loss"""] for t in eval_loss  )

            # dump tests result into json file to share in PR
            with open(f"{estimator.latest_training_job.name}.json"      ,      """w"""  ) as outfile:
                  json.dump({"""train_time""": train_runtime, """eval_accuracy""": eval_accuracy, """eval_loss""": eval_loss}      ,      __snake_case  )













'''simple docstring'''






import logging
import os
import threading
import time


try:
      import warnings
except ImportError:
      A__    :       Tuple              =None

try:
      import msvcrt
except ImportError:
      A__    :       str              =None

try:
      import fcntl
except ImportError:
      A__    :       Dict              =None


# Backward compatibility
# ------------------------------------------------
try:
      TimeoutError
except NameError:
      A__    :       Tuple              =OSError


# Data
# ------------------------------------------------
A__    :       Optional[Any]              =[
    '''Timeout''',
    '''BaseFileLock''',
    '''WindowsFileLock''',
    '''UnixFileLock''',
    '''SoftFileLock''',
    '''FileLock''',
]

A__    :       Union[str, Any]              ='''3.0.12'''


A__    :       Optional[Any]              =None
def        UpperCamelCase__				(		):





      """simple docstring"""


      global _logger
      _lowerCAmelCase												=							_logger or logging.getLogger(__name__  )
      return _logger




class 			UpperCAmelCase					(      snake_case_       ):




      def __init__(      self      :			Optional[int]      ,      __snake_case      :			str  )    ->				Optional[Any]:
            _lowerCAmelCase												=							lock_file
            return None




      def __str__(      self      :			Union[str, Any]  )    ->				Any:
            _lowerCAmelCase												=							f"The file lock '{self.lock_file}' could not be acquired."
            return temp




class 			UpperCAmelCase					:




      def __init__(      self      :			str      ,      __snake_case      :			str  )    ->				Tuple:
            _lowerCAmelCase												=							lock
            return None




      def __enter__(      self      :			Union[str, Any]  )    ->				Union[str, Any]:
            return self.lock




      def __exit__(      self      :			Tuple      ,      __snake_case      :			List[Any]      ,      __snake_case      :			int      ,      __snake_case      :			Optional[int]  )    ->				Union[str, Any]:
            self.lock.release()
            return None




class 			UpperCAmelCase					:




      def __init__(      self      :			List[Any]      ,      __snake_case      :			str      ,      __snake_case      :			Tuple=-1      ,      __snake_case      :			List[str]=None  )    ->				Any:
            _lowerCAmelCase												=							max_filename_length if max_filename_length is not None else 2_55
            # Hash the filename if it's too long
            _lowerCAmelCase												=							self.hash_filename_if_too_long(__snake_case      ,      __snake_case  )
            # The path to the lock file.
            _lowerCAmelCase												=							lock_file

            # The file descriptor for the *_lock_file* as it is returned by the
            # os.open() function.
            # This file lock is only NOT None, if the object currently holds the
            # lock.
            _lowerCAmelCase												=							None

            # The default timeout value.
            _lowerCAmelCase												=							timeout

            # We use this lock primarily for the lock counter.
            _lowerCAmelCase												=							threading.Lock()

            # The lock counter is used for implementing the nested locking
            # mechanism. Whenever the lock is acquired, the counter is increased and
            # the lock is only released, when this value is 0 again.
            _lowerCAmelCase												=							0
            return None




      @property
      def     lowercase__   (      self      :			int  )    ->				List[str]:
            return self._lock_file




      @property
      def     lowercase__   (      self      :			Dict  )    ->				List[Any]:
            return self._timeout




      @timeout.setter
      def     lowercase__   (      self      :			Tuple      ,      __snake_case      :			int  )    ->				Optional[Any]:
            _lowerCAmelCase												=							float(__snake_case  )
            return None




      def     lowercase__   (      self      :			Optional[int]  )    ->				Union[str, Any]:
            raise NotImplementedError()




      def     lowercase__   (      self      :			Optional[Any]  )    ->				Any:
            raise NotImplementedError()




      @property
      def     lowercase__   (      self      :			str  )    ->				Dict:
            return self._lock_file_fd is not None




      def     lowercase__   (      self      :			Union[str, Any]      ,      __snake_case      :			Union[str, Any]=None      ,      __snake_case      :			List[str]=0.05  )    ->				List[str]:
            # Use the default timeout, if no timeout is provided.
            if timeout is None:
                  _lowerCAmelCase												=							self.timeout

            # Increment the number right at the beginning.
            # We can still undo it, if something fails.
            with self._thread_lock:
                  self._lock_counter += 1

            _lowerCAmelCase												=							id(self  )
            _lowerCAmelCase												=							self._lock_file
            _lowerCAmelCase												=							time.time()
            try:
                  while True:
                        with self._thread_lock:
                              if not self.is_locked:
                                    logger().debug(f"Attempting to acquire lock {lock_id} on {lock_filename}"  )
                                    self._acquire()

                        if self.is_locked:
                              logger().debug(f"Lock {lock_id} acquired on {lock_filename}"  )
                              break
                        elif timeout >= 0 and time.time() - start_time > timeout:
                              logger().debug(f"Timeout on acquiring lock {lock_id} on {lock_filename}"  )
                              raise Timeout(self._lock_file  )
                        else:
                              logger().debug(
                                  f"Lock {lock_id} not acquired on {lock_filename}, waiting {poll_intervall} seconds ..."  )
                              time.sleep(__snake_case  )
            except:  # noqa
                  # Something did go wrong, so decrement the counter.
                  with self._thread_lock:
                        _lowerCAmelCase												=							max(0      ,      self._lock_counter - 1  )

                  raise
            return _Acquire_ReturnProxy(lock=self  )




      def     lowercase__   (      self      :			Dict      ,      __snake_case      :			Optional[int]=False  )    ->				Union[str, Any]:
            with self._thread_lock:
                  if self.is_locked:
                        self._lock_counter -= 1

                        if self._lock_counter == 0 or force:
                              _lowerCAmelCase												=							id(self  )
                              _lowerCAmelCase												=							self._lock_file

                              logger().debug(f"Attempting to release lock {lock_id} on {lock_filename}"  )
                              self._release()
                              _lowerCAmelCase												=							0
                              logger().debug(f"Lock {lock_id} released on {lock_filename}"  )

            return None




      def __enter__(      self      :			List[str]  )    ->				Union[str, Any]:
            self.acquire()
            return self




      def __exit__(      self      :			Union[str, Any]      ,      __snake_case      :			str      ,      __snake_case      :			Tuple      ,      __snake_case      :			Any  )    ->				int:
            self.release()
            return None




      def __del__(      self      :			Optional[Any]  )    ->				Any:
            self.release(force=__snake_case  )
            return None




      def     lowercase__   (      self      :			Optional[Any]      ,      __snake_case      :			str      ,      __snake_case      :			int  )    ->				str:
            _lowerCAmelCase												=							os.path.basename(__snake_case  )
            if len(__snake_case  ) > max_length and max_length > 0:
                  _lowerCAmelCase												=							os.path.dirname(__snake_case  )
                  _lowerCAmelCase												=							str(hash(__snake_case  )  )
                  _lowerCAmelCase												=							filename[: max_length - len(__snake_case  ) - 8] + """...""" + hashed_filename + """.lock"""
                  return os.path.join(__snake_case      ,      __snake_case  )
            else:
                  return path




class 			UpperCAmelCase					(      snake_case_       ):




      def __init__(      self      :			str      ,      __snake_case      :			Union[str, Any]      ,      __snake_case      :			Optional[int]=-1      ,      __snake_case      :			str=None  )    ->				Dict:
            from .file_utils import relative_to_absolute_path

            super().__init__(__snake_case      ,      timeout=__snake_case      ,      max_filename_length=__snake_case  )
            _lowerCAmelCase												=							"""\\\\?\\""" + relative_to_absolute_path(self.lock_file  )




      def     lowercase__   (      self      :			List[str]  )    ->				Tuple:
            _lowerCAmelCase												=							os.O_RDWR | os.O_CREAT | os.O_TRUNC

            try:
                  _lowerCAmelCase												=							os.open(self._lock_file      ,      __snake_case  )
            except OSError:
                  pass
            else:
                  try:
                        msvcrt.locking(__snake_case      ,      msvcrt.LK_NBLCK      ,      1  )
                  except OSError:
                        os.close(__snake_case  )
                  else:
                        _lowerCAmelCase												=							fd
            return None




      def     lowercase__   (      self      :			List[str]  )    ->				List[str]:
            _lowerCAmelCase												=							self._lock_file_fd
            _lowerCAmelCase												=							None
            msvcrt.locking(__snake_case      ,      msvcrt.LK_UNLCK      ,      1  )
            os.close(__snake_case  )

            try:
                  os.remove(self._lock_file  )
            # Probably another instance of the application
            # that acquired the file lock.
            except OSError:
                  pass
            return None




class 			UpperCAmelCase					(      snake_case_       ):




      def __init__(      self      :			Dict      ,      __snake_case      :			List[Any]      ,      __snake_case      :			List[Any]=-1      ,      __snake_case      :			List[str]=None  )    ->				Dict:
            _lowerCAmelCase												=							os.statvfs(os.path.dirname(__snake_case  )  ).f_namemax
            super().__init__(__snake_case      ,      timeout=__snake_case      ,      max_filename_length=__snake_case  )




      def     lowercase__   (      self      :			Optional[int]  )    ->				List[Any]:
            _lowerCAmelCase												=							os.O_RDWR | os.O_CREAT | os.O_TRUNC
            _lowerCAmelCase												=							os.open(self._lock_file      ,      __snake_case  )

            try:
                  fcntl.flock(__snake_case      ,      fcntl.LOCK_EX | fcntl.LOCK_NB  )
            except OSError:
                  os.close(__snake_case  )
            else:
                  _lowerCAmelCase												=							fd
            return None




      def     lowercase__   (      self      :			Optional[int]  )    ->				Optional[int]:
            # Do not remove the lockfile:
            #
            #   https://github.com/benediktschmitt/py-filelock/issues/31
            #   https://stackoverflow.com/questions/17708885/flock-removing-locked-file-without-race-condition
            _lowerCAmelCase												=							self._lock_file_fd
            _lowerCAmelCase												=							None
            fcntl.flock(__snake_case      ,      fcntl.LOCK_UN  )
            os.close(__snake_case  )
            return None




class 			UpperCAmelCase					(      snake_case_       ):




      def     lowercase__   (      self      :			Union[str, Any]  )    ->				Dict:
            _lowerCAmelCase												=							os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
            try:
                  _lowerCAmelCase												=							os.open(self._lock_file      ,      __snake_case  )
            except OSError:
                  pass
            else:
                  _lowerCAmelCase												=							fd
            return None




      def     lowercase__   (      self      :			Any  )    ->				Optional[Any]:
            os.close(self._lock_file_fd  )
            _lowerCAmelCase												=							None

            try:
                  os.remove(self._lock_file  )
            # The file is already deleted and that's what we want.
            except OSError:
                  pass
            return None




A__    :       Tuple              =None

if msvcrt:
      A__    :       List[Any]              =WindowsFileLock
elif fcntl:
      A__    :       Tuple              =UnixFileLock
else:
      A__    :       Tuple              =SoftFileLock

      if warnings is not None:
            warnings.warn('''only soft file lock is available''')






