o +if@sddlZddlmZmZddlmZddlmZmZe e Z GdddZ edd Z d d Zd d ZeejdefddZdS)N)PoolRLock)tqdm) experimentalloggingc@seZdZdZdS)ParallelBackendConfigN)__name__ __module__ __qualname__ backend_namer r S/var/www/html/corbot_env/lib/python3.10/site-packages/datasets/parallel/parallel.pyr src Cs:tjdurt||||||||| St||||||||| S)a **Experimental.** Apply a function to iterable elements in parallel, where the implementation uses either multiprocessing.Pool or joblib for parallelization. Args: function (`Callable[[Any], Any]`): Function to be applied to `iterable`. iterable (`list`, `tuple` or `np.ndarray`): Iterable elements to apply function to. num_proc (`int`): Number of processes (if no backend specified) or jobs (using joblib). types (`tuple`): Additional types (besides `dict` values) to apply `function` recursively to their elements. disable_tqdm (`bool`): Whether to disable the tqdm progressbar. desc (`str`): Prefix for the tqdm progressbar. single_map_nested_func (`Callable`): Map function that applies `function` to an element from `iterable`. Takes a tuple of function, data_struct, types, rank, disable_tqdm, desc as input, where data_struct is an element of `iterable`, and `rank` is used for progress bar. N)rr _map_with_multiprocessing_pool_map_with_joblib functioniterablenum_procbatched batch_sizetypes disable_tqdmdescsingle_map_nested_funcr r r parallel_maps rc  Cs|t|kr|nt|}g} t|D]4} t||} t||} | | t| | } | | | | kr2dnd}| ||| ||||| ||fqt|tdd| Dkrgtdt|dtdd| Dtd|d t|d d d | Dd \}}|stft j }}t |||d}| || }Wdn1swYtd|ddd |D}tdt|d|S)Nrcs|] }t|dVqdSrNlen.0ir r r 7z1_map_with_multiprocessing_pool..zHError dividing inputs iterable among processes. Total number of objects z , length: csrrrr!r r rr$;r%z Spawning z processes for z objects in slices of cSsg|]}t|dqS)rrr!r r r ?sz2_map_with_multiprocessing_pool..)NN)initargs initializerz Finished z processescSsg|] }|D]}|qqSr r )r"proc_resobjr r rr&Gsz Unpacked z objects) r rangeminappendsum ValueErrorloggerinforrset_lockrmap)rrrrrrrrr split_kwdsindexdivmodstartendr'r(poolmappedr r rr+s:   $"rc s`ddljtj|dfdd|DWdS1s)wYdS)Nr)n_jobsc 3s.|]}|dddfVqdS)NT)delayed)r"r*rrrjoblibrrr rr$Us  z#_map_with_joblib..)r?parallel_backendrr Parallelrr r>rrMs $rr ccs>|t_|dkrddlm}|z dVWdt_dSdt_w)a **Experimental.** Configures the parallel backend for parallelized dataset loading, which uses the parallelization implemented by joblib. Args: backend_name (str): Name of backend for parallelization implementation, has to be supported by joblib. Example usage: ```py with parallel_backend('spark'): dataset = load_dataset(..., num_proc=2) ``` sparkr)register_sparkN)rr joblibsparkrC)r rCr r rr@[s r@) contextlibmultiprocessingrr tqdm.autorutilsrr get_loggerr r0rrrrcontextmanagerstrr@r r r rs   "