'The `prange` tester
This is a hack. It basically switches out range calls for prange.
It does this by copying the live code object of a function
containing \'range\' then copying the .co_names and mutating it so
that \'range\' is replaced with \'prange\'. It then creates a new code
object containing the mutation and instantiates a function to contain
it. At this point three results are created:
1. The result of calling the original python function.
2. The result of calling a njit compiled version of the original
python function.
3. The result of calling a njit(parallel=True) version of the mutated
function containing `prange`.
The three results are then compared and the `prange` based function\'s
llvm_ir is inspected to ensure the scheduler code is present.
Arguments:
pyfunc - the python function to test
args - data arguments to pass to the pyfunc under test
Keyword Arguments:
patch_instance - iterable containing which instances of `range` to
replace. If not present all instance of `range` are
replaced.
Remaining kwargs are passed to np.testing.assert_almost_equal
Example:
def foo():
acc = 0
for x in range(5):
for y in range(10):
acc +=1
return acc
# calling as
prange_tester(foo)
# will test code equivalent to
# def foo():
#     acc = 0
#     for x in prange(5): # <- changed
#         for y in prange(10): # <- changed
#             acc +=1
#     return acc
# calling as
prange_tester(foo, patch_instance=[1])
# will test code equivalent to
# def foo():
#     acc = 0
#     for x in range(5): # <- outer loop (0) unchanged
#         for y in prange(10): # <- inner loop (1) changed
#             acc +=1
#     return acc'

def prange_tester(self, pyfunc, *args, **kwargs):

    pyfunc_code = pyfunc.__code__
    prange_names = list(pyfunc_code.co_names)
    patch_instance = kwargs.pop('patch_instance', None)
    if (not patch_instance):
        assert ('range' in pyfunc_code.co_names)
        prange_names = tuple([(x if (x != 'range') else 'prange') for x in pyfunc_code.co_names])
        new_code = bytes(pyfunc_code.co_code)
    else:
        range_idx = pyfunc_code.co_names.index('range')
        range_locations = []
        for (_, instr) in ByteCodeIter(pyfunc_code):
            if (instr.opname == 'LOAD_GLOBAL'):
                if (instr.arg == range_idx):
                    range_locations.append((instr.offset + 1))
        prange_names.append('prange')
        prange_names = tuple(prange_names)
        prange_idx = (len(prange_names) - 1)
        new_code = bytearray(pyfunc_code.co_code)
        assert (len(patch_instance) <= len(range_locations))
        for i in patch_instance:
            idx = range_locations[i]
            new_code[idx] = prange_idx
        new_code = bytes(new_code)
    co_args = [pyfunc_code.co_argcount]
    if (sys.version_info > (3, 0)):
        co_args.append(pyfunc_code.co_kwonlyargcount)
    co_args.extend([pyfunc_code.co_nlocals, pyfunc_code.co_stacksize, pyfunc_code.co_flags, new_code, pyfunc_code.co_consts, prange_names, pyfunc_code.co_varnames, pyfunc_code.co_filename, pyfunc_code.co_name, pyfunc_code.co_firstlineno, pyfunc_code.co_lnotab, pyfunc_code.co_freevars, pyfunc_code.co_cellvars])
    prange_code = pytypes.CodeType(*co_args)
    pfunc = pytypes.FunctionType(prange_code, globals())
    sig = tuple([numba.typeof(x) for x in args])
    cfunc = self.compile_njit(pyfunc, sig)
    cpfunc = self.compile_parallel(pfunc, sig)
    self.check_prange_vs_others(pyfunc, cfunc, cpfunc, *args, **kwargs)
