Repository: dask/dask
File: dask/array/reductions.py (Python)

Code before:
                'cumreduction with "blelloch" method required `preop=` argument'
            )
        return prefixscan_blelloch(func, preop, binop, x, axis, dtype, out=out)
    elif method != "sequential":
        raise ValueError(
            f'Invalid method for cumreduction.  Expected "sequential" or "blelloch".  Got: {method!r}'
        )

    if axis is None:
        if x.ndim > 1:
            x = x.flatten().rechunk(chunks=x.npartitions)
        axis = 0
    if dtype is None:
        dtype = getattr(func(np.ones((0,), dtype=x.dtype)), "dtype", object)
    assert isinstance(axis, Integral)
    axis = validate_axis(axis, x.ndim)

    use_dtype = False
    try:
        func_params = inspect.signature(func).parameters
        use_dtype = "dtype" in func_params
    except ValueError:
        try:
            # np.ufunc.accumulate doesn't have a signature, but it does accept dtype
            if isinstance(func.__self__, np.ufunc) and func.__name__ == "accumulate":
                use_dtype = True
        except AttributeError:
            pass

    if use_dtype:
        m = x.map_blocks(partial(func, dtype=dtype), axis=axis, dtype=dtype)
    else:
        m = x.map_blocks(func, axis=axis, dtype=dtype)

    name = f"{func.__name__}-{tokenize(func, axis, binop, ident, x, dtype)}"
    n = x.numblocks[axis]
    full = slice(None, None, None)
    slc = (full,) * axis + (slice(-1, None),) + (full,) * (x.ndim - axis - 1)

    indices = list(
        product(*[range(nb) if i != axis else [0] for i, nb in enumerate(x.numblocks)])
    )
    dsk = dict()
    for ind in indices:
        shape = tuple(x.chunks[i][ii] if i != axis else 1 for i, ii in enumerate(ind))
        dsk[(name, "extra") + ind] = (
            apply,
            np.full_like,
            (x._meta, ident, m.dtype),
            {"shape": shape},
        )

Review comment (suggestion):
We should probably handle all the ufuncs, not just `accumulate`

Code after:
                'cumreduction with "blelloch" method required `preop=` argument'
            )
        return prefixscan_blelloch(func, preop, binop, x, axis, dtype, out=out)
    elif method != "sequential":
        raise ValueError(
            f'Invalid method for cumreduction.  Expected "sequential" or "blelloch".  Got: {method!r}'
        )

    if axis is None:
        if x.ndim > 1:
            x = x.flatten().rechunk(chunks=x.npartitions)
        axis = 0
    if dtype is None:
        dtype = getattr(func(np.ones((0,), dtype=x.dtype)), "dtype", object)
    assert isinstance(axis, Integral)
    axis = validate_axis(axis, x.ndim)

    use_dtype = False
    try:
        func_params = inspect.signature(func).parameters
        use_dtype = "dtype" in func_params
    except ValueError:
        try:
            # Workaround for https://github.com/numpy/numpy/issues/30095
            # np.ufunc.accumulate doesn't have a signature, but it does accept dtype
            if isinstance(func.__self__, np.ufunc) and func.__name__ == "accumulate":
                use_dtype = True
        except AttributeError:
            pass

    if use_dtype:
        m = x.map_blocks(partial(func, dtype=dtype), axis=axis, dtype=dtype)
    else:
        m = x.map_blocks(func, axis=axis, dtype=dtype)

    name = f"{func.__name__}-{tokenize(func, axis, binop, ident, x, dtype)}"
    n = x.numblocks[axis]
    full = slice(None, None, None)
    slc = (full,) * axis + (slice(-1, None),) + (full,) * (x.ndim - axis - 1)

    indices = list(
        product(*[range(nb) if i != axis else [0] for i, nb in enumerate(x.numblocks)])
    )
    dsk = dict()
    for ind in indices:
        shape = tuple(x.chunks[i][ii] if i != axis else 1 for i, ii in enumerate(ind))
        dsk[(name, "extra") + ind] = (
            apply,
            np.full_like,
            (x._meta, ident, m.dtype),
            {"shape": shape},