a feza@s dZddlZddlZddlZddlZddlZddlmZddlm Z ddlm Z ddl m Z ddl mZddl Z ddlZddlmZddlZdd lmZdd lmZmZdd lmZdd lmZmZdd lmZddlmZmZmZm Z m!Z!m"Z"ddl#m$Z$ddl%m&Z&z ddl'Z'Wne(y,dZ'Yn0zddl)m*Z*Wne(yVdZ*Yn0z ddl+Z+Wne(y|dZ+Yn0ddl,m-Z-ddl,m.Z.ddl,m/Z/ddl,m0Z0ddl,m1Z1ddl,m2Z2ddl3m4Z4m5Z5ddl3m6Z6m7Z7ddl3m8Z8m9Z9ddl3m:Z:m;Z;mm?Z?ddl>m@Z@dgeAe;BZCeCdd e;D7ZCd!d"gZDeDd#gZEeFe:d$rxeCGe:Hd%e;e<ZId&d'ZJd(d)ZKd*d+ZLGd,d-d-eMZNd%d/d0ZOd1d2ZPd&d3d4ZQd5d6ZRd7d8ZSd9d:ZTd;d<ZUejVjd=d>d?e8d?d@fdAggdBdCedDdEZWedFeCedGgdHedIgdJdKdLZXedFeCdMdNZYdOdPZZeedQgdRdSdTZ[eedFgdUdVdWZ\dXdYZ]edZe;ed[e;d\d]Z^d^d_Z_d`daZ`dbdcZaedGgdddedfZbedFeEdgdhZcedidjZdedFeEdkdlZeeedFeEdmdnZfeedFeDdodpZgdqdrZhedFe;edsdgdtfdugdvfgdwdxZieedFeEdydzZjd{d|ZkeedFeDd}d~ZlddZmddZnddZoeedFeEejVjpddddZqGddde-ZrddZsedFeCddZtddZuddZvddZweEddZxexydd ezdDeedFexddZ{Gddde-Z|ddZ}ddZ~ddZddZGdddejj1Ze6deeedFgdddZeedGgdedFeEddZddZeedeEedeEddZeddZddZddZedgdddZedgdddZeddZeeedFeDddÄZdZeedFeDddƄZdZdZdZdZeedFeDe:durgnd%gedeeegedgd͢ddτZdjejejejdэZeddӄZdjejejejdэZeddքZdd؄ZddڄZeeedFeDdd܄ZddބZeeddZddZedFeCedGgdddZeeedFeDddZedFgdddZeedFeDddZddZddZddZddZddZd'ddZeedFd"d#gddZee e+duddddZd(ddZedFd"d#gddZddZedFdd"ge e*duddddZe e+dudddd Zd d Zd d ZddZddZddZeeedGgdddZeeedgdedGdd?gddZeedGdd?gedgdddZeeedFd!d#e/e.gd d!ZeedGgd"d#d$ZdS()z Test the parallel module. N)format_exceptionsqrt)sleep) PicklingError TimeoutError)reloadparallel)dumpload)get_reusable_executor)np with_numpy)with_multiprocessing) parametrizeraisescheck_subprocess_callskipifSkipTestwarns)TerminatedWorkerError)Queue) parallel_sum)SequentialBackend)ThreadingBackend)MultiprocessingBackend)ParallelBackendBase) LokyBackend) SafeFunction)Paralleldelayed)register_parallel_backendparallel_backend)effective_n_jobs cpu_count)mpBACKENDSDEFAULT_BACKENDEXTERNAL_BACKENDS)JoblibException)WorkerInterruptcCsg|]}t|qS)r)).0Z backend_strr.r.8lib/python3.9/site-packages/joblib/test/test_parallel.py Gr1multiprocessingloky threading get_contextspawncCst|dt|ddS)NZ_poolZ_workers)getattrbackendr.r.r0 get_workersRsr;cCs||SNr.)xyr.r.r0divisionVsr?cCs|dSNr.r=r.r.r0squareZsrCc@seZdZdZddZdS)MyExceptionWithFinickyInitz1An exception class with non trivial __init__ cCsdSr<r.)selfabcdr.r.r0__init__asz#MyExceptionWithFinickyInit.__init__N__name__ __module__ __qualname____doc__rJr.r.r.r0rD^srDFcCs"|dkr|rtddddnt|S)NrFrGrHrI)rD ValueError)r=custom_exceptionr.r.r0exception_raiseres rScCstdtdS)Ng?)timerKeyboardInterruptrBr.r.r0interrupt_raiserls rVcCs|d||S)zO A module-level function so that it can be spawn with multiprocessing. rAr.)r=r>zr.r.r0fqsrXcCsttdSNr)typer get_active_backendr.r.r.r0_active_backend_typexsr\cCst||dddtdDS)Nn_jobsr:css|]}tt|VqdSr<r#rCr/ir.r.r0 }sz parallel_func..r"range) inner_n_jobsr:r.r.r0 parallel_func|srgcCstdksJdSrY)r'r.r.r.r0test_cpu_countsrhcCstdksJdSrY)r&r.r.r.r0test_effective_n_jobssrizbackend_n_jobs, expected_n_jobs)rcrcr^Nr)z positive-intz negative-intNone)ZidscCsTtd|d"tdd|ks JWdn1s40YtdddksPJdS)Nr5rkr)r%r&)Zbackend_n_jobsZexpected_n_jobsr.r.r0test_effective_n_jobs_Nones 0rnr:r^)rrArjverbose)rA dcCs:ddtdDt|||dddtdDks6JdS)NcSsg|] }t|qSr.rCr/r=r.r.r0r1r2z(test_simple_parallel..)r^r:rpcss|]}tt|VqdSr<r_rtr.r.r0rbsz'test_simple_parallel..rer")r:r^rpr.r.r0test_simple_parallelsrwcCs|jtdddtd>}td|dddtdD}|gd ksLJWdn1s`0Yd d |D}t|d ksJdS) NnameZ!some_new_name_for_the_main_thread)targetrxvaluerAr]css|]}tt|VqdSr<r_rtr.r.r0rbsz6test_main_thread_renamed_no_warning..rc)rrcSsg|]}dt|jvr|qS)zworker timeout)strmessage)r/wr.r.r0r1r2z7test_main_thread_renamed_no_warning..r)setattrr5current_threadrr"relen)r: monkeypatchwarninforesultsr.r.r0#test_main_thread_renamed_no_warnings  .rcCsttd}t||dWdn1s*0Y|r\t|dkrXd|djjdvSdSt|dkslJdSdS)N)r:rfrzbacked parallel loops cannotFT)rrgrr}args)r:rfexpectedrecordsr.r.r0_assert_warning_nesteds * rz%parent_backend,child_backend,expected))r4r3T)r4r4F)r3r3T)r3r4T)r5r3T)r5r4Tcsltd|dfddtdDtd|dfddtdD}|dkr\t|shJn t|shJdS)NrAr]c3s |]}ttdddVqdS)rFr:rfrNr#rr/_ child_backendr.r0rbs z0test_nested_parallel_warnings..ruc3s |]}ttddVqdS)rArNrrrrr.r0rbs r5)r"reanyall)parent_backendrrresr.rr0test_nested_parallel_warningss r)r4r3r5cCs>dg}dd}tj||fd}|||ds:JdS)NFcSsdtd,}tddddtdDWdn1s:0Ytt|t|dk|d<dS)NrArkcss|]}ttdVqdS)皙?Nr#rrr.r.r0rbszPtest_background_thread_parallelism..background_thread..r{r)rr"reprintr)is_run_parallelrr.r.r0background_threads  & z=test_background_thread_parallelism..background_thread)ryrr)r5Threadstartjoin)r:rrtr.r.r0"test_background_thread_parallelisms rcCs"td|dddtdDdS)NrAr]css|]}ttdVqdS){Gz?Nr_rr.r.r0rbsznested_loop..rdr9r.r.r0 nested_loopsrrrcs&td|dfddtdDdS)NrAr]c3s|]}ttVqdSr<)r#rrrr.r0rbsz#test_nested_loop..rd)rrr.rr0test_nested_loopsrcCstdSr<)rQr9r.r.r0raise_exception src CstttXtddd,}|ttdttdgWdn1sH0YWdn1sf0YdS)NrAr4r])rrQr"r#rrr r.r.r0)test_nested_loop_with_exception_with_lokys    rcs<tddtdddfddtdDs8JdS) z1Input is mutable when using the threading backendru)maxsizerAr5r]c3s|]}tjdVqdSrN)r#Zputrqr.r0rbsz1test_mutate_input_with_threads..N)rr"reZfullr.r.rr0test_mutate_input_with_threadss  r)rrArccCs6td}dd|Dt|ddd|Dks2JdS)z.Check the keyword argument processing of pmap. cSsg|]}t|ddqSrr>rXrtr.r.r0r1#r2z(test_parallel_kwargs..rkcss|]}tt|ddVqdSrrNr#rXrtr.r.r0rb$r2z'test_parallel_kwargs..Nrv)r^lstr.r.r0test_parallel_kwargss rcCstd}dd|D}td|dh}|j}||dd|DksDJ||dd|Dks^Jtdur|t|t|jus|JWdn1s0Ytdurt|jdusJ||d d|DksJtdurt|jdusJdS) NrcSsg|]}t|ddqSrrrtr.r.r0r1*r2z4test_parallel_as_context_manager..r{r]css|]}tt|ddVqdSrrrtr.r.r0rb3r2z3test_parallel_as_context_manager..css|]}tt|ddVqdSrrrtr.r.r0rb4r2css|]}tt|ddVqdSrrrtr.r.r0rb@r2)rer"_backendr(r;)r:rrpZmanaged_backendr.r.r0 test_parallel_as_context_manager's4rcs`Gdddtttdd0tddfddtd DWd n1sR0Yd S) ze Check that pmap captures the errors when it is passed an object that cannot be pickled. c@seZdZddZdS)z1test_parallel_pickling..UnpicklableObjectcSs tddS)NZ123) RuntimeErrorrEr.r.r0 __reduce__Ksz.UnpicklableObject.__reduce__N)rLrMrNrr.r.r.r0UnpicklableObjectJsrzthe task to sendmatchrArkc3s|]}ttVqdSr<r#idrrr.r0rbOr2z)test_parallel_pickling..rN)objectrrr"rer.r.rr0test_parallel_picklingEsrcCs0ttd|ddddtdDdks,JdS)NrArr^r:timeoutcss|]}ttdVqdS)gMbP?Nrrtr.r.r0rbUsz0test_parallel_timeout_success..)rr"rer9r.r.r0test_parallel_timeout_successRs rcCsLtt0td|ddddtdDWdn1s>0YdS)NrArrcss|]}ttdVqdS)rNrrtr.r.r0rb^sz-test_parallel_timeout_fail..r)rrr"rer9r.r.r0test_parallel_timeout_failYs rc Cs`tdur(tt0td|dddtddDWdn1sH0Ytt*td|ddddDWdn1s0Ytd|d^}t|jdusJt|j}tt(|ddtddDWdn1s0Yt|jdusJt|j|us(Jd dtd D|d d td DksVJt|j}tt"|d ddDWdn1s0Yt|jdusJt|j|usJddtd D|dd td DksJWdn1s0Yt|jdusjJnBtt (tdddddDWdn1s`0Ytt.tddddtddDWdn1s0Ytt .tddddd tdDWdn1s0Yz"tdddd tddDWn6t yR}zt |t r>JWYd}~nd}~00tddS)NrAr]cSsg|]\}}tt||qSr.r#r?r/r=r>r.r.r0r1jsz&test_error_capture..)rr)rrcSsg|]}tt|qSr.r#rVrtr.r.r0r1nr2cSsg|]\}}tt||qSr.rrr.r.r0r1vscSsg|]}t|ddqSrrrtr.r.r0r1r2rcss|]}tt|ddVqdSrrrtr.r.r0rbr2z%test_error_capture..cSsg|]}tt|qSr.rrtr.r.r0r1r2cSsg|]}t|ddqSrrrtr.r.r0r1r2css|]}tt|ddVqdSrrrtr.r.r0rbr2rkcSsg|]}tt|qSr.rrtr.r.r0r1r2cSsg|]\}}tt||qSr.rrr.r.r0r1r2r)r^rpcss|]}tt|ddVqdS)T)rRNr#rSr`r.r.r0rbsrcss |]\}}tt||VqdSr<rrr.r.r0rbsz'The excepted error has not been raised.)r(rZeroDivisionErrorr"zipr-r;rrerUrD Exception isinstancer,rQ)r:r Zoriginal_workersexr.r.r0test_error_capturebsj   "   "  &   2*  $ $  $ &rcCs|d|dS)Nz Consumed %s)append)queueitemr.r.r0consumersrzbatch_size, expected_queue) Produced 0 Consumed 0 Produced 1 Consumed 1 Produced 2 Consumed 2 Produced 3 Consumed 3 Produced 4 Consumed 4 Produced 5 Consumed 5r{) rrrrrrrrrrrrcsTtfdd}td||dfdd|D|ks@JtdksPJdS) zC Test that with only one job, Parallel does act as a iterator. c3s&tdD]}d||VqdSNz Produced %irerrarr.r0producers z'test_dispatch_one_job..producerrr^ batch_sizer:c3s|]}tt|VqdSr<r#rrtrr.r0rbsz(test_dispatch_one_job.. N)listr"r)r:rZexpected_queuerr.rr0test_dispatch_one_jobs  rcst}|fdd}tddd|dfdd|Dt}|d d ksXJ|d d d }|dksvJ|d}||ksJtdksJd S)zW Check that using pre_dispatch Parallel does indeed dispatch items lazily. c3s&tdD]}d||VqdSrrrrr.r0rs z/test_dispatch_multiprocessing..producerrArrc)r^r pre_dispatchr:c3s|]}ttdVqdS)rNrrrr.r0rbsz0test_dispatch_multiprocessing..rrNr{z Consumed anyrjrr)r(ZManagerrr"indexr)r:ZmanagerrZqueue_contentsZfirst_consumption_indexZproduced_3_indexr.rr0test_dispatch_multiprocessings    rcCsZtdddd8}|ddtdD|jdks8JWdn1sL0YdS) NrAautor5rcss|]}tt|VqdSr<rr`r.r.r0rbr2z/test_batching_auto_threading..rr"rerZcompute_batch_size)rr.r.r0test_batching_auto_threadingsrcCsZtdd|d8}|ddtdD|jdks8JWdn1sL0YdS)NrArrcss|]}tt|VqdSr<rr`r.r.r0rbr2z2test_batching_auto_subprocesses..rrr)r:rr.r.r0test_batching_auto_subprocessessrcCsLtt0tddddddtdDWdn1s>0YdS) zCMake sure that exception raised during dispatch are indeed capturedrAr)r^rrpcss|]}tt|VqdSr<rr`r.r.r0rbsz*test_exception_dispatch..rN)rrQr"rer.r.r.r0test_exception_dispatchs rcCs tddddtdDdS)NrArkcss|]}tt|VqdSr<rr/jr.r.r0rb sz(nested_function_inner..rrdrr.r.r0nested_function_inner srcCs tddddtdDdS)NrArkcss|]}tt|VqdSr<)r#rrr.r.r0rbsz(nested_function_outer..rrdrr.r.r0nested_function_outersrz'https://github.com/joblib/loky/pull/255)reasoncCstt.}td|dddtdDWdn1s<0Yt|j|j|j}d|}d|vsnJd |vszJd |vsJt|jtusJdS) zEnsure errors for nested joblib cases gets propagated We rely on the Python 3 built-in __cause__ system that already report this kind of information to the user. rAr]css|]}tt|VqdSr<)r#rr`r.r.r0rbsz1test_nested_exception_dispatch..rNrrrS) rrQr"rerrZrztbr)r:excinfoZ report_linesreportr.r.r0test_nested_exception_dispatchs &    rc@s$eZdZdZdddZd ddZdS) FakeParallelBackendz8Pretends to run concurrently while running sequentially.rNcKs|||_||_|Sr<)r&r^r )rEr^r Z backend_argsr.r.r0 configure/s zFakeParallelBackend.configurecCs"|dkrttd|d}|S)Nrr)maxr(r'rEr^r.r.r0r&4sz$FakeParallelBackend.effective_n_jobs)rN)r)rLrMrNrOrr&r.r.r.r0r,s rcCs6tttddWdn1s(0YdS)Nz unit-testingr9rrQr"r.r.r.r0test_invalid_backend:s rcCsNtt }td|dWdn1s.0Ydt|jvsJJdS)Nrr]z&n_jobs == 0 in Parallel has no meaning)rrQr"Z_initialize_backendr|rz)r:rr.r.r0test_invalid_njobs?s .rcCs>z0tdtdtvsJtdtks(JWtd=ntd=0dS)NZ test_backend)r$rr)r.r.r.r0test_register_parallel_backendFs   rcCsTttksJz*tdtdddttks0JWtt_ntt_0ttksPJdS)Nr5T)Z make_default)r\DefaultBackendr$r)rr*r r.r.r.r0test_overwrite_default_backendOs rcCst|ddt\}}|dks&Jtddks6Jt}|jdksJJ|dkrvt|tksbJt|jtksJn|dkrt|t ksJt|jt ksJnX|dkrt|t ksJt|jt ksJn,| drt|t ksJt|jt ksJWdn1s0YdS)Nrcrkr3r4r5Ztest_) r%r r[r&r"r^rZrrr r startswithr)Z backend_nameactive_backend active_n_jobsrr.r.r0check_backend_context_manager[s$   r cCsg|] }d|qS)ztest_backend_%dr.r`r.r.r0r1rr2rccCsf|tvr|t|tttks$Jt|ttks:JtdddddtDttksbJdS)NrAr5r]css|]}|stt|VqdSr<)r#r )r/rGr.r.r0rbsz/test_backend_context_manager..)r)setitemrr\rr r" all_backends_for_context_manager)rr:r.r.r0test_backend_context_managervsrc@seZdZdZdddZdS)ParameterizedParallelBackendz9Pretends to run conncurrently while running sequentially.NcCs|durtd||_dS)Nzparam should not be None)rQparam)rErr.r.r0rJsz%ParameterizedParallelBackend.__init__)NrKr.r.r.r0rsrcCs|tdtttksJtdddd~t\}}t|tksHJ|j dksVJ|dksbJt }|j dksvJ|j |usJ|ddt dD}Wdn1s0Y|dd t dDksJttksJdS) NZ param_backend*rc)rr^css|]}tt|VqdSr<r#rr`r.r.r0rbr2z=test_parameterized_backend_context_manager..rucSsg|] }t|qSr.rr`r.r.r0r1r2z>test_parameterized_backend_context_manager..)r r)rr\rr%r r[rZrr"r^rre)rr r rrr.r.r0*test_parameterized_backend_context_managers  4rcCsttksJttdddd~t\}}t|tks>J|jdksLJ|dksXJt}|j dkslJ|j |uszJ|ddt dD}Wdn1s0Y|ddt dDksJttksJdS) N+)rrurkcss|]}tt|VqdSr<rr`r.r.r0rbr2zFtest_directly_parameterized_backend_context_manager..cSsg|] }t|qSr.rr`r.r.r0r1r2zGtest_directly_parameterized_backend_context_manager..) r\rr%rr r[rZrr"r^rre)r r rrr.r.r03test_directly_parameterized_backend_context_managers  4rcCstdtS)Nr)rosgetpidr.r.r.r0sleep_and_return_pidsrcCs<ttksJtdks JtddddtdDS)NrrArkcss|]}ttVqdSr<)r#rrr.r.r0rbsz"get_nested_pids..)r\rr"_effective_n_jobsrer.r.r.r0get_nested_pidss rcs eZdZdZfddZZS) MyBackendz:Backend to test backward compatibility with older backendscstt|dSrY)superrget_nested_backendr __class__r.r0rszMyBackend.get_nested_backend)rLrMrNrOr __classcell__r.r.rr0rsrback_compat_backend)r5r4r3r"cCsft|JtddddtdD}|D]}tt|dks*Jq*Wdn1sX0YdS)NrArkcss|]}ttVqdSr<)r#rrr.r.r0rbsz6test_nested_backend_context_manager..rr)r%r"rerset)r:Z pid_groupsZ pid_groupr.r.r0#test_nested_backend_context_managers  r$)rArjNcs|ddtddfddtdDtd4tddfddtdDWdn1sn0YdS) NcSs0tt|ksJt|}t|ks,JdSr<)r\r)r&r"r)Zexpected_backend_typeZexpected_n_jobr.r.r0check_nested_backendsz?test_nested_backend_in_sequential..check_nested_backendrrkc3s|]}tddVqdS)r4rNr#r)r%r.r0rbsz4test_nested_backend_in_sequential..rc3s|]}tVqdSr<r&rr:r%r^r.r0rbs)r"rer%)r:r^r.r'r0!test_nested_backend_in_sequentials r(cCs>t|"\}}|j|ksJWdn1s00YdSr<)r% nesting_level) inner_backendZexpected_levelr:r^r.r.r0check_nesting_levelsr+ outer_backendr*csxt|dtd|dfddtdDtdd,tfddtdDWdn1sj0YdS) NrrAr]c3s|]}ttdVqdSrr#r+rr*r.r0rbsz-test_backend_nesting_level..rrkc3s|]}ttdVqdSrr-rr.r.r0rbs)r+r"rer%)r,r*r.r.r0test_backend_nesting_level s r/csddlGfdddt}td|ddtdB\}}tddfd d td D|jd ksjJWdn1s~0YdS) NrcseZdZdZjddZdS)z)test_retrieval_context..MyBackendrcss|jd7_dVdSrlrrr.r.r0retrieval_context!sz;test_retrieval_context..MyBackend.retrieval_contextN)rLrMrNracontextmanagerr0r.) contextlibr.r0rsrZ retrievalcSstddddt|DS)NrArkcss|]}tt|VqdSr<rr`r.r.r0rb)r2z>test_retrieval_context..nested_call..rd)nr.r.r0 nested_call(sz+test_retrieval_context..nested_callrArkc3s|]}t|VqdSr<r&r`)r4r.r0rb,sz)test_retrieval_context..rur)r2rr$r%r"rera)rZbarr.)r2r4r0test_retrieval_contexts r5cCstd}t|t|dS)NZfoobar)r,reprpickledumps)er.r.r0test_joblib_exception5sr:cCsvtt}tt|ddWdn1s00Ytt}tt|dWdn1sh0YdS)Nrrr=)r!r?rrrVr-)Z safe_divisionZsafe_interruptr.r.r0test_safe_function>s  ( r;r)rrjgQ?cCs6ttt|dWdn1s(0YdS)Nrrr<r.r.r0test_invalid_batch_sizeHs r=z)n_tasks, n_jobs, pre_dispatch, batch_size) )rArArr)rArAr^rrrAr^r)irAr^rr>)rr{r^r)rr^r)r 2 * n_jobsr)rrr)rBrrArP)r?rrArcCsN|||d}ddt|D}tfi|ddt|D}||ksJJdS)N)r^rrcSsg|] }t|qSr.rsr`r.r.r0r1`r2z0test_dispatch_race_condition..css|]}tt|VqdSr<r_r`r.r.r0rbar2z/test_dispatch_race_condition..rv)Zn_tasksr^rrparamsrrr.r.r0test_dispatch_race_conditionNs  rDcCs8t}tddd}|jd}|}||ks4JdS)NrAr3r]context)r(get_start_methodr"Z _backend_argsget)Zmp_start_methodrrEZ start_methodr.r.r0test_default_mp_contextes   rHcs^|dkrtd}tjd}|ddtjtd|dfddt dDdS) Nr3r7rirAr]c3s |]}ttjjVqdSr<)r#rdotTr`rFr.r0rbszAtest_no_blas_crash_or_freeze_with_subprocesses..) r(r6rrandomZ RandomStaterandnrIrJr"re)r:rngr.rKr0.test_no_blas_crash_or_freeze_with_subprocessesns   rOafrom joblib import Parallel, delayed def square(x): return x ** 2 backend = "{}" if backend == "spawn": from multiprocessing import get_context backend = get_context(backend) print(Parallel(n_jobs=2, backend=backend)( delayed(square)(i) for i in range(5))) cCsB|dkrtdkrtdt|}ttjd|gddddS)Nr3forkzVRequire fork start method to use interactively defined functions with multiprocessing.z-cr\[0, 1, 4, 9, 16\]r stdout_regex) r(rFpytestskip,UNPICKLABLE_CALLABLE_SCRIPT_TEMPLATE_NO_MAINformatrsys executable)r:coder.r.r02test_parallel_with_interactively_defined_functionss   r[aimport sys # Make sure that joblib is importable in the subprocess launching this # script. This is needed in case we run the tests from the joblib root # folder without having installed joblib sys.path.insert(0, {joblib_root_folder!r}) from joblib import Parallel, delayed def run(f, x): return f(x) {define_func} if __name__ == "__main__": backend = "{backend}" if backend == "spawn": from multiprocessing import get_context backend = get_context(backend) callable_position = "{callable_position}" if callable_position == "delayed": print(Parallel(n_jobs=2, backend=backend)( delayed(square)(i) for i in range(5))) elif callable_position == "args": print(Parallel(n_jobs=2, backend=backend)( delayed(run)(square, i) for i in range(5))) else: print(Parallel(n_jobs=2, backend=backend)( delayed(run)(f=square, x=i) for i in range(5))) z!def square(x): return x ** 2 zcdef gen_square(): def square(x): return x ** 2 return square square = gen_square() zsquare = lambda x: x ** 2 define_funccallable_position)r#rkwargsc Csv|dvr$|tkstjdkr$tdtj|||tj tj t j d}| d}| |ttj|jgddddS) N)r3r7win32zNot picklable with pickle)r\r:r]joblib_root_folderzunpicklable_func_script.pyrrQrR) SQUARE_MAINrXplatformrTrU)UNPICKLABLE_CALLABLE_SCRIPT_TEMPLATE_MAINrWrpathdirnamejoblib__file__rwriterrYstrpath)r:r\r]tmpdirrZZ code_filer.r.r00test_parallel_with_unpicklable_functions_in_argss"    rkaimport sys # Make sure that joblib is importable in the subprocess launching this # script. This is needed in case we run the tests from the joblib root # folder without having installed joblib sys.path.insert(0, {joblib_root_folder!r}) from joblib import Parallel, delayed from functools import partial class MyClass: '''Class defined in the __main__ namespace''' def __init__(self, value): self.value = value def square(x, ignored=None, ignored2=None): '''Function defined in the __main__ namespace''' return x.value ** 2 square2 = partial(square, ignored2='something') # Here, we do not need the `if __name__ == "__main__":` safeguard when # using the default `loky` backend (even on Windows). # The following baroque function call is meant to check that joblib # introspection rightfully uses cloudpickle instead of the (faster) pickle # module of the standard library when necessary. In particular cloudpickle is # necessary for functions and instances of classes interactively defined in the # __main__ module. print(Parallel(n_jobs=2)( delayed(square2)(MyClass(i), ignored=[dict(a=MyClass(1))]) for i in range(5) )) )r`cCs.|d}|tttj|jgddddS)Nz(joblib_interactively_defined_function.pyrQru)rSr)rrh5INTERACTIVE_DEFINED_FUNCTION_AND_CLASS_SCRIPT_CONTENTrrXrYrirjZscriptr.r.r0Btest_parallel_with_interactively_defined_functions_default_backends    rnauimport sys # Make sure that joblib is importable in the subprocess launching this # script. This is needed in case we run the tests from the joblib root # folder without having installed joblib sys.path.insert(0, {joblib_root_folder!r}) from joblib import Parallel, delayed, hash import multiprocessing as mp mp.util.log_to_stderr(5) class MyList(list): '''MyList is interactively defined by MyList.append is a built-in''' def __hash__(self): # XXX: workaround limitation in cloudpickle return hash(self).__hash__() l = MyList() print(Parallel(n_jobs=2)( delayed(l.append)(i) for i in range(3) )) cCs0|d}|tttj|jgdddddS)Nz)joblib_interactive_bound_method_script.pyz\[None, None, None\]Z LokyProcess)rSZ stderr_regexr)rrh9INTERACTIVELY_DEFINED_SUBCLASS_WITH_METHOD_SCRIPT_CONTENTrrXrYrirmr.r.r05test_parallel_with_interactively_defined_bound_method9s   rqcCs"tg}tdd|gksJdS)NrArk)iterr")Zexhausted_iteratorr.r.r0%test_parallel_with_exhausted_iteratorCsrscCs"t|tjstdt||S)Nz#Expected np.memmap instance, got %r)rrmemmap TypeErrorrZcopyrKr.r.r0 check_memmapHs  rwcCsdd}tdd|ddd|dD}t||t|D]\}}tj||q:td d|dd d|dD}t||t|D]\}}tj||qdS) Ncss(t|D]}tjdtjd|VqdS)NrZdtype)reronesZfloat32)r3rar.r.r0generate_arraysWs zBtest_auto_memmap_on_arrays_from_generator..generate_arraysrArr^ max_nbytesr:css|]}tt|VqdSr<r#rwr/rFr.r.r0rb\sz.rrr{css|]}tt|VqdSr<r}r~r.r.r0rbds)r"rrrtestingassert_array_equal)r:rzrresultrr.r.r0)test_auto_memmap_on_arrays_from_generatorOsrcCs|Sr<r.)argr.r.r0identityjsrcs|dj}tj}tj|ddtj|ddg}t||t|ddt ddfdd d D\}t d tj svJd j |ksJtj ||dS) Nz test.mmapuint8rxr)Z mmap_moderArkc3s|]}ttVqdSr<)r#rrrtr.r0rbvr2z.test_memmap_with_big_offset..)rr)rrimmapZALLOCATIONGRANULARITYrZzerosryr r r"rrtoffsetrr)rjfnamesizeobjrr.rr0test_memmap_with_big_offsetns   rcCstd,}tddddtdDWdn1s:0Yt|dksTJ|d}t|jtslJt|jdks~JdS)Nr)rcss|]}tt|VqdSr<r_r`r.r.r0rb~r2zFtest_warning_about_timeout_not_supported_by_backend..2rzThe backend class 'SequentialBackend' does not support timeout. You have set 'timeout=1' in Parallel but the 'timeout' parameter will not be used.)rr"rerrr} UserWarningr|)rr~r.r.r03test_warning_about_timeout_not_supported_by_backend|s :r)rrArorjcCsvdgdgd}tt2t}t||ddd|DWdn1sP0Yt|}|dksrJdS)NrFrrrr]css|]}ttj|VqdSr<r#rTrr`r.r.r0rbsz%test_abort_backend..)rrurTr")r^r:ZdelaysZt_startZdtr.r.r0test_abort_backends & rcCs|j}tdd||dF}|ddtjdgdDtt|dksNJWdn1sb0YtdD]}t|sqtd qtt d tdd|d }|d dtjdgdDtdD]}t|sqtd qt d dS) NrAr)r^r|r:Z temp_foldercss|]}tt|VqdSr<r}r~r.r.r0rbr2z(test_memmapping_leaks..rrrrrz/temporary directory of Parallel was not removedr{css|]}tt|VqdSr<r}r~r.r.r0rbr2) rir"rrLrrlistdirrerAssertionError)r:rjrrr.r.r0test_memmapping_leakss& 4       r)Nr4r5cCs<td|dddtdD}|ddtdDks8JdS)NrAr]css|]}tdd|VqdS)cSs|dSr@r.rBr.r.r0r2z2test_lambda_expression...Nr&r`r.r.r0rbsz)test_lambda_expression..rcSsg|] }|dqS)rAr.r`r.r.r0r1r2z*test_lambda_expression..rd)r:rr.r.r0test_lambda_expressionsrcsd}d}d|td||d}|fddt|D|jj|jjksLJ|jj|jjks`J|fddt|D|jj|jjksJ|jj|jjksJd S) zCTest that a parallel backend correctly resets its batch statistics.rAig@r)rpr^r:c3s|]}ttjVqdSr<rr`Z task_timer.r0rbr2z6test_backend_batch_statistics_reset..c3s|]}ttjVqdSr<rr`rr.r0rbr2N)r"rerZ_effective_batch_sizeZ_DEFAULT_EFFECTIVE_BATCH_SIZEZ_smoothed_batch_durationZ _DEFAULT_SMOOTHED_BATCH_DURATION)r:r^Zn_inputsrr.rr0#test_backend_batch_statistics_resets$rcCsdD]v}tt|djtks Jt|dd}t|jtks>Jt|dd}t|jtks\Jt|dd}t|jtksJqtdd dd }t|jtksJtd dd:tdd }t|jtksJ|jdksJWdn1s0Ytd dd@td dd}t|jtks J|jd ks0JWdn1sF0Ytd dd>tdd }t|jtks|J|jdksJWdn1s0Ytd dd@td dd}t|jtksJ|jd ksJWdn1s0YdS)N)rrArjrkthreadsr^prefer processes sharedmemr^requirerAr4)r^r:rrrcrr)rZr"rr rr%r^)r^rr.r.r0$test_backend_hinting_and_constraintss4    , 0 0 rcCsGdddt}t|Ltddd}t|j|ks:Jtddd}t|j|ksXJWdn1sl0YGdd d t}t|tddd}t|j|ksJ|\}}|d ksJ|d ksJtddd d }t|jtksJ|\}}d }||ksJ|d ks$JWdn1s:0Ytt t|ddWdn1sr0YdS)Nc@s$eZdZdZdZddZddZdS)z[test_backend_hinting_and_constraints_with_custom_backends..MyCustomThreadingBackendTcSsdSr<r.rr.r.r0 apply_async szgtest_backend_hinting_and_constraints_with_custom_backends..MyCustomThreadingBackend.apply_asynccSs|Sr<r.rr.r.r0r&szltest_backend_hinting_and_constraints_with_custom_backends..MyCustomThreadingBackend.effective_n_jobsNrLrMrNZsupports_sharedmemZ use_threadsrr&r.r.r.r0MyCustomThreadingBackend srrArrrrc@s$eZdZdZdZddZddZdS)z\test_backend_hinting_and_constraints_with_custom_backends..MyCustomProcessingBackendFcSsdSr<r.rr.r.r0rszhtest_backend_hinting_and_constraints_with_custom_backends..MyCustomProcessingBackend.apply_asynccSs|Sr<r.rr.r.r0r&!szmtest_backend_hinting_and_constraints_with_custom_backends..MyCustomProcessingBackend.effective_n_jobsNrr.r.r.r0MyCustomProcessingBackendsrrr)r^rrpzUsing ThreadingBackend as joblib.Parallel backend instead of MyCustomProcessingBackend as the latter does not provide shared memory semantics.r:r) rr%r"rZrZ readouterrrstriprrQ)Zcapsysrrrouterrrr.r.r09test_backend_hinting_and_constraints_with_custom_backendss*   0      . rcCstttddWdn1s(0YtttddWdn1sZ0YtttdddWdn1s0YtttdddWdn1s0Ytttd ddWdn1s0YdS) NZinvalidrrrr)rrr4rr3rr.r.r.r0,test_invalid_backend_hinting_and_constraints:s ( ( * * rcCsFtj}td}ttjts"J|ttjt|usBJdS)Nr5)r"rr%rrZ unregisterrZ)defaultZpbr.r.r0test_global_parallel_backendNs rcCsNdd}|td<td"ttjts,JWdn1s@0YdS)NcSs ttd<dS)Nfoo)rr)r.r.r.r0 register_fooYsz,test_external_backends..register_foor)r+r%rr"rr)rr.r.r0test_external_backendsXs rc stddh}t|jj|jjfg}dkr<|WdS|fddtdD}||dWdS1sx0YdS)zCPerform nested parallel calls and introspect the backend on the wayrArkrNc3s(|] }ttfddiVqdS)limitrN)r#_recursive_backend_infor`r^rr.r0rbisz*_recursive_backend_info..r)r"rZrrLr)re)rr^rZ this_levelrr.rr0rbs rcCst|ddt}Wdn1s(0YtdkrFd}d}nd}d}|d}|df|dfd|fd|fg}||ksJdS)NrArkrrrZBackendr)r%rr'title)r:backend_types_and_levelsZsecond_level_backend_typeZ max_levelZtop_level_backend_typeZexpected_types_and_levelsr.r.r0test_nested_parallelism_limitns$  rzThis test requires daskc Cstjddd}tjtdtjd}tdD]\}tdt|d}Wdn1sV0Yt |dkspJt dd |Ds*Jq*tdt}Wdn1s0Yt |dksJt d d |DsJdS) NrA)Z n_workersZthreads_per_workergcArxdask)datar{css|]\}}|dkVqdSZDaskDistributedBackendNr.r/rxrr.r.r0rbsz4test_nested_parallelism_with_dask..css|]\}}|dkVqdSrr.rr.r.r0rbs) distributedZClientrryintrrer%rrr)Zclientrrarr.r.r0!test_nested_parallelism_with_dasks  ( $rcCstddtdDS)z6A horrible function that does recursive parallel callscss|]}ttVqdSr<)r#_recursive_parallelr`r.r.r0rbr2z&_recursive_parallel..rArd)Z nesting_limitr.r.r0rsrc Cst|dd>tt}tWdn1s20YWdn1sP0Y|j}|dkr~t|tr~tdnt|t sJdS)NrArkr4z1Loky worker crash when serializing RecursionError) r%r BaseExceptionrrzrrrTxfailRecursionError)r:rexcr.r.r0test_thread_bomb_mitigations B rcCs*i}dD]}tj|||<q|tdfS)N)OMP_NUM_THREADSOPENBLAS_NUM_THREADSMKL_NUM_THREADSZVECLIB_MAXIMUM_THREADSZNUMEXPR_NUM_THREADSZNUMBA_NUM_THREADS ENABLE_IPCrr)renvironrGr)Zenv_varsvarr.r.r0_run_parallel_sumsrzNeed OpenMP helper compiledcCstd|dddtdD}ttdd}|D]Z\}}||ksFJ|D]<\}}|drr|t|ksJqN|dks~J|dksNJqNq2dS) NrAr]css|]}ttVqdSr<)r#rrr.r.r0rbsz-test_parallel_thread_limit..rZ_THREADSr1)r"rerr'itemsendswithr|)r:rZexpected_num_threadsZworker_env_varsZomp_num_threadsrxrzr.r.r0test_parallel_thread_limits    rz%This test requires dask NOT installedcCs8ttddtdWdn1s*0YdS)NzPlease install daskrr)rrQr%r.r.r.r0)test_dask_backend_when_dask_not_installedsrc CsGdddt}d}t|Ztjt|d,tddddtdDWdn1s`0YWdn1s~0YdS) Nc@s&eZdZddZdddZddZdS) z3test_zero_worker_backend..ZeroWorkerBackendc_sdSrYr.)rErr^r.r.r0rsz=test_zero_worker_backend..ZeroWorkerBackend.configureNcSs tddS)NzNo worker availabler)rEfunccallbackr.r.r0rsz?test_zero_worker_backend..ZeroWorkerBackend.apply_asynccSsdSrYr.rr.r.r0r&szDtest_zero_worker_backend..ZeroWorkerBackend.effective_n_jobs)N)rLrMrNrrr&r.r.r.r0ZeroWorkerBackends rz&ZeroWorkerBackend has no active workerrrArkcss|]}tt|VqdSr<rr`r.r.r0rbr2z+test_zero_worker_backend..)rr%rTrrr"re)rZ expected_msgr.r.r0test_zero_worker_backends   rcsdadddksJtddfddtdD}t|dhksLJdadks^Jtddfd dtdD}t|dhksJdS) Nzoriginal valuecSstSr<)MY_GLOBAL_VARIABLEr.r.r.r0 check_globalssz@test_globals_update_at_each_parallel_call..check_globalsrArkc3s|]}tVqdSr<r&r`rr.r0rbsz.z changed valuec3s|]}tVqdSr<r&r`rr.r0rbs)rr"rer#)Zworkers_global_variabler.rr0)test_globals_update_at_each_parallel_callsrcCs4ddl}|jdd}|||ddlm}|S)Nrrr)threadpool_info)ZnumpyrLrMrIZ threadpoolctlr)rrFrr.r.r0_check_numpy_threadpool_limits s   rcCs8|D] }|d|dkr|dSqtd|dS)Nfilepath num_threadsz,An unexpected module was loaded in child: {})rQrW) child_module parent_info parent_moduler.r.r0_parent_max_num_threads_fors rcCs@|D]6}|D],}t||}t|||h}|d|vs Jq qdS)Nr)rmin)Z workers_inforrZchild_threadpool_inforZparent_max_num_threadsrr.r.r0check_child_num_threadssr)rAr{rorjcCsbt}t|dkrtjddt|dddtdD}t|}tt|d}t |||dS) Nr&Need a version of numpy linked to BLASmsgrkcss|]}ttVqdSr<r#rr`r.r.r0rb4sz6test_threadpool_limitation_in_child..rAr) rrrTrUr"rer&rr'r)r^rworkers_threadpool_infosexpected_child_num_threadsr.r.r0#test_threadpool_limitation_in_child(s  rinner_max_num_threads)rrAr{NrAcCst}t|dkrtjddtd|d,t|dddtd D}Wdn1s\0Yt|}|durtt |d }n|}t |||dS) Nrrrr4rrkcss|]}ttVqdSr<rr`r.r.r0rbLsz>test_threadpool_limitation_in_child_context..rAr) rrrTrUr%r"rer&rr'r)r^rrrrr.r.r0+test_threadpool_limitation_in_child_context>s   &rvar_name)rrrc stddddtj}zdtj<t|dfddtd D}|ddgks`Jtd d d 2t|dfd dtd D}Wdn1s0Y|ddgksJW|durtj=q|tj<n|durtj=n |tj<0dS)NTZreusecSs tj|Sr<)rrrG)rr.r.r0_get_envfsz>test_threadpool_limitation_in_child_override.._get_env4rkc3s|]}tVqdSr<r&r`rrr.r0rbmsz?test_threadpool_limitation_in_child_override..rAr4rrc3s|]}tVqdSr<r&r`rr.r0rbrsr)rZshutdownrrrGr"rer%)r^rZoriginal_var_valuerr.rr0,test_threadpool_limitation_in_child_overrideYs(   &   rcCs<ttddt|ddWdn1s.0YdS)Nz#does not acc.*inner_max_num_threadsrrr)rrr%r9r.r.r01test_threadpool_limitation_in_child_context_error}sr)rAr{rjcCsJdd}||tdd}tdD]"}||tdd}||ks"Jq"dS)Ncs,tdt|dfddtdDdS)Nrrkc3s|]}ttVqdSr<)r#sumr`rBr.r0rbr2zAtest_loky_reuse_workers..parallel_call..rvrkr.rBr0 parallel_callsz.test_loky_reuse_workers..parallel_callTrr)rre)r^rZfirst_executorrexecutorr.r.r0test_loky_reuse_workerss   r)F)rr)rc)N)rOrrXrTrr5 tracebackrZmathrrr7rr3rrT importlibr rfr r r Zjoblib.externals.lokyrZjoblib.test.commonrrrZjoblib.testingrrrrrrZ&joblib.externals.loky.process_executorrrrposix ImportErrorZ _openmp_test_helper.parallel_sumrrZjoblib._parallel_backendsrrrrr r!Zjoblib.parallelr"r#r$r%r&r'r(r)r*r+Zjoblib.my_exceptionsr,r-sortedkeysZALL_VALID_BACKENDSZPROCESS_BACKENDSZPARALLEL_BACKENDShasattrrr6rr;r?rCrrDrSrVrXr\rgrhriZmarkrnrwrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr rextendrerrrrrrZ_parallel_backendsrr$r(r+r/r5r:r;r=rDrHrOrVr[rcraZ SQUARE_LOCALZ SQUARE_LAMBDArkrWrdrergrlrnrprqrsrwrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr.r.r.r0s(                                             I                           $ (       #   .4          !