iDdZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddlmZmZmZmZddlmZmZmZmZmZmZddlZddlmZddlm Z edz Z! ddl"m#Z#e#dd l$m%Z%d Z& dd l$m(Z(d Z) ddl*Z+d Z,e jZe)d Z.e jZe&dZ/e)xre(j`dk(Z1e jZe1dZ2ejfjidZ5ejfjidZ6ejndkZ8e8Z9e5 Z:dZ;e,rdZGdde?Z@Gdde@ZAGdde?ZBGdd e@ZCGd!d"e@ZDGd#d$e@ZEd%ZFd&ZGe jZIGd'd(e?ZJd)ZKeGeIejZMeGeKd*ZNe5seGeKd+ZOeGeKd,ZPeGeKd-ZQGd.d/eZRe jd0vZTe jZeTd1Gd2d3eRZUeUjGd4d5e?ZWGd6d7eWeRZXeXjGd8d9eZYeGd:d;eYZZeZjeGd<d=eYZ[eGd>d?eYZ\ee2Gd@dAeYZ]ee/GdBdCeYZ^eGdDdEeZ_ee.GdFdGeZ`eadHk(re jyy#e'$rd Z&YwxYw#e'$rd Z)YwxYw#e'$rd Z,YwxYw)Iz Tests the parallel backend N)jit vectorize guvectorizeset_num_threads)temp_directoryoverride_configTestCasetagskip_parfors_unsupported linux_only)_TIMEOUT)configgN@)_check_tbb_version_compatible)tbbpoolTF)omppoolzOpenMP threadpool requiredzTBB threadpool requiredGNUzGNU OpenMP only testswindarwinlc2tj||zSN)nponesnvs v/Volumes/fast/ai/experiments/voice-extract-mac/.venv/lib/python3.12/site-packages/numba/tests/test_parallel_backend.pyfoorHs 771:>ctjtj||ftj||f}|tj|z|zSr)rdotrarange)rrxs rlinalgr#MsB FF277Aq6?BGGQFO 4299Q<!##rc2tj||zSr)rr!rs rr#r#Qsyy|arc ||zSr)abs r ufunc_foor)Vs q5Lrc||z|d<y)Nrr&)r'r(outs r gufunc_foor,Zs UCFrceZdZdZy)runnablec ||_yr)_options)selfoptionss r__init__zrunnable.__init___s  rN)__name__ __module__ __qualname__r3r&rrr.r.^s rr.ceZdZdZy) jit_runnerctdi|jt}d}d}t||}|||}tjj ||yN r&)rr0rrtestingassert_allcloser1cfuncr'r(expectedgots r__call__zjit_runner.__call__esN$$dmm$S)  q!9Aqk ""8S1rNr4r5r6rCr&rrr8r8c2rr8ceZdZdZdZy) mask_runnerc ||_||_yr)runnermask)r1rIrJr2s rr3zmask_runner.__init__os  rcf|jrt|j|jyr)rJrrI)r1s rrCzmask_runner.__call__ss 99 DII & rNr4r5r6r3rCr&rrrGrGns rrGceZdZdZy) linalg_runnerctdi|jt}d}d}t||}|||}tjj ||yr:)rr0r#rr=r>r?s rrCzlinalg_runner.__call__}sN$$dmm$V,  !Q<Aqk ""8S1rNrDr&rrrNrN{rErrNceZdZdZy)vectorize_runnerc,tdgfi|jt}tjj dj tj x}}t||}|||}tjj||y)Nz(f4, f4)r<) rr0r)rrandomastypefloat32r=r>r?s rrCzvectorize_runner.__call__sp8 :,8$--8C   $++BJJ77AQ?Aqk ""8S1rNrDr&rrrQrQs2rrQceZdZdZy)guvectorize_runnerc2dg}t|dfi|jt}tjj dj tj x}}t||}|||}tjj||y)Nz(f4, f4, f4[:])z (),()->()r<) rr0r,rrSrTrUr)r=r>)r1sigr@r'r(rArBs rrCzguvectorize_runner.__call__sx !> C> >zJ   $++BJJ77AQ?Aqk ""8S1rNrDr&rrrWrWrErrWc |jd} tjtt t |dzD]}t j|}| y#t$r}|j|Yd}~yd}~wwxYw)Nqueueg?) get faulthandlerenablerangeintlenrSchoice Exceptionput)fnlistkwargsq_fnes rchooserrksp 7As3v;,-.Av&B D/  asAA)) B 2BB cfd}|S)Nc  }d|i}tdDcgc]} t|f|}}|D]}|j|D]}|j|j syg}|j s1|j |j d|j s1d}t|dj|Dcgc] }t|c}zycc}wcc}w)Nr[r;)targetargsrfFz)Error(s) occurred in delegated runner: %s ) r_rkstartjoinemptyappendr\ RuntimeErrorrepr) rergkwsithstherrors_msgr"parallel_class queue_impls r run_compilez$compile_factory..run_compiles Lla" 1WF9SI  "B HHJB GGIwwyFggi aeeEl+ggi?Dtdii&0I&Qa&0I&JJK K  "1Js C+ C0r&)r}r~rs`` rcompile_factoryrsL rceZdZdZdZy)_proc_class_implc||_yr)_method)r1methods rr3z_proc_class_impl.__init__s  rcdtj|j}|j|i|Sr)multiprocessing get_contextrProcess)r1rorfctxs rrCz_proc_class_impl.__call__s,))$,,7s{{D+F++rNrLr&rrrrs ,rrcp|dk(rd}tj|}t|}|j}||fS)Ndefault)rrrQueue)rrprocr[s r_get_mp_classesrs<   % %f -C F #D IIE ;rspawnfork forkserverrceZdZdZededdeddededdededdedddededdedddg Ze s9edd eddd edd eddd gZ eje e jd krgZnd d gZgZeD]!ZeD]Zej%eee#d dgZej%der"ej%dej%dede geddgeddgeedZddhZddZy)TestParallelBackendBasez6 Base class for testing the parallel backends T)nopython)rcache)rnogilparallel)rrn)rrnr)rr)rrr threadingrSmultiprocessing_spawnmultiprocessing_forkmultiprocessing_forkserver)concurrent_jitconcurrent_vectorizeconcurrent_guvectorizeconcurrent_mix_useconcurrent_mix_use_masksomptbbc4t|jj|_t d|j5|dk(r t |n|dk(r t |n|dk(r t|n|dk(r t|n|dk(r t|nt|dk(rat tg}tr*|jt |jttj||D] }|| ntd|zdddy#1swYyxYw) N CACHE_DIRrrrrmultiprocessing_defaultrSzUnknown parallelism supplied %s)r __class__r4 _cache_dirr thread_implfork_proc_implforkserver_proc_implspawn_proc_impldefault_proc_impl _HAVE_OS_FORKrtrSshuffle ValueError)r1re parallelismpsimpls rrz#TestParallelBackendBase.run_compiles()@)@A [$// :k)F# 66v& <<$V, 77' 99!&)(!?3 IIn-II23r"DL!5 CEE+; : :s C DDN)r)r4r5r6__doc__r8rNrQrW all_impls_parfors_unsupported parfor_implsextendrNUMBA_NUM_THREADSmasks mask_implsrrJrtrGrrrunners safe_backendsrr&rrrrs D!D-D-t$t40$'$z:$zFD)D<D4H I t 4 t4 @ 4$ 7 4$d C  & !#AJD   k$5 6)K./1278 4H0H J  d: >!  Z @# ($. GENMErr)rr workqueuezThreading layer not explicitc eZdZdZedZy)TestParallelBackendav These are like the numba.tests.test_threadsafety tests but designed instead to torture the parallel backend. If a suitable backend is supplied via NUMBA_THREADING_LAYER these tests can be run directly. This test class cannot be run using the multiprocessing option to the test runner (i.e. `./runtests -m`) as daemon processes cannot have children. c|jD]O}|jjD]0\}}d|zdz|z}d}|||}||_t |||2Qy)Ntest_rhcfd}|S)Nctj}|jrd}|j|y|j y)Nz)daemonized processes cannot have children)r)rcurrent_processdaemonskipTestr)r1selfprocr|rps r test_methodzBTestParallelBackend.generate..methgen..test_methodJs<#2#B#B#D#??#ND MM$/ ,,Tq,Arr&)rrrs`` rmethgenz-TestParallelBackend.generate..methgenIsB'&r)rritemsr4setattr)clsrnamermethnamerris rgeneratezTestParallelBackend.generateCsdA!kk//1 d"Q;,t3 'T1%& Xr*2!rN)r4r5r6r classmethodrr&rrrr6s++rrcFeZdZeeej dddZdZdZ y)TestInSubprocessFrrrctj|tjtj|}tjt |j } |j|j\}}|jdk7r*td|jd|jd|j|jf|jS#|jwxYw)Nstdoutstderrenvrprocess failed with code : stderr follows rp subprocessPopenPIPErTimer _TEST_TIMEOUTkillrq communicate returncodeAssertionErrordecodecancelr1cmdlinerpopentimeoutr+errs rrun_cmdzTestInSubprocess.run_cmd`s  (2(2%(* //-<  MMO((*HC1$$%%szz|566::<- NN GNN s A;C''C9ctjj}t||d<tj dd|g}|j ||S)NNUMBA_THREADING_LAYERz-mznumba.runtests)osenvironcopystrsys executabler)r1testthreading_layerenv_copyrs rrun_test_in_separate_processz-TestInSubprocess.run_test_in_separate_processssF::??$,/,@()>>4)94@||GX..rN) r4r5r6 skip_no_tbb skip_no_ompunittestskipIfbackendsrrr&rrrr[s)"",X__UB79H&/rrc4eZdZdZdZedZedZy)TestSpecificBackenda This is quite contrived, for each test in the TestParallelBackend tests it generates a test that will run the TestParallelBackend test in a new python process with an environment modified to ensure a specific threadsafe backend is used. This is with view of testing the backends independently and in an isolated manner such that if they hang/crash/have issues, it doesn't kill the test suite. Fc  |j}tj}d|zdz|z}|d|d|  fd}d|d|d} t|| t d||y)Nrrh.cL|j\}}|jrtd|d|dtjd|}| |j |j d|jd||jd|v|jd|vy) Nz stdout: "z " stderr: ""z\.\.\. skipped '(.*?)'rOKFAILERROR) r_DEBUGprintresearchrgroupassertIn assertTrue)r1orjmbackendinjected_methods r test_templatez2TestSpecificBackend._inject..test_templates44_gNDAq{{1a@A 3Q7A} aggaj) MM$ " OOF!O , OOG1, -r long_running)r5rr4rr ) rrrr backend_guardthemodtheclsrr injected_testrs ` @r_injectzTestSpecificBackend._injectsj$--Q;$t+(.A .,-dG< ]#N#M-$@A Crc 6|jjD]|\}}|jD]h}|jj D]I}|dvr%|dk(r t j jdr,|dvr|dk(r6|j||||Kj~y)N)rrSrlinux)rrSr) rrrrkeysrplatform startswithr)rrrrrs rrzTestSpecificBackend.generates&)ll&8&8&: "G]__KK,,.D??5(LL33G< 44#{2 KK4-@/%';rN)r4r5r6rr rrrr&rrrrzs8FCC.AArrcbeZdZdZdej j ezZddeizZ ddZ y)ThreadLayerTestHelperzP Helper class for running an isolated piece of code based on a template z%raif 1: import sys sys.path.insert(0, "%(here)r") import multiprocessing import numpy as np from numba import njit import numba try: import threading_backend_usecases except ImportError as e: print("DEBUG:", sys.path) raise e import os sigterm_handler = threading_backend_usecases.sigterm_handler busy_func = threading_backend_usecases.busy_func def the_test(): %%s if __name__ == "__main__": the_test() hereNcV|,tjj}td|d<t j |tj tj |}tjt|j} |j|j\}}|jdk7r*td|jd|jd |j!|j|jfS#|j!wxYw)Nrrrrrrrp)rrrrrrrrrrrrqrrrrrrs rrzThreadLayerTestHelper.run_cmds ;**//#C+.u:C' (  (2(2%(* //-<  MMO((*HC1$$%%szz|566% NN zz|SZZ\)) NN s ADD(r) r4r5r6rrpathdirname__file___heretemplaterr&rrr$r$s: 277??8, ,E, 5/-H0*rr$cbeZdZdZdZeeejdddZ e dZ e dZ y)TestThreadingLayerSelectionz@ Checks that numba.threading_layer() reports correctly. Frrc ^fd}dz}t||td||y)Nc d}|j|zz}tjd|g}tjj }t |d<|j||\}}|jr t||yy)Nzif 1: X = np.arange(1000000.) Y = np.arange(1000000.) Z = busy_func(X, Y) assert numba.threading_layer() == '%s' -crr) r+rrrrrrrr r )r1bodyrunmerrr+rrs rrz:TestThreadingLayerSelection._inject..test_templatesxD MMTG^4E~~tU3G**//#C+.w @rcl|jjD]\}}|j||yr)rrr)rrrs rrz$TestThreadingLayerSelection.generate s,&)ll&8&8&: "G] KK /';rN) r4r5r6rr rrrrrrrrr&rrr-r-sUF"",X__UB79H@@(00rr-creZdZdefdZeedZeedZedZ edZ dZ y) TestThreadingLayerPriorityenv_varctjj}d|d<||d<d|d}tjdt j |g}|j||y) zJTest setting priority via env var NUMBA_THREADING_LAYER_PRIORITY. rrNUMBA_THREADING_LAYER_PRIORITYa import numba # trigger threading layer decision # hence catching invalid THREADING_LAYER_PRIORITY @numba.jit( 'float64[::1](float64[::1], float64[::1])', nopython=True, parallel=True, ) def plus(x, y): return x + y captured_envvar = list("a ".split()) assert numba.config.THREADING_LAYER_PRIORITY == captured_envvar, "priority mismatch" assert numba.threading_layer() == captured_envvar[0], "selected backend mismatch" r0r1N)rrrrrtextwrapdedentr)r1r8rcodecmds r each_env_varz'TestThreadingLayerPriority.each_env_varssjjoo'0 #$07 ,- )*1 2( NN  OOD !  Sc "rcgd}tj|D]$}dj|}|j|&y)Nr ) itertools permutationsrrr?)r1rrr8s rtest_valid_env_varz-TestThreadingLayerPriority.test_valid_env_var8s9.''0AhhqkG   g &1rcd}|jt5}|j|ddddD](}|j|t j *y#1swY7xYw)Nztbb omp workqueue notvalidhere)z!THREADING_LAYER_PRIORITY invalid:zIt must be a permutation of) assertRaisesrr?rr exception)r1r8raisesmsgs rtest_invalid_env_varz/TestThreadingLayerPriority.test_invalid_env_var@s^3   ~ .&   g &/ C MMSEC(8(8$9 :  / .s A  A)c4dD]}|j|y)N)zomp tbb workqueuezomp workqueue tbbr?r1r8s rtest_ompz#TestThreadingLayerPriority.test_ompLAG   g &Brc4dD]}|j|y)N)ztbb omp workqueueztbb workqueue omprLrMs rtest_tbbz#TestThreadingLayerPriority.test_tbbQrOrc4dD]}|j|y)N)zworkqueue tbb ompzworkqueue omp tbbrLrMs rtest_workqueuez)TestThreadingLayerPriority.test_workqueueVsAG   g &BrN) r4r5r6rr?rrrDrJrNrQrSr&rrr7r7st#C#B'' ;;'''''rr7cleZdZdZdZedZedZdZ e je ddZ y) TestMiscBackendIssueszL Checks fixes for the issues with threading backends implementation Fcd}tjd|g}tjj }d|d<d|d<|j ||y) z8 Tests that OMP does not overflow stack aif 1: from numba import vectorize, threading_layer import numpy as np @vectorize(['f4(f4,f4,f4,f4,f4,f4,f4,f4)'], target='parallel') def foo(a, b, c, d, e, f, g, h): return a+b+c+d+e+f+g+h x = np.ones(2**20, np.float32) foo(*([x]*8)) assert threading_layer() == "omp", "omp not found" r0rr100K OMP_STACKSIZEr1Nrrrrrrr1r3rrs rtest_omp_stack_overflowz-TestMiscBackendIssues.test_omp_stack_overflowbsP  >>4/jjoo', #$%O W# &rcd}tjd|g}tjj }d|d<d|d<|j ||y) zq Tests that TBB works well with single thread https://github.com/numba/numba/issues/3440 aMif 1: from numba import njit, prange, threading_layer @njit(parallel=True) def foo(n): acc = 0 for i in prange(n): acc += i return acc foo(100) assert threading_layer() == "tbb", "tbb not found" r0rr1rr1NrYrZs rtest_single_thread_tbbz,TestMiscBackendIssues.test_single_thread_tbbysQ  >>4/jjoo', #$#&   W# &rcjd}tjd|g}tjj }d|d<d|d< |j ||\}}y #t $rS}|jr tt|}|jd|d }|j||Yd }~y d }~wwxYw) zW Tests workqueue raises sigabrt if a nested parallel call is performed aif 1: from numba import njit, prange import numpy as np @njit(parallel=True) def nested(x): for i in prange(len(x)): x[i] += 1 @njit(parallel=True) def main(): Z = np.zeros((5, 10)) for i in prange(Z.shape[0]): nested(Z[i]) return Z main() r0rr4rr1zfailed with codezTNumba workqueue threading layer is terminating: Concurrent access has been detected.N) rrrrrrrr r rr) r1r3rrr+rrje_msgrAs r+test_workqueue_aborts_on_nested_parallelismzATestMiscBackendIssues.test_workqueue_aborts_on_nested_parallelisms &>>4/jjoo'2 #$#&   +||G|5HC +{{c3FE MM,e 4?H MM(E * * +sA B2A B--B2zTest needs fork(2)cd}tjd|g}tjj }d|d<d|d<|j ||y)Naif 1: from numba import njit, prange, threading_layer import numpy as np import multiprocessing if __name__ == "__main__": # Need for force fork context (OSX default is "spawn") multiprocessing.set_start_method('fork') @njit(parallel=True) def func(x): return 10. * x arr = np.arange(2.) # run in single process to start Numba's thread pool np.testing.assert_allclose(func(arr), func.py_func(arr)) # now run in a multiprocessing pool to get a fork from a # non-main thread with multiprocessing.Pool(10) as p: result = p.map(func, [arr]) np.testing.assert_allclose(result, func.py_func(np.expand_dims(arr, 0))) assert threading_layer() == "workqueue" r0rrr`rr1rYrZs r0test_workqueue_handles_fork_from_non_main_threadzFTestMiscBackendIssues.test_workqueue_handles_fork_from_non_main_threadsQ 6>>4/jjoo'2 #$#&   W# &rN)r4r5r6rr rr[rr^rbr skipUnlessrrdr&rrrUrU[s^F'',''0'+RX(<=&'>&'rrUcReZdZdZdZdZdZdZedZ dZ dZ ed Z y ) TestForkSafetyIssueszV Checks Numba's behaviour in various situations involving GNU OpenMP and fork FcVd}tjd|g}|j|\}}y)Nzsif 1: from numba.np.ufunc import omppool assert omppool.openmp_vendor == 'GNU' r0)rrr)r1r3rr+rs r!test_check_threading_layer_is_gnuz6TestForkSafetyIssues.test_check_threading_layer_is_gnus->>4/<<(Srcd}|j|z}tjd|g} |j|\}}y#t$r%}|j dt |Yd}~yd}~wwxYw)z~ Whilst normally valid, this actually isn't for Numba invariant of OpenMP Checks SIGABRT is received. zif 1: X = np.arange(1000000.) Y = np.arange(1000000.) Z = busy_func(X, Y) pid = os.fork() if pid == 0: Z = busy_func(X, Y) else: os.wait() r0zfailed with code -6N)r+rrrrrr)r1r2r3rr+rrjs r!test_par_parent_os_fork_par_childz6TestForkSafetyIssues.test_par_parent_os_fork_par_childsf   $>>4/ 9||G,HC 9 MM/Q 8 8 9s; A)A$$A)cd}|j|z}tjd|g}|j|\}}|jr t ||yy)au Implicit use of multiprocessing fork context. Does this: 1. Start with OpenMP 2. Fork to processes using OpenMP (this is invalid) 3. Joins fork 4. Check the exception pushed onto the queue that is a result of catching SIGTERM coming from the C++ aborting on illegal fork pattern for GNU OpenMP aif 1: mp = multiprocessing.get_context('fork') X = np.arange(1000000.) Y = np.arange(1000000.) q = mp.Queue() # Start OpenMP runtime on parent via parallel function Z = busy_func(X, Y, q) # fork() underneath with no exec, will abort proc = mp.Process(target = busy_func, args=(X, Y, q)) proc.start() err = q.get() assert "Caught SIGTERM" in str(err) r0Nr+rrrr r r1r2r3rr+rs r*test_par_parent_implicit_mp_fork_par_childz?TestForkSafetyIssues.test_par_parent_implicit_mp_fork_par_child sP   $>>4/<<(S ;; #sO rcd}|j|z}tjd|g}|j|\}}|jr t ||yy)au Explicit use of multiprocessing fork context. Does this: 1. Start with OpenMP 2. Fork to processes using OpenMP (this is invalid) 3. Joins fork 4. Check the exception pushed onto the queue that is a result of catching SIGTERM coming from the C++ aborting on illegal fork pattern for GNU OpenMP aif 1: X = np.arange(1000000.) Y = np.arange(1000000.) ctx = multiprocessing.get_context('fork') q = ctx.Queue() # Start OpenMP runtime on parent via parallel function Z = busy_func(X, Y, q) # fork() underneath with no exec, will abort proc = ctx.Process(target = busy_func, args=(X, Y, q)) proc.start() proc.join() err = q.get() assert "Caught SIGTERM" in str(err) r0Nrmrns r*test_par_parent_explicit_mp_fork_par_childz?TestForkSafetyIssues.test_par_parent_explicit_mp_fork_par_child-sP " $>>4/<<(S ;; #sO rcd}|j|z}tjd|g}|j|\}}|jr t ||yy)z Explicit use of multiprocessing spawn, this is safe. Does this: 1. Start with OpenMP 2. Spawn to processes using OpenMP 3. Join spawns 4. Run some more OpenMP aif 1: X = np.arange(1000000.) Y = np.arange(1000000.) ctx = multiprocessing.get_context('spawn') q = ctx.Queue() # Start OpenMP runtime and run on parent via parallel function Z = busy_func(X, Y, q) procs = [] for x in range(20): # start a lot to try and get overlap ## fork() + exec() to run some OpenMP on children proc = ctx.Process(target = busy_func, args=(X, Y, q)) procs.append(proc) sys.stdout.flush() sys.stderr.flush() proc.start() [p.join() for p in procs] try: q.get(False) except multiprocessing.queues.Empty: pass else: raise RuntimeError("Queue was not empty") # Run some more OpenMP on parent Z = busy_func(X, Y, q) r0Nrmrns r-test_par_parent_mp_spawn_par_child_par_parentzBTestForkSafetyIssues.test_par_parent_mp_spawn_par_child_par_parentPsP : $>>4/<<(S ;; #sO rcd}|j|z}tjd|g}|j|\}}|jr t ||yy)a Implicit use of multiprocessing (will be fork, but cannot declare that in Py2.7 as there's no process launch context). Does this: 1. Start with no OpenMP 2. Fork to processes using OpenMP 3. Join forks 4. Run some OpenMP a,if 1: X = np.arange(1000000.) Y = np.arange(1000000.) q = multiprocessing.Queue() # this is ok procs = [] for x in range(10): # fork() underneath with but no OpenMP in parent, this is ok proc = multiprocessing.Process(target = busy_func, args=(X, Y, q)) procs.append(proc) proc.start() [p.join() for p in procs] # and this is still ok as the OpenMP happened in forks Z = busy_func(X, Y, q) try: q.get(False) except multiprocessing.queues.Empty: pass else: raise RuntimeError("Queue was not empty") r0Nrmrns r=test_serial_parent_implicit_mp_fork_par_child_then_par_parentzRTestForkSafetyIssues.test_serial_parent_implicit_mp_fork_par_child_then_par_parent|P 2 $>>4/<<(S ;; #sO rcd}|j|z}tjd|g}|j|\}}|jr t ||yy)z Explicit use of multiprocessing 'fork'. Does this: 1. Start with no OpenMP 2. Fork to processes using OpenMP 3. Join forks 4. Run some OpenMP aif 1: X = np.arange(1000000.) Y = np.arange(1000000.) ctx = multiprocessing.get_context('fork') q = ctx.Queue() # this is ok procs = [] for x in range(10): # fork() underneath with but no OpenMP in parent, this is ok proc = ctx.Process(target = busy_func, args=(X, Y, q)) procs.append(proc) proc.start() [p.join() for p in procs] # and this is still ok as the OpenMP happened in forks Z = busy_func(X, Y, q) try: q.get(False) except multiprocessing.queues.Empty: pass else: raise RuntimeError("Queue was not empty") r0Nrmrns r=test_serial_parent_explicit_mp_fork_par_child_then_par_parentzRTestForkSafetyIssues.test_serial_parent_explicit_mp_fork_par_child_then_par_parentrvrN) r4r5r6rr rirkror rqrsrurxr&rrrgrgsSF)9,B  D*X'R&&rrgc0eZdZdZedZedZy)TestTBBSpecificIssuesFcd}tjd|g}|j|\}}d}|j|||jrt d|t d|yy)Na%if 1: import threading import numba numba.config.THREADING_LAYER='tbb' from numba import njit, prange, objmode from numba.core.serialize import PickleCallableByPath import os e_running = threading.Event() e_proceed = threading.Event() def indirect_core(): e_running.set() # wait for forker() to have forked while not e_proceed.isSet(): pass indirect = PickleCallableByPath(indirect_core) @njit def obj_mode_func(): with objmode(): indirect() @njit(parallel=True, nogil=True) def work(): acc = 0 for x in prange(10): acc += x obj_mode_func() return acc def runner(): work() def forker(): # wait for the jit function to say it's running while not e_running.isSet(): pass # then fork os.fork() # now fork is done signal the runner to proceed to exit e_proceed.set() numba_runner = threading.Thread(target=runner,) fork_runner = threading.Thread(target=forker,) threads = (numba_runner, fork_runner) for t in threads: t.start() for t in threads: t.join() r0z9Attempted to fork from a non-main thread, the TBB libraryOUT:ERR:)rrrrr r )r1r3rr+rmsg_heads rtest_fork_from_non_main_threadz4TestTBBSpecificIssues.test_fork_from_non_main_threadsa64 l>>4/<<(SN h$ ;; &#  &#  rc~|jd}djd|}tjd|g}tj j }d|d<|j||\}}||vr|jdn|jd ||jrtd |td |yy) NzSKIP: COMPILATION FAILEDax if 1: import ctypes import sys import multiprocessing as mp from tempfile import TemporaryDirectory, NamedTemporaryFile from numba.pycc.platform import Toolchain, external_compiler_works from numba import njit, prange, threading_layer import faulthandler faulthandler.enable() if not external_compiler_works(): raise AssertionError('External compilers are not found.') with TemporaryDirectory() as tmpdir: with NamedTemporaryFile(dir=tmpdir) as tmpfile: try: src = """ #define TBB_PREVIEW_WAITING_FOR_WORKERS 1 #include static tbb::task_scheduler_handle tsh; extern "C" { void launch(void) { tsh = tbb::task_scheduler_handle::get(); } } """ cxxfile = f"{tmpfile.name}.cxx" with open(cxxfile, 'wt') as f: f.write(src) tc = Toolchain() object_files = tc.compile_objects([cxxfile,], output_dir=tmpdir) dso_name = f"{tmpfile.name}.so" tc.link_shared(dso_name, object_files, libraries=['tbb',], export_symbols=['launch']) # Load into the process, it doesn't matter whether the # DSO exists on disk once it's loaded in. DLL = ctypes.CDLL(dso_name) except Exception as e: # Something is broken in compilation, could be one of # many things including, but not limited to: missing tbb # headers, incorrect permissions, compilers that don't # work for the above print(e) print('BROKEN_COMPILERS') sys.exit(0) # Do the test, launch this library and also execute a # function with the TBB threading layer. DLL.launch() @njit(parallel=True) def foo(n): acc = 0 for i in prange(n): acc += i return acc foo(1) # Check the threading layer used was TBB assert threading_layer() == 'tbb' # Use mp context for a controlled version of fork, this triggers the # reported bug. ctx = mp.get_context('fork') def nowork(): pass p = ctx.Process(target=nowork) p.start() p.join(10) print("SUCCESS") BROKEN_COMPILERSr0rrr1z3Compilation of DSO failed. Check output for detailsSUCCESSr|r}) skip_if_no_external_compilerreplacerrrrrrrrr r )r1rr3rrr+rs r&test_lifetime_of_task_scheduler_handlez>4/jjoo', #$<<S<1S s " MMO P MM)S ) ;; &#  &#  rN)r4r5r6r r rrr&rrrzrzs4FXXt__rrzc,eZdZdZdZedZdZy)TestInitSafetyIssuesFctj|tjtj}tjt |j } |j|j\}}|jdk7r*td|jd|jd |j|j|jfS#|jwxYw)N)rrrrrrpr)r1rrrr+rs rrzTestInitSafetyIssues.run_cmds  (2(2://-<  MMO((*HC1$$%%szz|566% NN zz|SZZ\)) NN s AC''C9c@tjjtjjtd}t j |g}|j|\}}|jd||jrtd|td|yy)Nzorphaned_semaphore_usecase.pyzleaked semaphorer|r}) rr'rrr(r)rrr assertNotInr r )r1 test_filerr+rs rtest_orphaned_semaphorez,TestInitSafetyIssues.test_orphaned_semaphores| GGLL!:!@B >>9-<<(S +S1 ;; &#  &#  rc8dD]} tj|d}tjd|j |g}|j |\}}|jrtd|td||j||y#t$rYwxYw)N)rrrzimport numba; import multiprocessing;multiprocessing.set_start_method('{}');print(multiprocessing.get_context().get_start_method())r0r|r}) rrrrrformatrr r r)r1methr>rr+rs rtest_lazy_lock_initz(TestInitSafetyIssues.test_lazy_lock_inits4D ++D1MC~~tSZZ-=>G||G,HC{{fc"fc" MM$ $4   sB  BBN)r4r5r6r rr rrr&rrrrs'F*" %rrceZdZdZy)TestOpenMPVendorsct}d|d<d|d<d|d<|jD]E}tjj |s#|j ||t jGy)z> Checks the OpenMP vendor strings are correct MSwin32IntelrrrN)dictr rr!r" assertEqualr openmp_vendor)r1rAks r test_vendorszTestOpenMPVendors.test_vendorssg6 $!A||&&q)  !g.C.CD!rN)r4r5r6rr&rrrrs Err__main__)crr]rBrrrSrrrr;rrnumpyrnumbarrrrnumba.tests.supportrrr r r r r[t_queuenumba.testing.mainr _RUNNER_TIMEOUT numba.corerrnumba.np.ufunc.parallelrnumba.np.ufuncr_HAVE_TBB_POOL ImportErrorr_HAVE_OMP_POOLscipy.linalg.cython_lapackscipy _HAVE_LAPACKrerrr_gnuompskip_unless_gnu_ompr!r"_windows_osxmaxsize_32bitrrrr#r)r,objectr.r8rGrNrQrWrkrThread _thread_classrrrrrrrrrTHREADING_LAYER_specific_backendsrrrrr$r-r7rUrgrzrrr4mainr&rrrs  >>GG: #% E!#&N&N%L "h!!.2NO !h!!.2KL  ;W22e;)h))'3JK << " "5 ) ||x(   $  v 22 & 2H22x222(  ,v,mW]]; !?7#;<$of&=>N*OL,IJ$_Y%?@UEhUEp++/JJ')GH+1+I+B/v/>5A*,C5Ap3*H3*l"0"7"0"0J$$&C'!6C'C'LF'1F'F'Td0ddN 1 D9%89%9%x EE E$ zHMMOy%N N Ls6%K8;LL8LLLLLL