





"""simple docstring"""




import logging
import os
from typing import List, Tuple

import numpy as np
import psutil
import torch
import torch.distributed as dist

from transformers import RagRetriever


_snake_case           =			logging.getLogger(__name__)





class  lowerCAmelCase							(       lowerCamelCase__ ):







  def __init__(      self  :Tuple   ,							_lowercase  :Optional[int]   ,							_lowercase  :List[Any]   ,							_lowercase  :Union[str, Any]   ,							_lowercase  :Union[str, Any]=None ):






     '''simple docstring'''





     super().__init__(
         _lowercase   ,							question_encoder_tokenizer=_lowercase   ,							generator_tokenizer=_lowercase   ,							index=_lowercase   ,							init_retrieval=_lowercase   ,							)
     lowercase__			       =  None







  def   UpperCAmelCase					(      self  :int   ,							_lowercase  :int ):






     '''simple docstring'''





     logger.info("initializing retrieval" )

     # initializing a separate process group for retrieval as the default
     # nccl backend doesn't support gather/scatter operations while gloo
     # is too slow to replace nccl for the core gpu communication
     if dist.is_initialized():
        logger.info("dist initialized" )
        # needs to be set manually
        lowercase__			       =  self._infer_socket_ifname()
        # avoid clash with the NCCL port
        lowercase__			       =  str(distributed_port + 1 )
        lowercase__			       =  dist.new_group(ranks=_lowercase   ,							backend="gloo" )

     # initialize retriever only on the main worker
     if not dist.is_initialized() or self._is_main():
        logger.info("dist not initialized / main" )
        self.index.init_index()

     # all processes wait untill the retriever is initialized by the main process
     if dist.is_initialized():
        torch.distributed.barrier(group=self.process_group )







  def   UpperCAmelCase					(      self  :Optional[Any] ):






     '''simple docstring'''





     return dist.get_rank(group=self.process_group ) == 0







  def   UpperCAmelCase					(      self  :List[str]   ,							_lowercase  :Optional[int]   ,							_lowercase  :Tuple   ,							_lowercase  :List[Any]=torch.floataa ):






     '''simple docstring'''





     lowercase__			       =  torch.empty(_lowercase   ,							dtype=_lowercase )
     dist.scatter(_lowercase   ,							src=0   ,							scatter_list=_lowercase   ,							group=self.process_group )
     return target_tensor







  def   UpperCAmelCase					(      self  :Any ):






     '''simple docstring'''





     lowercase__			       =  psutil.net_if_addrs()
     # a hacky way to deal with varying network interface names
     lowercase__			       =  next((addr for addr in addrs if addr.startswith("e" ))   ,							_lowercase )
     return ifname







  def   UpperCAmelCase					(      self  :Tuple   ,							_lowercase  :np.ndarray   ,							_lowercase  :int ):






     '''simple docstring'''





     if not dist.is_initialized():
        lowercase__			       =  self._main_retrieve(_lowercase   ,							_lowercase )
        return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(_lowercase )

     # distributed training
     lowercase__			       =  dist.get_world_size(group=self.process_group )

     # gather logic
     lowercase__			       =  None
     if self._is_main():
        lowercase__			       =  [torch.empty(question_hidden_states.shape   ,							dtype=torch.floataa ) for _ in range(_lowercase )]
     dist.gather(torch.tensor(_lowercase )   ,							dst=0   ,							gather_list=_lowercase   ,							group=self.process_group )

     # scatter logic
     lowercase__			       =  question_hidden_states.shape[0]
     lowercase__			       =  []
     lowercase__			       =  []
     if self._is_main():
        assert len(_lowercase ) == world_size
        lowercase__			       =  self._main_retrieve(torch.cat(_lowercase ).numpy()   ,							_lowercase )
        lowercase__			       =  torch.tensor(_lowercase ), torch.tensor(_lowercase )
        lowercase__			       =  self._chunk_tensor(_lowercase   ,							_lowercase )
        lowercase__			       =  self._chunk_tensor(_lowercase   ,							_lowercase )
     lowercase__			       =  self._scattered(_lowercase   ,							[n_queries, n_docs]   ,							target_type=torch.intaa )
     lowercase__			       =  self._scattered(_lowercase   ,							[n_queries, n_docs, question_hidden_states.shape[1]] )

     return retrieved_doc_embeds.numpy(), doc_ids.numpy(), self.index.get_doc_dicts(_lowercase )













import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
  from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
  import torch_xla.core.xla_model as xm


def _A       (   __magic_name__	):
   if is_torch_version("<"  , "2.0.0"	) or not hasattr(__magic_name__  , "_dynamo"	):
      return False
   return isinstance(__magic_name__  , torch._dynamo.eval_frame.OptimizedModule	)


def _A       (   __magic_name__  , __magic_name__ = True	):
   lowercase__			       =  (torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

   lowercase__			       =  is_compiled_module(__magic_name__	)
   if is_compiled:
      lowercase__			       =  model
      lowercase__			       =  model._orig_mod

   if is_deepspeed_available():
      options += (DeepSpeedEngine,)

   while isinstance(__magic_name__  , __magic_name__	):
      lowercase__			       =  model.module

   if not keep_fpaa_wrapper:
      lowercase__			       =  getattr(__magic_name__  , "forward"	)
      lowercase__			       =  model.__dict__.pop("_original_forward"  , __magic_name__	)
      if original_forward is not None:
         while hasattr(__magic_name__  , "__wrapped__"	):
            lowercase__			       =  forward.__wrapped__
            if forward == original_forward:
               break
         lowercase__			       =  forward
      if getattr(__magic_name__  , "_converted_to_transformer_engine"  , __magic_name__	):
         convert_model(__magic_name__  , to_transformer_engine=__magic_name__	)

   if is_compiled:
      lowercase__			       =  model
      lowercase__			       =  compiled_model

   return model


def _A       (   ):
   PartialState().wait_for_everyone()


def _A       (   __magic_name__  , __magic_name__	):
   if PartialState().distributed_type == DistributedType.TPU:
      xm.save(__magic_name__  , __magic_name__	)
   elif PartialState().local_process_index == 0:
      torch.save(__magic_name__  , __magic_name__	)


@contextmanager
def _A       (   **__magic_name__	):
   for key, value in kwargs.items():
      lowercase__			       =  str(__magic_name__	)

   yield

   for key in kwargs:
      if key.upper() in os.environ:
         del os.environ[key.upper()]


def _A       (   __magic_name__	):
   if not hasattr(__magic_name__  , "__qualname__"	) and not hasattr(__magic_name__  , "__name__"	):
      lowercase__			       =  getattr(__magic_name__  , "__class__"  , __magic_name__	)
   if hasattr(__magic_name__  , "__qualname__"	):
      return obj.__qualname__
   if hasattr(__magic_name__  , "__name__"	):
      return obj.__name__
   return str(__magic_name__	)


def _A       (   __magic_name__  , __magic_name__	):
   for key, value in source.items():
      if isinstance(__magic_name__  , __magic_name__	):
         lowercase__			       =  destination.setdefault(__magic_name__  , {}	)
         merge_dicts(__magic_name__  , __magic_name__	)
      else:
         lowercase__			       =  value

   return destination







def _A       (   __magic_name__ = None	):
   if port is None:
      lowercase__			       =  2_9500
   with socket.socket(socket.AF_INET  , socket.SOCK_STREAM	) as s:
      return s.connect_ex(("localhost", port)	) == 0






