Wrong result on non-blocking collectives with user-defined op

[yukiM-fj]

Our team(Fujitsu MPI team) found a problem in libnbc at Open MPI 2.0.0rc1.
Results may be wrong when using non-blocking collectives with user-defined op.
The problem occurrs because of the following two reasons:

• Non-commutative operations are not considered in libnbc.
• Results may be overwritten unexpectedly in ompi_3buff_op_user() function.
  • Overwritten when src2 and dst are pointed same buffer in ompi_3buff_op_user()'s arguments.
  • "Chain" algorithm at MPI_Ireduce is this pattern.

I wrote a program to reproduce this problem at gist.

We fixed libnbc by following ways:

• Changed algorithm selection and algorithm behavior with non-commutative op.
  • when non-commutative op is set, not call iallreduce ring alg and rank 0 is root temporarily at ireduce binomial alg.
• Added a wrapper when call "NBC_Sched_op" in each algorithm and set parameters in this wrapper to compute correctly.

I wrote two files to show pseudo-code to fix it at another gist.
These files are based on Open MPI v1.8.4. In my gist, "psedo-alg-selection.c" is for the algorithm selection and behavior and "pseudo-wrapper_libnbcop.c" is for the wrapper.

[ggouaillardet]

with two ranks, this looks like a bug in Open MPI and here is a patch

diff --git a/ompi/mca/coll/libnbc/nbc_iallreduce.c b/ompi/mca/coll/libnbc/nbc_iallreduce.c
index 2e1b0dd..c47c988 100644
--- a/ompi/mca/coll/libnbc/nbc_iallreduce.c
+++ b/ompi/mca/coll/libnbc/nbc_iallreduce.c
@@ -319,11 +319,11 @@ static inline int allred_sched_diss(int rank, int p, int count, MPI_Datatype dat
         /* this cannot be done until handle->tmpbuf is unused :-( so barrier after the op */
         if (firstred && MPI_IN_PLACE != sendbuf) {
           /* perform the reduce with the senbuf */
-          res = NBC_Sched_op (recvbuf, false, sendbuf, false, 0, true, count, datatype, op, schedule, true);
+          res = NBC_Sched_op (recvbuf, false, 0, true, sendbuf, false, count, datatype, op, schedule, true);
           firstred = 0;
         } else {
           /* perform the reduce in my local buffer */
-          res = NBC_Sched_op (recvbuf, false, recvbuf, false, 0, true, count, datatype, op, schedule, true);
+          res = NBC_Sched_op (recvbuf, false, 0, true, recvbuf, false, datatype, op, schedule, true);
         }

         if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) {

i will double check what happens with 3 tasks and more.

if i understand the standard correctly, a user defined reduction must be associative but might not be commutative, am i right ?

[ggouaillardet]

@bosilca

static inline void ompi_3buff_op_user (ompi_op_t *op, void * restrict source1, void * restrict source2,
                                       void * restrict result, int count, struct ompi_datatype_t *dtype)
{
    ompi_datatype_copy_content_same_ddt (dtype, count, result, source1);
    op->o_func.c_fn (source2, result, &count, &dtype);
}

basically, it does
result = op(source2, source1)
which is counter-intuitive (at least to me ...), is this intentional ?

also, it does not work if result == source2

that being said, libnbc could/should be rewritten to use ompi_op_reduce, I will work on that tomorrow

[ggouaillardet]

@yukiM-fj i made PR #1760 for master

the inline patch is for v2.x

diff --git a/ompi/mca/coll/libnbc/nbc.c b/ompi/mca/coll/libnbc/nbc.c
index 75c37c4..aee5418 100644
--- a/ompi/mca/coll/libnbc/nbc.c
+++ b/ompi/mca/coll/libnbc/nbc.c
@@ -10,7 +10,7 @@
  *                         rights reserved.
  * Copyright (c) 2015      Los Alamos National Security, LLC.  All rights
  *                         reserved.
- * Copyright (c) 2015      Research Organization for Information Science
+ * Copyright (c) 2015-2016 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  *
  * Author(s): Torsten Hoefler <htor@cs.indiana.edu>
@@ -159,6 +159,32 @@ int NBC_Sched_recv (void* buf, char tmpbuf, int count, MPI_Datatype datatype, in
 }

 /* this function puts an operation into the schedule */
+int NBC_Sched_op2 (const void* buf1, char tmpbuf1, void* buf2, char tmpbuf2, int count, MPI_Datatype datatype, MPI_Op op, NBC_Schedule *schedule, bool barrier) {
+  NBC_Args_op op_args;
+  int ret;
+
+  /* store the passed arguments */
+  op_args.type = OP2;
+  op_args.buf1 = buf1;
+  op_args.buf2 = buf2;
+  op_args.tmpbuf1 = tmpbuf1;
+  op_args.tmpbuf2 = tmpbuf2;
+  op_args.count = count;
+  op_args.op = op;
+  op_args.datatype = datatype;
+
+  /* append to the round-schedule */
+  ret = nbc_schedule_round_append (schedule, &op_args, sizeof (op_args), barrier);
+  if (OMPI_SUCCESS != ret) {
+    return ret;
+  }
+
+  NBC_DEBUG(10, "added op2 - ends at byte %i\n", nbc_schedule_get_size (schedule));
+
+  return OMPI_SUCCESS;
+}
+
+/* this function puts an operation into the schedule */
 int NBC_Sched_op(void *buf3, char tmpbuf3, const void* buf1, char tmpbuf1, void* buf2, char tmpbuf2, int count, MPI_Datatype datatype, MPI_Op op, NBC_Schedule *schedule, bool barrier) {
   NBC_Args_op op_args;
   int ret;
@@ -477,6 +503,24 @@ static inline int NBC_Start_round(NBC_Handle *handle) {
         }
         ompi_3buff_op_reduce(opargs.op, buf1, buf2, buf3, opargs.count, opargs.datatype);
         break;
+      case OP2:
+        NBC_DEBUG(5, "  OP2  (offset %li) ", offset);
+        NBC_GET_BYTES(ptr,opargs);
+        NBC_DEBUG(5, "*buf1: %p, buf2: %p, count: %i, type: %p)\n", opargs.buf1, opargs.buf2,
+                  opargs.count, opargs.datatype);
+        /* get buffers */
+        if(opargs.tmpbuf1) {
+          buf1=(char*)handle->tmpbuf+(long)opargs.buf1;
+        } else {
+          buf1=(void *)opargs.buf1;
+        }
+        if(opargs.tmpbuf2) {
+          buf2=(char*)handle->tmpbuf+(long)opargs.buf2;
+        } else {
+          buf2=opargs.buf2;
+        }
+        ompi_op_reduce(opargs.op, buf1, buf2, opargs.count, opargs.datatype);
+        break;
       case COPY:
         NBC_DEBUG(5, "  COPY   (offset %li) ", offset);
         NBC_GET_BYTES(ptr,copyargs);
diff --git a/ompi/mca/coll/libnbc/nbc_iallreduce.c b/ompi/mca/coll/libnbc/nbc_iallreduce.c
index 2e1b0dd..ee99b1a 100644
--- a/ompi/mca/coll/libnbc/nbc_iallreduce.c
+++ b/ompi/mca/coll/libnbc/nbc_iallreduce.c
@@ -7,7 +7,7 @@
  *                         rights reserved.
  * Copyright (c) 2013-2015 Los Alamos National Security, LLC. All rights
  *                         reserved.
- * Copyright (c) 2014-2015 Research Organization for Information Science
+ * Copyright (c) 2014-2016 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  *
  * Author(s): Torsten Hoefler <htor@cs.indiana.edu>
@@ -16,6 +16,7 @@
 #include "nbc_internal.h"
 #include "ompi/communicator/communicator.h"
 #include "ompi/datatype/ompi_datatype.h"
+#include "ompi/op/op.h"

 #include <assert.h>

@@ -101,7 +102,7 @@ int ompi_coll_libnbc_iallreduce(const void* sendbuf, void* recvbuf, int count, M
   }

   /* algorithm selection */
-  if(p < 4 || size*count < 65536 || inplace) {
+  if(p < 4 || size*count < 65536 || !ompi_op_is_commute(op) || inplace) {
     alg = NBC_ARED_BINOMIAL;
   } else {
     alg = NBC_ARED_RING;
@@ -299,10 +300,24 @@ int ompi_coll_libnbc_iallreduce_inter(const void* sendbuf, void* recvbuf, int co
 static inline int allred_sched_diss(int rank, int p, int count, MPI_Datatype datatype, const void *sendbuf, void *recvbuf,
                                     MPI_Op op, NBC_Schedule *schedule, NBC_Handle *handle) {
   int root, vrank, maxr, vpeer, peer, res;
+  char *rbuf, *lbuf, *buf;
+  int tmprbuf, tmplbuf;

   root = 0; /* this makes the code for ireduce and iallreduce nearly identical - could be changed to improve performance */
   RANK2VRANK(rank, vrank, root);
   maxr = (int)ceil((log((double)p)/LOG2));
+  /* ensure the result ends up in recvbuf on vrank 0 */
+  if (0 == (maxr%2)) {
+    rbuf = 0;
+    tmprbuf = true;
+    lbuf = recvbuf;
+    tmplbuf = false;
+  } else {
+    lbuf = 0;
+    tmplbuf = true;
+    rbuf = recvbuf;
+    tmprbuf = false;
+  }

   for (int r = 1, firstred = 1 ; r <= maxr ; ++r) {
     if ((vrank % (1 << r)) == 0) {
@@ -311,7 +326,7 @@ static inline int allred_sched_diss(int rank, int p, int count, MPI_Datatype dat
       VRANK2RANK(peer, vpeer, root)
       if (peer < p) {
         /* we have to wait until we have the data */
-        res = NBC_Sched_recv (0, true, count, datatype, peer, schedule, true);
+        res = NBC_Sched_recv (rbuf, tmprbuf, count, datatype, peer, schedule, true);
         if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) {
           return res;
         }
@@ -319,16 +334,18 @@ static inline int allred_sched_diss(int rank, int p, int count, MPI_Datatype dat
         /* this cannot be done until handle->tmpbuf is unused :-( so barrier after the op */
         if (firstred && MPI_IN_PLACE != sendbuf) {
           /* perform the reduce with the senbuf */
-          res = NBC_Sched_op (recvbuf, false, sendbuf, false, 0, true, count, datatype, op, schedule, true);
+          res = NBC_Sched_op2 (sendbuf, false, rbuf, tmprbuf, count, datatype, op, schedule, true);
           firstred = 0;
         } else {
           /* perform the reduce in my local buffer */
-          res = NBC_Sched_op (recvbuf, false, recvbuf, false, 0, true, count, datatype, op, schedule, true);
+          res = NBC_Sched_op2 (lbuf, tmplbuf, rbuf, tmprbuf, count, datatype, op, schedule, true);
         }
-
         if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) {
           return res;
         }
+        /* swap left and right buffers */
+        buf = rbuf; rbuf = lbuf ; lbuf = buf;
+        tmprbuf ^= 1; tmplbuf ^= 1;
       }
     } else {
       /* we have to send this round */
@@ -338,8 +355,8 @@ static inline int allred_sched_diss(int rank, int p, int count, MPI_Datatype dat
         /* we have to use the sendbuf in the first round .. */
         res = NBC_Sched_send (sendbuf, false, count, datatype, peer, schedule, false);
       } else {
-        /* and the recvbuf in all remeining rounds */
-        res = NBC_Sched_send (recvbuf, false, count, datatype, peer, schedule, false);
+        /* and the recvbuf in all remaining rounds */
+        res = NBC_Sched_send (lbuf, tmplbuf, count, datatype, peer, schedule, false);
       }

       if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) {
@@ -373,6 +390,7 @@ static inline int allred_sched_diss(int rank, int p, int count, MPI_Datatype dat
     }
   }

+  if (0 == vrank) assert(lbuf == recvbuf);
   /* now send to the right hosts */
   for (int r = 0; r < maxr; ++r) {
     if (((vrank + (1 << r) < p) && (vrank < (1 << r))) || (vrank == 0)) {
diff --git a/ompi/mca/coll/libnbc/nbc_internal.h b/ompi/mca/coll/libnbc/nbc_internal.h
index fbf6833..c32dfb6 100644
--- a/ompi/mca/coll/libnbc/nbc_internal.h
+++ b/ompi/mca/coll/libnbc/nbc_internal.h
@@ -10,7 +10,7 @@
  *
  * Copyright (c) 2012      Oracle and/or its affiliates.  All rights reserved.
  * Copyright (c) 2014      NVIDIA Corporation.  All rights reserved.
- * Copyright (c) 2015      Research Organization for Information Science
+ * Copyright (c) 2015-2016 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * Copyright (c) 2015      Los Alamos National Security, LLC. All rights
  *                         reserved.
@@ -80,6 +80,7 @@ typedef enum {
   SEND,
   RECV,
   OP,
+  OP2,
   COPY,
   UNPACK
 } NBC_Fn_type;
@@ -147,6 +148,7 @@ int NBC_Sched_create(NBC_Schedule* schedule);
 int NBC_Sched_send(const void* buf, char tmpbuf, int count, MPI_Datatype datatype, int dest, NBC_Schedule *schedule, bool barrier);
 int NBC_Sched_recv(void* buf, char tmpbuf, int count, MPI_Datatype datatype, int source, NBC_Schedule *schedule, bool barrier);
 int NBC_Sched_op(void* buf3, char tmpbuf3, const void* buf1, char tmpbuf1, void* buf2, char tmpbuf2, int count, MPI_Datatype datatype, MPI_Op op, NBC_Schedule *schedule, bool barrier);
+int NBC_Sched_op2 (const void* buf1, char tmpbuf1, void* buf2, char tmpbuf2, int count, MPI_Datatype datatype,MPI_Op op, NBC_Schedule *schedule, bool barrier);
 int NBC_Sched_copy(void *src, char tmpsrc, int srccount, MPI_Datatype srctype, void *tgt, char tmptgt, int tgtcount, MPI_Datatype tgttype, NBC_Schedule *schedule, bool barrier);
 int NBC_Sched_unpack(void *inbuf, char tmpinbuf, int count, MPI_Datatype datatype, void *outbuf, char tmpoutbuf, NBC_Schedule *schedule, bool barrier);
 int NBC_Sched_barrier(NBC_Schedule *schedule);
@@ -314,6 +316,7 @@ static inline void nbc_get_round_size (char *p, unsigned long *size) {
       offset += sizeof(NBC_Args_recv);
       break;
     case OP:
+    case OP2:
       /*printf("found a OP at offset %li\n", (long)p-(long)schedule); */
       offset += sizeof(NBC_Args_op);            \
       break;
@@ -390,6 +393,7 @@ static inline void nbc_schedule_inc_round (NBC_Schedule *schedule) {
          printf("*buf: %lu, count: %i, type: %lu, source: %i)\n", (unsigned long)recvargs.buf, recvargs.count, (unsigned long)recvargs.datatype, recvargs.source); \
          break; \
        case OP: \
+       case OP2: \
          printf("[%i]  OP   (offset %li) ", myrank, (long)p-(long)schedule); \
          NBC_GET_BYTES(p,opargs); \
          printf("*buf1: %lu, buf2: %lu, count: %i, type: %lu)\n", (unsigned long)opargs.buf1, (unsigned long)opargs.buf2, opargs.count, (unsigned long)opargs.datatype); \

could you please give it a try ?

fwiw, if i replace MPI_Iallreduce by MPI_Ireduce followed by MPI_Bcast, wrong behavior is observed.

i will review this and other non blocking collectives that use MPI_Op

thanks for the report and test case, it was very helpful !

[yukiM-fj]

@ggouaillardet, thank you. I try building & testing your PR now.
I think so, too (a user defined reduction must be associative but may be non-commutative ) .

fwiw, I add other TPs at gist
These TPs are checking non-blocking collectives (iscan, iexscan, ireduce_scatter, and ireduce_scatter_block). But not check recently Open MPI, I checked only Fujitsu MPI.
I will write here test's result.

[ggouaillardet]

@yukiM-fj thanks !
do you mind me adding these test to the open-mpi/ompi-tests test suite ?

out of curiosity, what is the difference between Open MPI and Fujitsu MPI regarding non blocking collectives ?
does Fujitsu MPI use libnbc ? an other (proprietary ?) module ?

[yukiM-fj]

@ggouaillardet it is ok to adding these tests.
Fujitsu MPI also use libnbc. But Fujitsu MPI has already fixed this issue.
Some issues are fixed in Fujitsu MPI, but not fixed in Open MPI.
So, we're trying to PR or issue now...

[yukiM-fj]

Sorry! I missed operation.

[yukiM-fj]

@ggouaillardet, I tested your PR with these tests in my gist.
Then, iallreduce, iscan, iexscan look OK, but ireduce, ireduce_scatter and ireduce_scatter_block are abort.

[ggouaillardet]

@yukiM-fj i just pushed some more commits into #1760
all tests should pass now.
please note there is still an issue with how temporary buffers are allocated/used.
it is basically broken if lower_bound is non zero, or if extent < true_extent
i will also consider some code factorization, since the binomial reduce algo is duplicated in several places.

[yukiM-fj]

@ggouaillardet I look your PR, but may be tomorrow...

binomial ireduce alg needs more buffer when op is non-commutative.
Root may not be 0 , but communication and computation is like that root is 0.
And recvbuf may be not exist when rank is not root, so need other temporary buffer.
so i wrote gist when alg selection.
(Sorry, i cannot display here my idea, so i write my idea at gist)

[ggouaillardet]

@yukiM-fj i kind of simplified that
when root is not zero and op is not commutative, then the reduction is done on rank 0, and the result is sent to root.
regardless root, root rank has one temp buffer and recv buffer, and other ranks have two temp buffers, so i think this is fine ... did i miss something ?

[yukiM-fj]

@ggouaillardet , I look your PR. I think your modification is ok.
But I'm still confused by two following things.

1. we need to check non-collective reductions with inter-communicator?
2. I don't understand well "it is basically broken if lower_bound is non zero, or if extent < true_extent ".
   If I understand correctly about 2, result may be still wrong with your PR when use specified datatype & user-defined op?