

import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


a						=logging.get_logger(__name__)






class 			A_						:
							_UpperCAmelCase       :    List[str]		 = None






@experimental
def   SCREAMING_SNAKE_CASE__			(    lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__     )    ->	Optional[Any]:
	if ParallelBackendConfig.backend_name is None:
		return _map_with_multiprocessing_pool(
		    _lowerCAmelCase      ,					_lowerCAmelCase      ,					_lowerCAmelCase      ,					_lowerCAmelCase      ,					_lowerCAmelCase      ,					_lowerCAmelCase      ,					_lowerCAmelCase     )

	return _map_with_joblib(_lowerCAmelCase      ,					_lowerCAmelCase      ,					_lowerCAmelCase      ,					_lowerCAmelCase      ,					_lowerCAmelCase      ,					_lowerCAmelCase      ,					_lowerCAmelCase     )






def   SCREAMING_SNAKE_CASE__			(    lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__     )    ->	Optional[int]:
	__lowerCamelCase							:							int   			=					num_proc if num_proc <= len(_lowerCAmelCase     ) else len(_lowerCAmelCase     )
	__lowerCamelCase							:							Tuple   			=					[]  # We organize the splits ourselve (contiguous splits)
	for index in range(_lowerCAmelCase     ):
		__lowerCamelCase							:							Union[str, Any]   			=					len(_lowerCAmelCase     ) // num_proc
		__lowerCamelCase							:							Optional[Any]   			=					len(_lowerCAmelCase     ) % num_proc
		__lowerCamelCase							:							int   			=					div * index + min(_lowerCAmelCase      ,					_lowerCAmelCase     )
		__lowerCamelCase							:							Union[str, Any]   			=					start + div + (1 if index < mod else 0)
		split_kwds.append((function, iterable[start:end], types, index, disable_tqdm, desc)     )

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

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

	return mapped






def   SCREAMING_SNAKE_CASE__			(    lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__      ,					lowerCamelCase__     )    ->	Optional[Any]:
	# progress bar is not yet supported for _map_with_joblib, because tqdm couldn't accurately be applied to joblib,
	# and it requires monkey-patching joblib internal classes which is subject to change
	import joblib

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






@experimental
@contextlib.contextmanager
def   SCREAMING_SNAKE_CASE__			(    lowerCamelCase__     )    ->	int:
	__lowerCamelCase							:							Optional[int]   			=					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:
		__lowerCamelCase							:							List[Any]   			=					None











__lowerCamelCase			:	str					    =	6_5521



def 						A_							(	_lowerCAmelCase	)		->					int:
 UpperCamelCase      :   Any			      =  1
 UpperCamelCase      :   str			      =  0
 for plain_chr in plain_text:
  UpperCamelCase      :   List[Any]			      =  (a + ord(_lowerCAmelCase	)) % MOD_ADLER
  UpperCamelCase      :   List[Any]			      =  (b + a) % MOD_ADLER
 return (b << 16) | a
