Unexpected result given by custom operator for buffer object

[Yiyang Li]

This issue is reproduced from the question on stack overflow.

When I tried to implement a customer operator that returns a numpy buffer with the smallest second item, it appears reduce gives the correct result, but Reduce does not.

Here is the working code with reduce

import numpy as np
from mpi4py import MPI

def find_min(a1:np.ndarray, a2:np.ndarray, datatype:MPI.Datatype) -> np.ndarray:
    """ Get the one with the smaller 2nd term."""
    print(a1, a2)
    if a1[1] > a2[1]:
        a1 = a2
    return a1

comm = MPI.COMM_WORLD
rank = comm.rank
RS = np.random.RandomState(rank)
a  = np.array([rank, RS.random()], dtype=np.float64)
print(rank, a)

FIND_MIN = MPI.Op.Create(find_min, commute=True)
amin = comm.reduce(a, op=FIND_MIN, root=0)
if rank == 0:
    print(amin)

and its output

0 [0.        0.5488135]
1 [1.        0.417022]
2 [2.        0.4359949]
3 [3.        0.5507979]

# 4 comparisons are conducted
[0.        0.5488135] [1.        0.417022]
[1.        0.417022]  [2.        0.4359949]
[2.        0.4359949] [3.        0.5507979]
[1.        0.417022]  [3.        0.5507979]

[1.       0.417022]  # <- Correct result

The testing code with Reduce is as follows

from typing import Tuple
import numpy as np
from mpi4py import MPI

def find_min(a1:MPI.memory, a2:MPI.memory, datatype:MPI.Datatype) -> MPI.memory:
    """ Get the one with the smaller 2nd term. """
    dim = (2,)          # Hard-coded
    dtype = np.float64  # Hard-coded

    a1p = to_ndarray(a1, dtype, dim)
    a2p = to_ndarray(a2, dtype, dim)
    print(a1p, a2p)

    if a1p[1] < a2p[1]:
        return a1
    return a2

def to_ndarray(a:MPI.memory, dtype:type, dim:Tuple[int,...]) -> np.ndarray:
    """ Convert a mpi4py.MPI.memory object to a numpy ndarray. """
    buf = np.array(a, dtype='B', copy=True)
    return np.ndarray(buffer=buf, dtype=dtype, shape=dim)

comm = MPI.COMM_WORLD
rank = comm.rank
RS = np.random.RandomState(rank)
a  = np.array([rank, RS.random()], dtype=np.float64)  # dim is (2,)
print(rank, a)

FIND_MIN = MPI.Op.Create(find_min, commute=True)
amin = np.zeros(2)
comm.Reduce(a, amin, op=FIND_MIN, root=0)
if rank == 0:
    print(amin) 

and it gives incorrect output as

0  [0.       0.5488135]
1  [1.       0.417022]
2  [2.       0.4359949]
3  [3.       0.5507979]

# 3 comparisons are conducted
[0.        0.5488135] [3.        0.5507979]
[1.        0.417022]  [3.        0.5507979]
[2.        0.4359949] [3.        0.5507979]

[3.        0.5507979]  # <- Wrong! Should be [1.       0.417022]

The comparisons made by the code with Reduce are incomplete. My tests show that either returning a MPI.memory object or a numpy.ndarray does not improve the result. The codes are tested with openmpi-4.0.3 and mpi4py-3.0.3.

[Lisandro Dalcin]

For the Reduce() case, yo are doing things wrong. You should emulated what you would do in C or Fortran, that is, the second argument is an input-output argument. Look at mpi4py’s sources, mysum_buf()in test/test_op.py. You have to write your reduction function as this:

def find_min(a1, a2, datatype.Datatype):
    dim = (2,)          # Hard-coded
    dtype = np.float64  # Hard-coded
    a1p = to_ndarray(a1, dtype, dim)
    a2p = to_ndarray(a2, dtype, dim)
    a2p[0] = a2p[0] # what do you want here??
    a2p[1] = min(a1p[1], a2p[1]) # a2p is both input and output.
    # note there is no return statement.
    # anything you return will be ignored.

Note that I added an explicit line with a2p[0] =….,.

However, I’m not really sure what you are tring to do. Note that reduce() works on “scalar“ (single item) Python values, while Reduce() operates on possibly many entrires of an array (such that you can reduce many values in a single call).

Next time, write these kind of questions to our mailing list. I prefer to use this issue tracker for actual bugs/issues, and not for general questions about how to code MPI programs.

[Yiyang Li]

Thank you Lisandro. Apologize for posting it here.

There is no more confusion now. reduce requires explicit return, so either the 1st or the 2nd argument can be modified and returned in find_min. Reduce works differently in its belly.

The actual code uses the first item as an index of some finite element, and the second item as some metric associated to that element. The purpose of the actual code is to filter a set of elements and return the one with the best metric.

[Yiyang Li]

If you’d like to, please just delete this “issue“ from tracker since it is a false one. I am ok with that. Thank you for your kind answer.

[Lisandro Dalcin]

Oh, so you are just reimplementing MPI’s builtin MPI_MINLOCoperation? Maybe you can use it directly (as long as your indices are 32bit integers). Note that you can create a user-defined MPI datatype enconding (int32, float64) type, and a corresponding NumPy datatype for creating buffers. Although more code is required, that’s the proper way to do it, this way you do not have to represent indices as floating point values.