



import pytest

from datasets.utils.sharding import _distribute_shards, _number_of_shards_in_gen_kwargs, _split_gen_kwargs






@pytest.mark.parametrize(
    '''kwargs, expected''',    [
        ({'''num_shards''': 0, '''max_num_jobs''': 1}, []),
        ({'''num_shards''': 10, '''max_num_jobs''': 1}, [range(10						)]),
        ({'''num_shards''': 10, '''max_num_jobs''': 10}, [range(A_,    i + 1						) for i in range(10						)]),
        ({'''num_shards''': 1, '''max_num_jobs''': 10}, [range(1						)]),
        ({'''num_shards''': 10, '''max_num_jobs''': 3}, [range(0,    4						), range(4,    7						), range(7,    10						)]),
        ({'''num_shards''': 3, '''max_num_jobs''': 10}, [range(0,    1						), range(1,    2						), range(2,    3						)]),
    ],    )
def 	snake_case_  (				A_ :		Dict,    A_ :		int						):


       '''simple docstring'''





       _lowerCamelCase						:    Any         =			_distribute_shards(**A_						)
       assert out == expected






@pytest.mark.parametrize(
    '''gen_kwargs, max_num_jobs, expected''',    [
        ({'''foo''': 0}, 10, [{'''foo''': 0}]),
        ({'''shards''': [0, 1, 2, 3]}, 1, [{'''shards''': [0, 1, 2, 3]}]),
        ({'''shards''': [0, 1, 2, 3]}, 4, [{'''shards''': [0]}, {'''shards''': [1]}, {'''shards''': [2]}, {'''shards''': [3]}]),
        ({'''shards''': [0, 1]}, 4, [{'''shards''': [0]}, {'''shards''': [1]}]),
        ({'''shards''': [0, 1, 2, 3]}, 2, [{'''shards''': [0, 1]}, {'''shards''': [2, 3]}]),
    ],    )
def 	snake_case_  (				A_ :		str,    A_ :		Union[str, Any],    A_ :		Union[str, Any]						):


       '''simple docstring'''





       _lowerCamelCase						:    Tuple         =			_split_gen_kwargs(A_,    A_						)
       assert out == expected







@pytest.mark.parametrize(
    '''gen_kwargs, expected''',    [
        ({'''foo''': 0}, 1),
        ({'''shards''': [0]}, 1),
        ({'''shards''': [0, 1, 2, 3]}, 4),
        ({'''shards''': [0, 1, 2, 3], '''foo''': 0}, 4),
        ({'''shards''': [0, 1, 2, 3], '''other''': (0, 1)}, 4),
        ({'''shards''': [0, 1, 2, 3], '''shards2''': [0, 1]}, RuntimeError),
    ],    )
def 	snake_case_  (				A_ :		Union[str, Any],    A_ :		int						):


       '''simple docstring'''





       if expected is RuntimeError:
              with pytest.raises(A_						):
                     _number_of_shards_in_gen_kwargs(A_						)
       else:
              _lowerCamelCase						:    str         =			_number_of_shards_in_gen_kwargs(A_						)
              assert out == expected







"""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


lowerCAmelCase__      =    logging.getLogger(__name__)






class 	__snake_case	(     _lowercase):




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






         """simple docstring"""






         super().__init__(
             __lowerCAmelCase						, question_encoder_tokenizer=__lowerCAmelCase						, generator_tokenizer=__lowerCAmelCase						, index=__lowerCAmelCase						, init_retrieval=__lowerCAmelCase						, )
         _lowerCamelCase						:    Dict         =			None




  def 						SCREAMING_SNAKE_CASE    (			self					:						Dict						, __lowerCAmelCase					:						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
                _lowerCamelCase						:    List[str]         =			self._infer_socket_ifname()
                # avoid clash with the NCCL port
                _lowerCamelCase						:    Dict         =			str(distributed_port + 1		)
                _lowerCamelCase						:    str         =			dist.new_group(ranks=__lowerCAmelCase						, 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 						SCREAMING_SNAKE_CASE    (			self					:						Optional[Any]		):






         """simple docstring"""






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




  def 						SCREAMING_SNAKE_CASE    (			self					:						Union[str, Any]						, __lowerCAmelCase					:						Union[str, Any]						, __lowerCAmelCase					:						List[Any]						, __lowerCAmelCase					:						Tuple=torch.floataa		):






         """simple docstring"""






         _lowerCamelCase						:    Optional[Any]         =			torch.empty(__lowerCAmelCase						, dtype=__lowerCAmelCase		)
         dist.scatter(__lowerCAmelCase						, src=0						, scatter_list=__lowerCAmelCase						, group=self.process_group		)
         return target_tensor




  def 						SCREAMING_SNAKE_CASE    (			self					:						Any		):






         """simple docstring"""






         _lowerCamelCase						:    List[str]         =			psutil.net_if_addrs()
         # a hacky way to deal with varying network interface names
         _lowerCamelCase						:    str         =			next((addr for addr in addrs if addr.startswith('''e'''		))						, __lowerCAmelCase		)
         return ifname




  def 						SCREAMING_SNAKE_CASE    (			self					:						str						, __lowerCAmelCase					:						np.ndarray						, __lowerCAmelCase					:						int		):






         """simple docstring"""






         if not dist.is_initialized():
                _lowerCamelCase      ,   _lowerCamelCase						:    Any         =			self._main_retrieve(__lowerCAmelCase						, __lowerCAmelCase		)
                return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(__lowerCAmelCase		)

         # distributed training
         _lowerCamelCase						:    Dict         =			dist.get_world_size(group=self.process_group		)

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

         # scatter logic
         _lowerCamelCase						:    int         =			question_hidden_states.shape[0]
         _lowerCamelCase						:    str         =			[]
         _lowerCamelCase						:    Optional[int]         =			[]
         if self._is_main():
                assert len(__lowerCAmelCase		) == world_size
                _lowerCamelCase      ,   _lowerCamelCase						:    Tuple         =			self._main_retrieve(torch.cat(__lowerCAmelCase		).numpy()						, __lowerCAmelCase		)
                _lowerCamelCase      ,   _lowerCamelCase						:    List[str]         =			torch.tensor(__lowerCAmelCase		), torch.tensor(__lowerCAmelCase		)
                _lowerCamelCase						:    List[Any]         =			self._chunk_tensor(__lowerCAmelCase						, __lowerCAmelCase		)
                _lowerCamelCase						:    Optional[Any]         =			self._chunk_tensor(__lowerCAmelCase						, __lowerCAmelCase		)
         _lowerCamelCase						:    int         =			self._scattered(__lowerCAmelCase						, [n_queries, n_docs]						, target_type=torch.intaa		)
         _lowerCamelCase						:    str         =			self._scattered(__lowerCAmelCase						, [n_queries, n_docs, question_hidden_states.shape[1]]		)

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


