

import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


_SCREAMING_SNAKE_CASE					:  str             =      logging.get_logger(__name__)






class 	UpperCAmelCase__		:






  """simple docstring"""


  a									= None



@experimental
def     UpperCAmelCase_       (						_A				,		_A				,		_A				,		_A				,		_A				,		_A				,		_A    ):





     '''simple docstring'''




     if ParallelBackendConfig.backend_name is None:
          return _map_with_multiprocessing_pool(
              _A				,		_A				,		_A				,		_A				,		_A				,		_A				,		_A    )

     return _map_with_joblib(_A				,		_A				,		_A				,		_A				,		_A				,		_A				,		_A    )
def     UpperCAmelCase_       (						_A				,		_A				,		_A				,		_A				,		_A				,		_A				,		_A    ):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__   =   num_proc if num_proc <= len(_A    ) else len(_A    )
     SCREAMING_SNAKE_CASE__   =   []  # We organize the splits ourselve (contiguous splits)
     for index in range(_A    ):
          SCREAMING_SNAKE_CASE__   =   len(_A    ) // num_proc
          SCREAMING_SNAKE_CASE__   =   len(_A    ) % num_proc
          SCREAMING_SNAKE_CASE__   =   div * index + min(_A				,		_A    )
          SCREAMING_SNAKE_CASE__   =   start + div + (1 if index < mod else 0)
          split_kwds.append((function, iterable[start:end], types, index, disable_tqdm, desc)    )

     if len(_A    ) != sum(len(i[1]    ) for i in split_kwds    ):
          raise ValueError(
              F'''Error dividing inputs iterable among processes. '''
              F'''Total number of objects {len(_A    )}, '''
              F'''length: {sum(len(i[1]    ) for i in split_kwds    )}'''    )

     logger.info(
         F'''Spawning {num_proc} processes for {len(_A    )} objects in slices of {[len(i[1]    ) for i in split_kwds]}'''    )
     SCREAMING_SNAKE_CASE__,SCREAMING_SNAKE_CASE__   =   None, None
     if not disable_tqdm:
          SCREAMING_SNAKE_CASE__,SCREAMING_SNAKE_CASE__   =   (RLock(),), tqdm.set_lock
     with Pool(_A				,		initargs=_A				,		initializer=_A    ) as pool:
          SCREAMING_SNAKE_CASE__   =   pool.map(_A				,		_A    )
     logger.info(F'''Finished {num_proc} processes'''    )
     SCREAMING_SNAKE_CASE__   =   [obj for proc_res in mapped for obj in proc_res]
     logger.info(F'''Unpacked {len(_A    )} objects'''    )

     return mapped
def     UpperCAmelCase_       (						_A				,		_A				,		_A				,		_A				,		_A				,		_A				,		_A    ):





     '''simple docstring'''




     import joblib

     with joblib.parallel_backend(ParallelBackendConfig.backend_name				,		n_jobs=_A    ):
          return joblib.Parallel()(
              joblib.delayed(_A    )((function, obj, types, None, True, None)    ) for obj in iterable    )







@experimental
@contextlib.contextmanager
def     UpperCAmelCase_       (						_A    ):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__   =   backend_name

     if backend_name == "spark":
          from joblibspark import register_spark

          register_spark()

          # TODO: call create_cache_and_write_probe if "download" in steps
          # TODO: raise NotImplementedError when Dataset.map etc is called

     try:
          yield
     finally:
          SCREAMING_SNAKE_CASE__   =   None





def     UpperCAmelCase_       (						_A    ):





     '''simple docstring'''




     if divisor % 5 == 0 or divisor % 2 == 0:
          return 0
     SCREAMING_SNAKE_CASE__   =   1
     SCREAMING_SNAKE_CASE__   =   1
     while repunit:
          SCREAMING_SNAKE_CASE__   =   (10 * repunit + 1) % divisor
          repunit_index += 1
     return repunit_index
def     UpperCAmelCase_       (						_A = 1_00_00_00    ):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__   =   limit - 1
     if divisor % 2 == 0:
          divisor += 1
     while least_divisible_repunit(_A    ) <= limit:
          divisor += 2
     return divisor


if __name__ == "__main__":
       print(F"{solution() = }")


