code: import signal
  import time
  from typing import Any, Callable
  
  import torch
  from easydict import EasyDict
  from .time_helper_base import TimeWrapper
  from .time_helper_cuda import get_cuda_time_wrapper
  
  
  def build_time_helper(cfg: EasyDict = None, wrapper_type: str = None) -> Callable[[], 'TimeWrapper']:
      r"""
      Overview:
          Build the timehelper
  
      Arguments:
          - cfg (:obj:`dict`):
              The config file, which is a multilevel dict, have large domain like
              evaluate, common, model, train etc, and each large domain
              has it's smaller domain.
          - wrapper_type (:obj:`str`): The type of wrapper returned, support ``['time', 'cuda']``
  
      Returns:
          - time_wrapper (:obj:`TimeWrapper`):
              Return the corresponding timewrapper, Reference: ``ding.utils.timehelper.TimeWrapperTime``
              and ``ding.utils.timehelper.get_cuda_time_wrapper``.
      """
      # Note: wrapper_type has higher priority
      if wrapper_type is not None:
          time_wrapper_type = wrapper_type
      elif cfg is not None:
          time_wrapper_type = cfg.common.time_wrapper_type
      else:
          raise RuntimeError('Either wrapper_type or cfg should be provided.')
  
      if time_wrapper_type == 'time':
          return TimeWrapperTime
      elif time_wrapper_type == 'cuda':
          if torch.cuda.is_available():
              # lazy initialize to make code runnable locally
              return get_cuda_time_wrapper()
          else:
              return TimeWrapperTime
      else:
          raise KeyError('invalid time_wrapper_type: {}'.format(time_wrapper_type))
  
  
  class EasyTimer:
      r"""
      Overview:
          A decent timer wrapper that can be used easily.
  
      Interface:
          ``__init__``, ``__enter__``, ``__exit__``
  
      Example:
          >>> wait_timer = EasyTimer()
          >>> with wait_timer:
          >>>    func(...)
          >>> time_ = wait_timer.value  # in second
      """
  
      def __init__(self, cuda=True):
          r"""
          Overview:
              Init class EasyTimer
  
          Arguments:
              - cuda (:obj:`bool`): Whether to build timer with cuda type
          """
          if torch.cuda.is_available() and cuda:
              time_wrapper_type = "cuda"
          else:
              time_wrapper_type = "time"
          self._timer = build_time_helper(wrapper_type=time_wrapper_type)
          self.value = 0.0
  
      def __enter__(self):
          r"""
          Overview:
              Enter timer, start timing
          """
          self.value = 0.0
          self._timer.start_time()
  
      def __exit__(self, *args):
          r"""
          Overview:
              Exit timer, stop timing
          """
          self.value = self._timer.end_time()
  
  
  class TimeWrapperTime(TimeWrapper):
      r"""
      Overview:
          A class method that inherit from ``TimeWrapper`` class
  
      Interface:
          ``start_time``, ``end_time``
      """
  
      # overwrite
      @classmethod
      def start_time(cls):
          r"""
          Overview:
              Implement and overide the ``start_time`` method in ``TimeWrapper`` class
          """
          cls.start = time.time()
  
      # overwrite
      @classmethod
      def end_time(cls):
          r"""
          Overview:
              Implement and overide the end_time method in ``TimeWrapper`` class
  
          Returns:
              - time(:obj:`float`): The time between ``start_time`` and end_time
          """
          cls.end = time.time()
          return cls.end - cls.start
  
  
  class WatchDog(object):
      """
      Overview:
          Simple watchdog timer to detect timeouts
  
      Arguments:
          - timeout (:obj:`int`): Timeout value of the ``watchdog [seconds]``.
  
      .. note::
              If it is not reset before exceeding this value, ``TimeourError`` raised.
  
      Interface:
          ``start``, ``stop``
  
      Examples:
          >>> watchdog = WatchDog(x) # x is a timeout value
          >>> ...
          >>> watchdog.start()
          >>> ... # Some function
  
      """
  
      def __init__(self, timeout: int = 1):
          self._timeout = timeout + 1
          self._failed = False
  
      def start(self):
          r"""
          Overview:
              Start watchdog.
          """
          signal.signal(signal.SIGALRM, self._event)
          signal.alarm(self._timeout)
  
      @staticmethod
      def _event(signum: Any, frame: Any):
          raise TimeoutError()
  
      def stop(self):
          r"""
          Overview:
              Stop watchdog with ``alarm(0)``, ``SIGALRM``, and ``SIG_DFL`` signals.
          """
          signal.alarm(0)
          signal.signal(signal.SIGALRM, signal.SIG_DFL)
   import time
  import torch
  from hpc_rll.origin.rnn import get_lstm
  from hpc_rll.torch_utils.network.rnn import LSTM
  from testbase import mean_relative_error, times
  
  assert torch.cuda.is_available()
  use_cuda = True
  
  seq_len = 64
  batch_size = 3
  input_size = 1792
  hidden_size = 384
  num_layers = 3
  norm_type = 'LN'
  dropout = 0  # 0.1
  
  
  # Note: need open load_params for hpc_lstm to validation
  # Note: only used to case of num_layers = 3
  def lstm_val():
      ori_lstm = get_lstm('normal', input_size, hidden_size, num_layers, norm_type, dropout)
      hpc_lstm = LSTM(seq_len, batch_size, input_size, hidden_size, num_layers, norm_type, dropout)
  
      ori_x = torch.randn(seq_len, batch_size, input_size)
      ori_h0 = torch.randn(num_layers, batch_size, hidden_size)
      ori_c0 = torch.randn(num_layers, batch_size, hidden_size)
  
      if use_cuda:
          ori_x = ori_x.cuda()
          ori_h0 = ori_h0.cuda()
          ori_c0 = ori_c0.cuda()
          ori_lstm = ori_lstm.cuda()
          hpc_lstm = hpc_lstm.cuda()
  
      ori_x.requires_grad_(True)
      ori_output, ori_next_state = ori_lstm(ori_x, [ori_h0, ori_c0])
      ori_loss = ori_output.mean()
      ori_loss.backward()
  
      hpc_x = ori_x.clone().detach()
      hpc_h0 = ori_h0.clone().detach()
      hpc_c0 = ori_c0.clone().detach()
      hpc_x.requires_grad_(True)
      hpc_output, hpc_next_state = hpc_lstm(hpc_x, [hpc_h0, hpc_c0])
      hpc_loss = hpc_output.mean()
      hpc_loss.backward()
      torch.cuda.synchronize()
  
      mre = mean_relative_error(
          torch.flatten(ori_loss).cpu().detach().numpy(),
          torch.flatten(hpc_loss).cpu().detach().numpy()
      )
      print("lstm fp mean_relative_error: " + str(mre))
      mre = mean_relative_error(
          torch.flatten(ori_x.grad).cpu().detach().numpy(),
          torch.flatten(hpc_x.grad).cpu().detach().numpy()
      )
      print("lstm bp mean_relative_error: " + str(mre))
  
      ori_wx_grad = torch.cat((ori_lstm.wx[0].grad, ori_lstm.wx[1].grad, ori_lstm.wx[2].grad))
      hpc_wx_grad = hpc_lstm.wx.grad
      mre = mean_relative_error(torch.flatten(ori_wx_grad).cpu().numpy(), torch.flatten(hpc_wx_grad).cpu().numpy())
      print("wx grad mean_relative_error: " + str(mre))
  
      ori_wh_grad = torch.cat((ori_lstm.wh[0].grad, ori_lstm.wh[1].grad, ori_lstm.wh[2].grad))
      hpc_wh_grad = hpc_lstm.wh.grad
      mre = mean_relative_error(torch.flatten(ori_wh_grad).cpu().numpy(), torch.flatten(hpc_wh_grad).cpu().numpy())
      print("wh grad mean_relative_error: " + str(mre))
  
      ori_bias_grad = ori_lstm.bias.grad
      hpc_bias_grad = hpc_lstm.bias.grad
      mre = mean_relative_error(torch.flatten(ori_bias_grad).cpu().numpy(), torch.flatten(hpc_bias_grad).cpu().numpy())
      print("bias grad mean_relative_error: " + str(mre))
  
      params = list(ori_lstm.parameters())
      gamma_0_x = params[1]
      beta_0_x = params[2]
      gamma_0_h = params[3]
      beta_0_h = params[4]
      gamma_1_x = params[5]
      beta_1_x = params[6]
      gamma_1_h = params[7]
      beta_1_h = params[8]
      gamma_2_x = params[9]
      beta_2_x = params[10]
      gamma_2_h = params[11]
      beta_2_h = params[12]
      ori_gamma_grad = torch.cat(
          (gamma_0_x.grad, gamma_0_h.grad, gamma_1_x.grad, gamma_1_h.grad, gamma_2_x.grad, gamma_2_h.grad)
      )
      ori_beta_grad = torch.cat(
          (beta_0_x.grad, beta_0_h.grad, beta_1_x.grad, beta_1_h.grad, beta_2_x.grad, beta_2_h.grad)
      )
      hpc_gamma_grad = hpc_lstm.ln_gamma.grad
      hpc_beta_grad = hpc_lstm.ln_beta.grad
      mre = mean_relative_error(torch.flatten(ori_gamma_grad).cpu().numpy(), torch.flatten(hpc_gamma_grad).cpu().numpy())
      print("ln gamma grad mean_relative_error: " + str(mre))
      mre = mean_relative_error(torch.flatten(ori_beta_grad).cpu().numpy(), torch.flatten(hpc_beta_grad).cpu().numpy())
      print("ln beta grad mean_relative_error: " + str(mre))
  
  
  def lstm_perf():
      ori_lstm = get_lstm('normal', input_size, hidden_size, num_layers, norm_type, dropout)
      hpc_lstm = LSTM(seq_len, batch_size, input_size, hidden_size, num_layers, norm_type, dropout)
  
      lstms = {'normal': ori_lstm, 'hpc': hpc_lstm}
  
      for lstm_type, lstm in lstms.items():
          x = torch.rand(seq_len, batch_size, input_size)
          h0 = torch.randn(num_layers, batch_size, hidden_size)
          c0 = torch.randn(num_layers, batch_size, hidden_size)
          if use_cuda:
              x = x.cuda()
              h0 = h0.cuda()
              c0 = c0.cuda()
              lstm = lstm.cuda()
  
          prev_state = [h0, c0]
          x.requires_grad_(True)
          for i in range(times):
              t = time.time()
              output, _ = lstm(x, prev_state)
              loss = output.mean()
              loss.backward()
              if use_cuda:
                  torch.cuda.synchronize()
              print('epoch: {}, {} lstm cost time: {}'.format(i, lstm_type, time.time() - t))
  
  
  if __name__ == '__main__':
      print(
          "target problem: seq_len = {}, batch_size = {}, input_size = {}, hidden_size = {}, num_layers = {}, norm_type = {}, dropout = {}"  # noqa
          .format(seq_len, batch_size, input_size, hidden_size, num_layers, norm_type, dropout)
      )
      print("==============lstm has no validation test================")
      #print("===============run lstm validation test==================")
      #lstm_val()
      print("===============run lstm performance test=================")
      lstm_perf()
  
apis: torch.cuda.is_available
apis: torch.randn torch.flatten torch.rand torch.cuda.synchronize torch.cuda.is_available torch.cat