gdZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl mZmZmZmZddlmZmZmZmZmZmZddlZddlmZddlmZedz Z ddl!m"Z"e"dd l#m$Z$d Z%n #e&$rd Z%YnwxYw dd l#m'Z'd Z(n #e&$rd Z(YnwxYw ddl)Z*d Z+n #e&$rd Z+YnwxYwe j,e(d Z-e j,e%dZ.e(o e'j/dkZ0e j,e0dZ1ej23dZ4ej23dZ5ej6dkZ7e7Z8e4 Z9dZ:e+rdZ;ndZ;dZZ?Gdde?Z@Gdde>ZAGdd e?ZBGd!d"e?ZCGd#d$e?ZDd%ZEd&ZFe jGZHGd'd(e>ZId)ZJeFeHejKZLeFeJd*ZMe4seFeJd+ZNeFeJd,ZOeFeJd-ZPGd.d/eZQejRd0vZSe j,eSd1Gd2d3eQZTeTUGd4d5e>ZVGd6d7eVeQZWeWUGd8d9eZXeGd:d;eXZYeYUeGd<d=eXZZeGd>d?eXZ[ee1Gd@dAeXZ\ee.GdBdCeXZ]eGdDdEeZ^ee-GdFdGeZ_e`dHkre jadSdS)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 testswindarwinlc0tj||zSN)nponesnvs W/granary/edev/quantum/lib/python3.11/site-packages/numba/tests/test_parallel_backend.pyfoorGs 71::>ctjtj||ftj||f}|tj|z|zSr)rdotrarange)rrxs rlinalgr#LsC F27Aq6??BGQFOO 4 429Q<<!##rc0tj||zSr)rr!rs rr#r#Psy||arc ||zSr)abs r ufunc_foor)Us q5Lrc||z|d<dS)Nrr&)r'r(outs r gufunc_foor,Ys UCFFFrceZdZdZdS)runnablec ||_dSr)_options)selfoptionss r__init__zrunnable.__init__^s  rN)__name__ __module__ __qualname__r3r&rrr.r.]s#     rr.ceZdZdZdS) jit_runnerctdi|jt}d}d}t||}|||}tj||dSN r&)rr0rrtestingassert_allcloser1cfuncr'r(expectedgots r__call__zjit_runner.__call__dse$$$dm$$S))  q!99eAqkk ""8S11111rNr4r5r6rCr&rrr8r8b#22222rr8ceZdZdZdZdS) mask_runnerc "||_||_dSr)runnermask)r1rIrJr2s rr3zmask_runner.__init__ns  rcd|jrt|j|dSr)rJrrI)r1s rrCzmask_runner.__call__rs/ 9 ' DI & & & rNr4r5r6r3rCr&rrrGrGms2rrGceZdZdZdS) linalg_runnerctdi|jt}d}d}t||}|||}tj||dSr:)rr0r#rr=r>r?s rrCzlinalg_runner.__call__|se$$$dm$$V,,  !Q<<eAqkk ""8S11111rNrDr&rrrNrNzrErrNceZdZdZdS)vectorize_runnerc<tdgfi|jt}tjdtjx}}t||}|||}tj||dS)Nz(f4, f4)r<) rr0r)rrandomastypefloat32r=r>r?s rrCzvectorize_runner.__call__s8 :,88$-88CC   $$++BJ777AQ??eAqkk ""8S11111rNrDr&rrrQrQs#22222rrQceZdZdZdS)guvectorize_runnercBdg}t|dfi|jt}tjdtjx}}t||}|||}tj ||dS)Nz(f4, f4, f4[:])z (),()->()r<) rr0r,rrSrTrUr)r=r>)r1sigr@r'r(rArBs rrCzguvectorize_runner.__call__s !> C>> >>zJJ   $$++BJ777AQ??eAqkk ""8S11111rNrDr&rrrWrWrErrWc L|d} tjtt t |dzD] }t j|}|!dS#t$r }| |Yd}~dSd}~wwxYw)Nqueueg?) get faulthandlerenablerangeintlenrSchoice Exceptionput)fnlistkwargsq_fnes rchooserrks 7As3v;;,--..  Av&&B BDDDD    asA A99 B#BB#cfd}|S)Nc}d|ifdtdD}|D]}||D]}||sg}|s<||d|z8compile_factory..run_compile..s<"""~WF9SIII"""rr;Fz)Error(s) occurred in delegated runner: %s c,g|]}t|Sr&)repr)rqr"s rruz8compile_factory..run_compile..s0I0I0IQa0I0I0Ir)r_startjoinemptyappendr\ RuntimeError) rergthstherrors_msgrsrt queue_impls ` @r run_compilez$compile_factory..run_compiles  JLLl""""""a"""  B HHJJJJ  B GGIIIIwwyy LFggii , aeeEll+++ggii ,?Dtdii0I0I&0I0I0I&J&JJKK K  L Lrr&)rtrrs`` rcompile_factoryrs0LLLLLL rceZdZdZdZdS)_proc_class_implc||_dSr)_method)r1methods rr3z_proc_class_impl.__init__s  rcNtj|j}|j|i|Sr)multiprocessing get_contextrProcess)r1rprfctxs rrCz_proc_class_impl.__call__s+)$,77s{D+F+++rNrLr&rrrrs2,,,,,rrcn|dkrd}tj|}t|}|j}||fS)Ndefault)rrrQueue)rrprocr[s r_get_mp_classesrs@   %f - -C F # #D IE ;rspawnfork forkserverrceZdZdZededdeddededdededdedddededdedddg Ze sIedd eddd edd eddd gZ e e e j d krgZnd d gZgZeD]&ZeD]!Zeeee"'d dgZeder*ededede geddgeddgeedZddhZddZdS)TestParallelBackendBasez6 Base class for testing the parallel backends T)nopython)rcache)rnogilparallel)rro)rror)rr)rrr threadingrSmultiprocessing_spawnmultiprocessing_forkmultiprocessing_forkserver)concurrent_jitconcurrent_vectorizeconcurrent_guvectorizeconcurrent_mix_useconcurrent_mix_use_masksomptbbct|jj|_t d|j5|dkrt |n|dkrt |n|dkrt|n|dkrt|n|dkrt|n|dkrnt tg}tr4| t | ttj ||D] }||ntd|zddddS#1swxYwYdS) N CACHE_DIRrrrrmultiprocessing_defaultrSzUnknown parallelism supplied %s)r __class__r4 _cache_dirr thread_implfork_proc_implforkserver_proc_implspawn_proc_impldefault_proc_impl _HAVE_OS_FORKr|rSshuffle ValueError)r1re parallelismpsimpls rrz#TestParallelBackendBase.run_compiles()@AA [$/ : : E Ek))F#### 666v&&&& <<<$V,,,, 777'''' 999!&))))((!?3 4IIn---II2333r"""!!DDLLLL!!5 CEEE+ E E E E E E E E E E E E E E E E E EsC5D66D:=D:N)r)r4r5r6__doc__r8rNrQrW all_impls_parfors_unsupported parfor_implsextendrNUMBA_NUM_THREADSmasks mask_implsrrJr|rGrrrunners safe_backendsrr&rrrrs  D!!! D--- D--- t$$$ t4000$'''$z:::$zFFFD)))D<<<D4HHH I ' Jt 4 4 4 Jt4 @ @ @ M4$ 7 7 7 M4$d C C C  &&& !##AJ77 7 7D   kk$55 6 6 6 6 7)K.///912227888 J4H0H J J J   d: > > >!   Z @ @ @# ($.  GENMEEEEEErr)rr workqueuezThreading layer not explicitc(eZdZdZedZdS)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]S}|jD]7\}}d|zdz|z}d}|||}||_t |||8TdS)Ntest_rhcfd}|S)Nctj}|jrd}||dS|dS)Nz)daemonized processes cannot have children)r)rcurrent_processdaemonskipTestr)r1selfprocrrps r test_methodzBTestParallelBackend.generate..methgen..test_methodIsX#2#B#D#D#?B#ND MM$///// ,,Tq,AAAAArr&)rrrs`` rmethgenz-TestParallelBackend.generate..methgenHs0BBBBBB'&r)rritemsr4setattr)clsrnamermethnamerris rgeneratezTestParallelBackend.generateBs + +A!k//11 + + d"Q;,t3 ' ' 'WT1%%& Xr**** + + +rN)r4r5r6r classmethodrr&rrrr5s9++[+++rrcDeZdZeeejdddZdZdZ dS)TestInSubprocessFrrrc tj|tjtj|}tjt |j} ||\}}|j dkr-td|j d| d| | f| S#| wxYw)Nstdoutstderrenvrprocess failed with code : stderr follows rv subprocessPopenPIPErTimer _TEST_TIMEOUTkillry communicate returncodeAssertionErrordecodecancelr1cmdlinerpopentimeoutr+errs rrun_cmdzTestInSubprocess.run_cmd_s (2(2%(*** /-<<  MMOOO((**HC1$$$n%%%szz||||5666::<<- NN    GNN    s B C,,Dctj}t||d<tjdd|g}|||S)NNUMBA_THREADING_LAYERz-mznumba.runtests)osenvironcopystrsys executabler)r1testthreading_layerenv_copyrs rrun_test_in_separate_processz-TestInSubprocess.run_test_in_separate_processrsK:??$$,/,@,@()>4)94@||GX...rN) r4r5r6 skip_no_tbb skip_no_ompunittestskipIfbackendsrrr&rrrrZsU"",X_UB7799H&/////rrcBeZdZdZdZedZedZdS)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||dS)Nrrh.c|\}}|jrtd|d|d|d||d|v|d|vdS)Nz stdout: "z " stderr: ""OKFAILERROR)r_DEBUGprintassertIn assertTrue)r1orjbackendinjected_methods r test_templatez2TestSpecificBackend._inject..test_templates44_gNNDAq{ B111aaa@AAA MM$ " " " OOF!O , , , OOG1, - - - - -r long_running)r5rr4rr ) rrrr backend_guardthemodtheclsrr injected_testrs ` @r_injectzTestSpecificBackend._injects$-Q;$t+(.A . . . . . . .,-11dddGG< ]#N##MM-$@$@AA C C C C Crc *|jD]x\}}|jD]k}|jD]O}|dvr&|dkr t jdr,|dvr|dkr7|||||PlydS)N)rrSrlinux)rrSr) rrrrkeysrplatform startswithr)rrrrrs rrzTestSpecificBackend.generates&)l&8&8&:&: A A "G]_ A AK,,..AAD???5((L33G<<) 444#{22 KK4-@@@@A A A ArN)r4r5r6rrrrrr&rrrrysaFCC[C$AA[AAArrcbeZdZdZdejezZddeizZ ddZ dS)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() hereNcr|0tj}td|d<t j|tjtj|}tjt|j } | | \}}|j dkr-td|j d|d |n#|wxYw||fS)Nrrrrrrrv)rrrrrrrrrrrryrrrrrrs rrzThreadLayerTestHelper.run_cmds ;*//##C+.u::C' ( (2(2%(*** /-<<  MMOOO((**HC1$$$n%%%szz||||5666% NN    GNN    zz||SZZ\\))s ?A#C88Dr) r4r5r6rrpathdirname__file___heretemplaterr&rrr$r$sa 27??8,, ,E, 5/-H0******rr$cleZdZdZdZeeejdddZ e dZ e dZ dS)TestThreadingLayerSelectionz@ Checks that numba.threading_layer() reports correctly. Frrc |fd}dz}t||td||dS)Nc d}|j|zz}tjd|g}tj}t |d<|||\}}|jrt||dSdS)Nzif 1: X = np.arange(1000000.) Y = np.arange(1000000.) Z = busy_func(X, Y) assert numba.threading_layer() == '%s' -crr) r+rrrrrrrrr)r1bodyrunmerrr+rrs rrz:TestThreadingLayerSelection._inject..test_templatesD MTG^4E~tU3G*//##C+.w<> @ @ @ @ @rcp|jD]\}}|||dSr)rrr)rrrs rrz$TestThreadingLayerSelection.generatesE&)l&8&8&:&: 0 0 "G] KK / / / / 0 0rN) r4r5r6rrrrrrrrrrr&rrr-r-sF"",X_UB7799H@@[@(00[000rr-ceZdZdefdZeedZeedZedZ edZ dZ dS) TestThreadingLayerPriorityenv_varctj}d|d<||d<d|d}tjdt j|g}|||dS) 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_varsjoo'0 #$07 ,-*1( N  OD ! !  Sc """""rcgd}tj|D],}d|}||-dS)Nr ) itertools permutationsrzr?)r1rrr8s rtest_valid_env_varz-TestThreadingLayerPriority.test_valid_env_var2sY.--'00 ' 'AhhqkkG   g & & & & ' 'rcd}|t5}||dddn #1swxYwYdD]+}||t |j,dS)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:s3   ~ . . '&   g & & & ' ' ' ' ' ' ' ' ' ' ' ' ' ' '  ; ;C MMS(C(8$9$9 : : : :  ; ;s?AAc:dD]}||dS)N)zomp tbb workqueuezomp workqueue tbbr?r1r8s rtest_ompz#TestThreadingLayerPriority.test_ompF2A ' 'G   g & & & & ' 'rc:dD]}||dS)N)ztbb omp workqueueztbb workqueue omprLrMs rtest_tbbz#TestThreadingLayerPriority.test_tbbKrOrc:dD]}||dS)N)zworkqueue tbb ompzworkqueue omp tbbrLrMs rtest_workqueuez)TestThreadingLayerPriority.test_workqueuePs2A ' 'G   g & & & & ' 'rN) r4r5r6rr?rrrDrJrNrQrSr&rrr7r7s#C####B''[[' ;;[[;''['''[''''''rr7c|eZdZdZdZedZedZdZ e j e ddZ dS) TestMiscBackendIssueszL Checks fixes for the issues with threading backends implementation Fcd}tjd|g}tj}d|d<d|d<|||dS) 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_overflow\s[  >4/joo', #$%O W# &&&&&rcd}tjd|g}tj}d|d<d|d<|||dS) 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_tbbss\  >4/joo', #$#&   W# &&&&&rcd}tjd|g}tj}d|d<d|d< |||\}}d S#t $r_}|jrt||t|}| d|d }| ||Yd }~d Sd }~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) rrrrrrrrrrr) r1r3rrr+rrje_msgrAs r+test_workqueue_aborts_on_nested_parallelismzATestMiscBackendIssues.test_workqueue_aborts_on_nested_parallelisms &>4/joo'2 #$#&   +||G|55HC + + +{ c3FFE MM,e 4 4 4?H MM(E * * * * * * * * * +sA C!AB;;CzTest needs fork(2)cd}tjd|g}tj}d|d<d|d<|||dS)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_threads\ 6>4/joo'2 #$#&   W# &&&&&rN)r4r5r6rrrr[rr^rbr skipUnlessrrdr&rrrUrUUsF''[',''['0'+'+'+RX(<==&'&'>=&'&'&'rrUc`eZdZdZdZdZdZdZedZ dZ dZ ed Z d S) TestForkSafetyIssueszV Checks Numba's behaviour in various situations involving GNU OpenMP and fork FcXd}tjd|g}||\}}dS)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_gnus2>4/<<((SSSrcd}|j|z}tjd|g} ||\}}dS#t$r.}|dt |Yd}~dSd}~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_childs   $>4/ 9||G,,HC 9 9 9 MM/Q 8 8 8 8 8 8 8 8 8 9s7 A/#A**A/cd}|j|z}tjd|g}||\}}|jrt ||dSdS)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+rrrrrr1r2r3rr+rs r*test_par_parent_implicit_mp_fork_par_childz?TestForkSafetyIssues.test_par_parent_implicit_mp_fork_par_childs_   $>4/<<((S ;  #sOOOOO  rcd}|j|z}tjd|g}||\}}|jrt ||dSdS)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's_ " $>4/<<((S ;  #sOOOOO  rcd}|j|z}tjd|g}||\}}|jrt ||dSdS)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_parentJs_ : $>4/<<((S ;  #sOOOOO  rcd}|j|z}tjd|g}||\}}|jrt ||dSdS)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_parentv_ 2 $>4/<<((S ;  #sOOOOO  rcd}|j|z}tjd|g}||\}}|jrt ||dSdS)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) r4r5r6rrrirkror rqrsrurxr&rrrgrgsF)))999,B  Z D***X'''R&&Z&&&rrgc>eZdZdZedZedZdS)TestTBBSpecificIssuesFcd}tjd|g}||\}}d}||||jr"t d|t d|dSdS)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:)rrrrrr)r1r3rr+rmsg_heads rtest_fork_from_non_main_threadz4TestTBBSpecificIssues.test_fork_from_non_main_threads64 l>4/<<((SN h$$$ ;  &#    &#       rc|d}dd|}tjd|g}tj}d|d<|||\}}||vr|dn| d ||j r"td |td |dSdS) 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_compilerreplacerrrrrrrrrr)r1rr3rrr+rs r&test_lifetime_of_task_scheduler_handlez4/joo', #$<<S<11S s " " MMO P P P P MM)S ) ) ) ;  &#    &#       rN)r4r5r6rr rrr&rrrzrzsXFXXZXt__Z___rrzc4eZdZdZdZedZdZdS)TestInitSafetyIssuesFc tj|tjtj}tjt |j} ||\}}|j dkr-td|j d| d | n#| wxYw| | fS)N)rrrrrrvr)r1rrrr+rs rrzTestInitSafetyIssues.run_cmds (2(2:::/-<<  MMOOO((**HC1$$$n%%%szz||||5666% NN    GNN    zz||SZZ\\))s A#CCcTtjtjtd}t j|g}||\}}|d||j r"td|td|dSdS)Nzorphaned_semaphore_usecase.pyzleaked semaphorer|r}) rr'rzr(r)rrr assertNotInrr)r1 test_filerr+rs rtest_orphaned_semaphorez,TestInitSafetyIssues.test_orphaned_semaphores GLL!:!:!@BB >9-<<((S +S111 ;  &#    &#       rcNdD]} tj|n#t$rY$wxYwd}tjd||g}||\}}|jr td|td|| ||dS)N)rrrzimport numba; import multiprocessing;multiprocessing.set_start_method('{}');print(multiprocessing.get_context().get_start_method())r0r|r}) rrrrrformatrrrr)r1methr>rr+rs rtest_lazy_lock_initz(TestInitSafetyIssues.test_lazy_lock_inits4 % %D +D1111    MC~tSZZ-=-=>G||G,,HC{ #fc"""fc""" MM$ $ $ $ $ % %s  ((N)r4r5r6rrr rrr&rrrrsTF***"Z %%%%%rrceZdZdZdS)TestOpenMPVendorsct}d|d<d|d<d|d<|D]G}tj|r&|||t jHdS)z> Checks the OpenMP vendor strings are correct MSwin32IntelrrrN)dictr rr!r" assertEqualr openmp_vendor)r1rAks r test_vendorszTestOpenMPVendors.test_vendorss66 $! E EA|&&q)) E  !g.CDDD E ErN)r4r5r6rr&rrrrs( E E E E Err__main__)brr]rBrrrSrrr;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&rrrsI  >>>>>>>>>>>>GGGGGGGGGGGGGGGG:::::: #% EEEEEE!!###&&&&&&NNNNN&&&&&&NNNNN%%%%LLLLL"h!.2NOO !h!.2KLL  ;W2e;)h)'3JKK < " "5 ) ) |x((    $$$$        v   22222222     &   22222H22222222x22222222222( ,,,,,v,,,omW];; !/??7#;#;<K$_oof&=&=>N*?OOL,I,IJ$O__Y%?%?@UEUEUEUEUEhUEUEUEp+/JJ')GHH+++++1++IH+B/////v///>0A0A0A0A0A*,C0A0A0Af3*3*3*3*3*H3*3*3*l"0"0"0"0"0"7"0"0"0J$$&&&C'C'C'C'C'!6C'C'C'LF'F'F'F'F'1F'F'F'Tddddd0dddN 1 D9%9%9%9%9%89%9%9%x EEEEEEE E$ zHMOOOOOs6!A::BBBBBB&&B0/B0