Numba: functools.reduce doesn't work with numpy functions in parallel accelerator

[akaalharbi]

Reporting a bug

• [*] I have tried using the latest released version of Numba (most recent is
  visible in the change log (https://github.com/numba/numba/blob/master/CHANGE_LOG).
• [ *] I have included below a minimal working reproducer (if you are unsure how
  to write one see http://matthewrocklin.com/blog/work/2018/02/28/minimal-bug-reports).

I am trying to run the below code which use reduce from functools (numba websites indicates it supports reduce from functools)

import numpy as np
import functools as ft
from numba import jit, prange

@jit(nopython=True)
def list_gcd(L):
    return ft.reduce(np.gcd, L, 0)

A = np.array(list(range(0, 10, 2)))
list_gcd(A)

However, I get the error

No definition for lowering <built-in function reduce>(Function(<ufunc 'gcd'>), array(int64, 1d, C), Literal[int](0)) -> int64

Below is the full error

---------------------------------------------------------------------------
NotImplementedError                       Traceback (most recent call last)
~/opt/anaconda3/lib/python3.8/site-packages/numba/core/errors.py in new_error_context(fmt_, *args, **kwargs)
    744     try:
--> 745         yield
    746     except NumbaError as e:

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/lowering.py in lower_block(self, block)
    272                                    loc=self.loc, errcls_=defaulterrcls):
--> 273                 self.lower_inst(inst)
    274         self.post_block(block)

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/lowering.py in lower_inst(self, inst)
    369             ty = self.typeof(inst.target.name)
--> 370             val = self.lower_assign(ty, inst)
    371             self.storevar(val, inst.target.name)

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/lowering.py in lower_assign(self, ty, inst)
    543         elif isinstance(value, ir.Expr):
--> 544             return self.lower_expr(ty, value)
    545 

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/lowering.py in lower_expr(self, resty, expr)
   1069         elif expr.op == 'call':
-> 1070             res = self.lower_call(resty, expr)
   1071             return res

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/lowering.py in lower_call(self, resty, expr)
    805         else:
--> 806             res = self._lower_call_normal(fnty, expr, signature)
    807 

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/lowering.py in _lower_call_normal(self, fnty, expr, signature)
   1032             )
-> 1033         impl = self.context.get_function(fnty, signature)
   1034         if signature.recvr:

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/base.py in get_function(self, fn, sig, _firstcall)
    569             self.refresh()
--> 570             return self.get_function(fn, sig, _firstcall=False)
    571 

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/base.py in get_function(self, fn, sig, _firstcall)
    571 
--> 572         raise NotImplementedError("No definition for lowering %s%s" % (key, sig))
    573 

NotImplementedError: No definition for lowering <built-in function reduce>(Function(<ufunc 'gcd'>), array(int64, 1d, C), Literal[int](0)) -> int64

During handling of the above exception, another exception occurred:

LoweringError                             Traceback (most recent call last)
<ipython-input-38-9cbfe228bf0e> in <module>
     23 
     24 A = np.array(list(range(0, 10, 2)))
---> 25 list_gcd(A)
     26 
     27 ft.reduce(np.gcd, A, 0)

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in _compile_for_args(self, *args, **kws)
    432                     e.patch_message('\n'.join((str(e).rstrip(), help_msg)))
    433             # ignore the FULL_TRACEBACKS config, this needs reporting!
--> 434             raise e
    435 
    436     def inspect_llvm(self, signature=None):

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in _compile_for_args(self, *args, **kws)
    365                 argtypes.append(self.typeof_pyval(a))
    366         try:
--> 367             return self.compile(tuple(argtypes))
    368         except errors.ForceLiteralArg as e:
    369             # Received request for compiler re-entry with the list of arguments

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_lock.py in _acquire_compile_lock(*args, **kwargs)
     30         def _acquire_compile_lock(*args, **kwargs):
     31             with self:
---> 32                 return func(*args, **kwargs)
     33         return _acquire_compile_lock
     34 

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in compile(self, sig)
    806             self._cache_misses[sig] += 1
    807             try:
--> 808                 cres = self._compiler.compile(args, return_type)
    809             except errors.ForceLiteralArg as e:
    810                 def folded(args, kws):

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in compile(self, args, return_type)
     76 
     77     def compile(self, args, return_type):
---> 78         status, retval = self._compile_cached(args, return_type)
     79         if status:
     80             return retval

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in _compile_cached(self, args, return_type)
     90 
     91         try:
---> 92             retval = self._compile_core(args, return_type)
     93         except errors.TypingError as e:
     94             self._failed_cache[key] = e

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in _compile_core(self, args, return_type)
    103 
    104         impl = self._get_implementation(args, {})
--> 105         cres = compiler.compile_extra(self.targetdescr.typing_context,
    106                                       self.targetdescr.target_context,
    107                                       impl,

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler.py in compile_extra(typingctx, targetctx, func, args, return_type, flags, locals, library, pipeline_class)
    601     pipeline = pipeline_class(typingctx, targetctx, library,
    602                               args, return_type, flags, locals)
--> 603     return pipeline.compile_extra(func)
    604 
    605 

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler.py in compile_extra(self, func)
    337         self.state.lifted = ()
    338         self.state.lifted_from = None
--> 339         return self._compile_bytecode()
    340 
    341     def compile_ir(self, func_ir, lifted=(), lifted_from=None):

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler.py in _compile_bytecode(self)
    399         """
    400         assert self.state.func_ir is None
--> 401         return self._compile_core()
    402 
    403     def _compile_ir(self):

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler.py in _compile_core(self)
    379                 self.state.status.fail_reason = e
    380                 if is_final_pipeline:
--> 381                     raise e
    382         else:
    383             raise CompilerError("All available pipelines exhausted")

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler.py in _compile_core(self)
    370             res = None
    371             try:
--> 372                 pm.run(self.state)
    373                 if self.state.cr is not None:
    374                     break

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_machinery.py in run(self, state)
    339                     (self.pipeline_name, pass_desc)
    340                 patched_exception = self._patch_error(msg, e)
--> 341                 raise patched_exception
    342 
    343     def dependency_analysis(self):

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_machinery.py in run(self, state)
    330                 pass_inst = _pass_registry.get(pss).pass_inst
    331                 if isinstance(pass_inst, CompilerPass):
--> 332                     self._runPass(idx, pass_inst, state)
    333                 else:
    334                     raise BaseException("Legacy pass in use")

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_lock.py in _acquire_compile_lock(*args, **kwargs)
     30         def _acquire_compile_lock(*args, **kwargs):
     31             with self:
---> 32                 return func(*args, **kwargs)
     33         return _acquire_compile_lock
     34 

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_machinery.py in _runPass(self, index, pss, internal_state)
    289             mutated |= check(pss.run_initialization, internal_state)
    290         with SimpleTimer() as pass_time:
--> 291             mutated |= check(pss.run_pass, internal_state)
    292         with SimpleTimer() as finalize_time:
    293             mutated |= check(pss.run_finalizer, internal_state)

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_machinery.py in check(func, compiler_state)
    262 
    263         def check(func, compiler_state):
--> 264             mangled = func(compiler_state)
    265             if mangled not in (True, False):
    266                 msg = ("CompilerPass implementations should return True/False. "

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/typed_passes.py in run_pass(self, state)
    440 
    441         # TODO: Pull this out into the pipeline
--> 442         NativeLowering().run_pass(state)
    443         lowered = state['cr']
    444         signature = typing.signature(state.return_type, *state.args)

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/typed_passes.py in run_pass(self, state)
    368                 lower = lowering.Lower(targetctx, library, fndesc, interp,
    369                                        metadata=metadata)
--> 370                 lower.lower()
    371                 if not flags.no_cpython_wrapper:
    372                     lower.create_cpython_wrapper(flags.release_gil)

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/lowering.py in lower(self)
    177         if self.generator_info is None:
    178             self.genlower = None
--> 179             self.lower_normal_function(self.fndesc)
    180         else:
    181             self.genlower = self.GeneratorLower(self)

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/lowering.py in lower_normal_function(self, fndesc)
    231         # Init argument values
    232         self.extract_function_arguments()
--> 233         entry_block_tail = self.lower_function_body()
    234 
    235         # Close tail of entry block

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/lowering.py in lower_function_body(self)
    257             bb = self.blkmap[offset]
    258             self.builder.position_at_end(bb)
--> 259             self.lower_block(block)
    260         self.post_lower()
    261         return entry_block_tail

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/lowering.py in lower_block(self, block)
    271             with new_error_context('lowering "{inst}" at {loc}', inst=inst,
    272                                    loc=self.loc, errcls_=defaulterrcls):
--> 273                 self.lower_inst(inst)
    274         self.post_block(block)
    275 

~/opt/anaconda3/lib/python3.8/contextlib.py in __exit__(self, type, value, traceback)
    129                 value = type()
    130             try:
--> 131                 self.gen.throw(type, value, traceback)
    132             except StopIteration as exc:
    133                 # Suppress StopIteration *unless* it's the same exception that

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/errors.py in new_error_context(fmt_, *args, **kwargs)
    750         newerr = errcls(e).add_context(_format_msg(fmt_, args, kwargs))
    751         tb = sys.exc_info()[2] if numba.core.config.FULL_TRACEBACKS else None
--> 752         reraise(type(newerr), newerr, tb)
    753 
    754 

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/utils.py in reraise(tp, value, tb)
     79     if value.__traceback__ is not tb:
     80         raise value.with_traceback(tb)
---> 81     raise value
     82 
     83 

LoweringError: Failed in nopython mode pipeline (step: nopython mode backend)
No definition for lowering <built-in function reduce>(Function(<ufunc 'gcd'>), array(int64, 1d, C), Literal[int](0)) -> int64

File "<ipython-input-38-9cbfe228bf0e>", line 17:
def list_gcd(L):
    return ft.reduce(np.gcd, L, 0)
    ^

During: lowering "$14call_method.6 = call $4load_method.1($8load_attr.3, L, $const12.5, func=$4load_method.1, args=[Var($8load_attr.3, <ipython-input-38-9cbfe228bf0e>:17), Var(L, <ipython-input-38-9cbfe228bf0e>:17), Var($const12.5, <ipython-input-38-9cbfe228bf0e>:17)], kws=(), vararg=None)" at <ipython-input-38-9cbfe228bf0e> (17)

[stuartarchibald]

Thanks for the report. The spelling for functools use is a bit contrived.

1. The function call must be exactly the function reduce.
2. The function over which reduce operates must be jitted (there's another ticket about this somewhere, I'll try and find it.).

This sort of thing should work:

import numpy as np
from functools import reduce
from numba import prange, njit

@njit
def npgcd(x, y):
    return np.gcd(x, y)

@njit
def list_gcd(L):
    return reduce(npgcd, L, 0)

A = np.array(list(range(0, 10, 2)))
list_gcd(A)

Am going to mark this as a bug because the spelling of reduce shouldn't matter.

[akaalharbi]

Thank you very much for the swift and detailed response.

The you've posted works without errors, however, when I try to make it parallel it fails

import numpy as np
from functools import reduce
from numba import prange, njit

@njit
def npgcd(x, y):
    return np.gcd(x, y)

@njit(parallel=True)
def list_gcd(L):
    return reduce(npgcd, L, 0)

A = np.array(list(range(0, 10, 2)))
list_gcd(A)

Full error:

ValueError Traceback (most recent call last)
in
12
13 A = np.array(list(range(0, 10, 2)))
---> 14 list_gcd(A)

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in _compile_for_args(self, args, *kws)
432 e.patch_message('\n'.join((str(e).rstrip(), help_msg)))
433 # ignore the FULL_TRACEBACKS config, this needs reporting!
--> 434 raise e
435
436 def inspect_llvm(self, signature=None):

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in _compile_for_args(self, args, *kws)
365 argtypes.append(self.typeof_pyval(a))
366 try:
--> 367 return self.compile(tuple(argtypes))
368 except errors.ForceLiteralArg as e:
369 # Received request for compiler re-entry with the list of arguments

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_lock.py in _acquire_compile_lock(args, *kwargs)
30 def _acquire_compile_lock(args, *kwargs):
31 with self:
---> 32 return func(args, *kwargs)
33 return _acquire_compile_lock
34

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in compile(self, sig)
806 self._cache_misses[sig] += 1
807 try:
--> 808 cres = self._compiler.compile(args, return_type)
809 except errors.ForceLiteralArg as e:
810 def folded(args, kws):

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in compile(self, args, return_type)
76
77 def compile(self, args, return_type):
---> 78 status, retval = self._compile_cached(args, return_type)
79 if status:
80 return retval

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in _compile_cached(self, args, return_type)
90
91 try:
---> 92 retval = self._compile_core(args, return_type)
93 except errors.TypingError as e:
94 self._failed_cache[key] = e

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/dispatcher.py in _compile_core(self, args, return_type)
103
104 impl = self._get_implementation(args, {})
--> 105 cres = compiler.compile_extra(self.targetdescr.typing_context,
106 self.targetdescr.target_context,
107 impl,

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler.py in compile_extra(typingctx, targetctx, func, args, return_type, flags, locals, library, pipeline_class)
601 pipeline = pipeline_class(typingctx, targetctx, library,
602 args, return_type, flags, locals)
--> 603 return pipeline.compile_extra(func)
604
605

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler.py in compile_extra(self, func)
337 self.state.lifted = ()
338 self.state.lifted_from = None
--> 339 return self._compile_bytecode()
340
341 def compile_ir(self, func_ir, lifted=(), lifted_from=None):

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler.py in _compile_bytecode(self)
399 """
400 assert self.state.func_ir is None
--> 401 return self._compile_core()
402
403 def _compile_ir(self):

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler.py in _compile_core(self)
379 self.state.status.fail_reason = e
380 if is_final_pipeline:
--> 381 raise e
382 else:
383 raise CompilerError("All available pipelines exhausted")

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler.py in _compile_core(self)
370 res = None
371 try:
--> 372 pm.run(self.state)
373 if self.state.cr is not None:
374 break

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_machinery.py in run(self, state)
339 (self.pipeline_name, pass_desc)
340 patched_exception = self._patch_error(msg, e)
--> 341 raise patched_exception
342
343 def dependency_analysis(self):

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_machinery.py in run(self, state)
330 pass_inst = _pass_registry.get(pss).pass_inst
331 if isinstance(pass_inst, CompilerPass):
--> 332 self._runPass(idx, pass_inst, state)
333 else:
334 raise BaseException("Legacy pass in use")

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_lock.py in _acquire_compile_lock(args, *kwargs)
30 def _acquire_compile_lock(args, *kwargs):
31 with self:
---> 32 return func(args, *kwargs)
33 return _acquire_compile_lock
34

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_machinery.py in _runPass(self, index, pss, internal_state)
289 mutated |= check(pss.run_initialization, internal_state)
290 with SimpleTimer() as pass_time:
--> 291 mutated |= check(pss.run_pass, internal_state)
292 with SimpleTimer() as finalize_time:
293 mutated |= check(pss.run_finalizer, internal_state)

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/compiler_machinery.py in check(func, compiler_state)
262
263 def check(func, compiler_state):
--> 264 mangled = func(compiler_state)
265 if mangled not in (True, False):
266 msg = ("CompilerPass implementations should return True/False. "

~/opt/anaconda3/lib/python3.8/site-packages/numba/core/typed_passes.py in run_pass(self, state)
286 state.flags,
287 state.parfor_diagnostics)
--> 288 parfor_pass.run()
289
290 remove_dels(state.func_ir.blocks)

~/opt/anaconda3/lib/python3.8/site-packages/numba/parfors/parfor.py in run(self)
2692 # Validate reduction in parfors.
2693 for p in parfors:
-> 2694 get_parfor_reductions(self.func_ir, p, p.params, self.calltypes)
2695
2696 # Validate parameters:

~/opt/anaconda3/lib/python3.8/site-packages/numba/parfors/parfor.py in get_parfor_reductions(func_ir, parfor, parfor_params, calltypes, reductions, reduce_varnames, param_uses, param_nodes, var_to_param)
3315 if param_name in used_vars and param_name not in reduce_varnames:
3316 param_nodes[param].reverse()
-> 3317 reduce_nodes = get_reduce_nodes(param, param_nodes[param], func_ir)
3318 # Certain kinds of ill-formed Python (like potentially undefined
3319 # variables) in combination with SSA can make things look like

~/opt/anaconda3/lib/python3.8/site-packages/numba/parfors/parfor.py in get_reduce_nodes(reduction_node, nodes, func_ir)
3418
3419 if not supported_reduction(rhs, func_ir):
-> 3420 raise ValueError(("Use of reduction variable " + unversioned_name +
3421 " in an unsupported reduction function."))
3422 args = [ (x.name, lookup(x, True)) for x in get_expr_args(rhs) ]

ValueError: Failed in nopython mode pipeline (step: convert to parfors)
Use of reduction variable $10call_function.4 in an unsupported reduction function.

[stuartarchibald]

@akaalharbi I think the parallel=True code is hitting a guard to make sure the reduction is a supported one (as noted here https://numba.readthedocs.io/en/stable/user/parallel.html#supported-operations), but I'm not sure it should apply in the case of reduce. CC @DrTodd13