Changeset 3871

Timestamp:

02/18/09 17:05:16 (less than one hour ago)

Author:

goodell

Message:

Add a new non-commutative reduce-scatter algorithm.

This change implements Jesper's butterfly algorithm for non-commutative
reduce-scatters. It currently only covers the power-of-2, block-regular case.
It also cleans up much of the existing error handling and memory allocation.

Reviewed by thakur@.

Files:

• mpich2/trunk/src/mpi/coll/red_scat.c (modified) (24 diffs)
• mpich2/trunk/test/mpi/coll/redscat2.c (modified) (1 diff)

mpich2/trunk/src/mpi/coll/red_scat.c

@@ -24 +24 @@
-#define MPI_Reduce_scatter PMPI_Reduce_scatter
-
+
+/* Implements the "mirror permutation" of "bits" bits of an integer "x".
+
+   positions 76543210, bits==3 yields 76543012.
+
+   This function could/should be moved to a common utility location for use in
+   other collectives as well. */
+ATTRIBUTE((const)) /* tells the compiler that this func only depends on its args
+                      and may be optimized much more aggressively, similar to "pure" */
+static inline int mirror_permutation(unsigned int x, int bits)
+{
+    /* a mask for the high order bits that should be copied as-is */
+    int high_mask = ~((0x1 << bits) - 1);
+    int retval = x & high_mask;
+    int i;
+
+    for (i = 0; i < bits; ++i) {
+        unsigned int bitval = (x & (0x1 << i)) >> i; /* 0x1 or 0x0 */
+        retval |= bitval << ((bits - i) - 1);
+    }
+
+    return retval;
+}
+
+/* FIXME should we be checking the op_errno here? */
+#ifdef HAVE_CXX_BINDING
+/* NOTE: assumes 'uop' is the operator function pointer and
+   that 'is_cxx_uop' is is a boolean indicating the obvious */
+#define call_uop(in_, inout_, count_, datatype_)                                     \
+do {                                                                                 \
+    if (is_cxx_uop) {                                                                \
+        (*MPIR_Process.cxx_call_op_fn)((in_), (inout_), (count_), (datatype_), uop); \
+    }                                                                                \
+    else {                                                                           \
+        (*uop)((in_), (inout_), &(count_), &(datatype_));                            \
+    }                                                                                \
+} while (0)
+
+#else
+#define call_uop(in_, inout_, count_, datatype_)      \
+    (*uop)((in_), (inout_), &(count_), &(datatype_))
+#endif
+
+/* Implements the reduce-scatter butterfly algorithm described in J. L. Traff's
+ * "An Improved Algorithm for (Non-commutative) Reduce-Scatter with an Application"
+ * from EuroPVM/MPI 2005.  This function currently only implements support for
+ * the power-of-2, block-regular case (all receive counts are equal). */
+#undef FUNCNAME
+#define FUNCNAME MPIR_Reduce_scatter_noncomm
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+static int MPIR_Reduce_scatter_noncomm (
+    void *sendbuf,
+    void *recvbuf,
+    int *recvcnts,
+    MPI_Datatype datatype,
+    MPI_Op op,
+    MPID_Comm *comm_ptr )
+{
+    int mpi_errno = MPI_SUCCESS;
+    int comm_size = comm_ptr->local_size;
+    int rank = comm_ptr->rank;
+    int pof2;
+    int log2_comm_size;
+    int i, k;
+    int recv_offset, send_offset;
+    int block_size, total_count, size;
+    MPI_Aint extent, true_extent, true_lb;
+    int is_commutative;
+    int buf0_was_inout;
+    void *tmp_buf0;
+    void *tmp_buf1;
+    void *result_ptr;
+    MPI_Comm comm = comm_ptr->handle;
+    MPI_User_function *uop;
+    MPID_Op *op_ptr;
+#ifdef HAVE_CXX_BINDING
+    int is_cxx_uop = 0;
+#endif
+    MPIU_CHKLMEM_DECL(3);
+
+    MPID_Datatype_get_extent_macro(datatype, extent);
+    /* assumes nesting is handled by the caller right now, may not be true in the future */
+    mpi_errno = NMPI_Type_get_true_extent(datatype, &true_lb, &true_extent);
+
+    if (HANDLE_GET_KIND(op) == HANDLE_KIND_BUILTIN) {
+        is_commutative = 1;
+        /* get the function by indexing into the op table */
+        uop = MPIR_Op_table[op%16 - 1];
+    }
+    else {
+        MPID_Op_get_ptr(op, op_ptr);
+        if (op_ptr->kind == MPID_OP_USER_NONCOMMUTE)
+            is_commutative = 0;
+        else
+            is_commutative = 1;
+
+#ifdef HAVE_CXX_BINDING
+        if (op_ptr->language == MPID_LANG_CXX) {
+            uop = (MPI_User_function *) op_ptr->function.c_function;
+            is_cxx_uop = 1;
+        }
+        else
+#endif
+        if ((op_ptr->language == MPID_LANG_C))
+            uop = (MPI_User_function *) op_ptr->function.c_function;
+        else
+            uop = (MPI_User_function *) op_ptr->function.f77_function;
+    }
+
+    pof2 = 1;
+    log2_comm_size = 0;
+    while (pof2 < comm_size) {
+        pof2 <<= 1;
+        ++log2_comm_size;
+    }
+
+    /* begin error checking */
+    MPIU_Assert(pof2 == comm_size); /* FIXME this version only works for power of 2 procs */
+
+    for (i = 0; i < (comm_size - 1); ++i) {
+        MPIU_Assert(recvcnts[i] == recvcnts[i+1]);
+    }
+    /* end error checking */
+
+    /* size of a block (count of datatype per block, NOT bytes per block) */
+    block_size = recvcnts[0];
+    total_count = block_size * comm_size;
+
+    MPIU_CHKLMEM_MALLOC(tmp_buf0, void *, true_extent * total_count, mpi_errno, "tmp_buf0");
+    MPIU_CHKLMEM_MALLOC(tmp_buf1, void *, true_extent * total_count, mpi_errno, "tmp_buf1");
+    /* adjust for potential negative lower bound in datatype */
+    tmp_buf0 = (void *)((char*)tmp_buf0 - true_lb);
+    tmp_buf1 = (void *)((char*)tmp_buf1 - true_lb);
+
+    /* Copy our send data to tmp_buf0.  We do this one block at a time and
+       permute the blocks as we go according to the mirror permutation. */
+    for (i = 0; i < comm_size; ++i) {
+        mpi_errno = MPIR_Localcopy((char *)(sendbuf == MPI_IN_PLACE ? recvbuf : sendbuf) + (i * true_extent * block_size), block_size, datatype,
+                                   (char *)tmp_buf0 + (mirror_permutation(i, log2_comm_size) * true_extent * block_size), block_size, datatype);
+        if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+    }
+    buf0_was_inout = 1;
+
+    send_offset = 0;
+    recv_offset = 0;
+    size = total_count;
+    for (k = 0; k < log2_comm_size; ++k) {
+        /* use a double-buffering scheme to avoid local copies */
+        char *incoming_data = (buf0_was_inout ? tmp_buf1 : tmp_buf0);
+        char *outgoing_data = (buf0_was_inout ? tmp_buf0 : tmp_buf1);
+        int peer = rank ^ (0x1 << k);
+        size /= 2;
+
+        if (rank > peer) {
+            /* we have the higher rank: send top half, recv bottom half */
+            recv_offset += size;
+        }
+        else {
+            /* we have the lower rank: recv top half, send bottom half */
+            send_offset += size;
+        }
+
+        mpi_errno = MPIC_Sendrecv(outgoing_data + send_offset*true_extent,
+                                  size, datatype, peer, MPIR_REDUCE_SCATTER_TAG,
+                                  incoming_data + recv_offset*true_extent,
+                                  size, datatype, peer, MPIR_REDUCE_SCATTER_TAG,
+                                  comm, MPI_STATUS_IGNORE);
+        /* always perform the reduction at recv_offset, the data at send_offset
+           is now our peer's responsibility */
+        if (rank > peer) {
+            /* higher ranked value so need to call op(received_data, my_data) */
+            call_uop(incoming_data + recv_offset*true_extent,
+                     outgoing_data + recv_offset*true_extent,
+                     size, datatype);
+            buf0_was_inout = buf0_was_inout;
+        }
+        else {
+            /* lower ranked value so need to call op(my_data, received_data) */
+            call_uop(outgoing_data + recv_offset*true_extent,
+                     incoming_data + recv_offset*true_extent,
+                     size, datatype);
+            buf0_was_inout = !buf0_was_inout;
+        }
+
+        /* the next round of send/recv needs to happen within the block (of size
+           "size") that we just received and reduced */
+        send_offset = recv_offset;
+    }
+
+    MPIU_Assert(size == recvcnts[rank]);
+
+    /* copy the reduced data to the recvbuf */
+    result_ptr = (char *)(buf0_was_inout ? tmp_buf0 : tmp_buf1) + recv_offset * true_extent;
+    mpi_errno = MPIR_Localcopy(result_ptr, size, datatype,
+                               recvbuf, size, datatype);
+fn_exit:
+    MPIU_CHKLMEM_FREEALL();
+    return mpi_errno;
+fn_fail:
+    goto fn_exit;
+}
+
-/* This is the default implementation of reduce_scatter. The algorithm is:
-
+
-   Algorithm: MPI_Reduce_scatter
-
@@ -40 +243 @@
-   their left neighbor as well in the recursive halving algorithm, and
-   then at  the end send the result back to the processes that didn't
- 
+
-   Therefore, if p is a power-of-two,
-   Cost = lgp.alpha + n.((p-1)/p).beta + n.((p-1)/p).gamma
@@ -49 +252 @@
-   because some processes do the work of their neighbors as well.
-
- 
+
-   we use a pairwise exchange algorithm similar to
-   the one used in MPI_Alltoall. At step i, each process sends n/p
- 
+
-   (rank-i).
-   Cost = (p-1).alpha + n.((p-1)/p).beta + n.((p-1)/p).gamma
@@ -66 +269 @@
-   Cost = lgp.alpha + n.(lgp-(p-1)/p).beta + n.(lgp-(p-1)/p).gamma
-
- 
+
-
-   End Algorithm: MPI_Reduce_scatter
-*/
-
+#undef FUNCNAME
+#define FUNCNAME MPIR_Reduce_scatter
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
-/* begin:nested */
-/* not declared static because a machine-specific function may call this one in some cases */
@@ -81 +288 @@
-    MPID_Comm *comm_ptr )
-{
-    static const char FCNAME[] = "MPIR_Reduce_scatter";
-    int   rank, comm_size, i;
-    MPI_Aint extent, true_extent, true_lb; 
@@ -101 +307 @@
-    int is_cxx_uop = 0;
-#endif
-
+
+
-    comm = comm_ptr->handle;
-    comm_size = comm_ptr->local_size;
@@ -110 +317 @@
-    MPIU_THREADPRIV_FIELD(op_errno) = 0;
-
+    MPIR_Nest_incr();
+
-    MPID_Datatype_get_extent_macro(datatype, extent);
-    mpi_errno = NMPI_Type_get_true_extent(datatype, &true_lb,
-                                          &true_extent);  
-
-
-
-
-
-
-
+
-    
-    if (HANDLE_GET_KIND(op) == HANDLE_KIND_BUILTIN) {
@@ -146 +349 @@
-    }
-
-
-
-
-
-
-
-
+
-
-    total_count = 0;
@@ -161 +358 @@
-    
-    if (total_count == 0) {
-
-
+
-    }
-
@@ -175 +371 @@
-    MPID_Ensure_Aint_fits_in_pointer(total_count * MPIR_MAX(true_extent, extent));
-
-    MPIR_Nest_incr();
-
-    if ((is_commutative) && (nbytes < MPIR_REDSCAT_COMMUTATIVE_LONG_MSG)) {
-        /* commutative and short. use recursive halving algorithm */
-
-        /* allocate temp. buffer to receive incoming data */
-
-
-
-
-
-
-
+
-        /* adjust for potential negative lower bound in datatype */
-        tmp_recvbuf = (void *)((char*)tmp_recvbuf - true_lb);
@@ -193 +381 @@
-        /* need to allocate another temporary buffer to accumulate
-           results because recvbuf may not be big enough */
-
-
-
-
-
-
-
+
-        /* adjust for potential negative lower bound in datatype */
-        tmp_results = (void *)((char*)tmp_results - true_lb);
@@ -211 +393 @@
-                                       tmp_results, total_count, datatype);
-        
-
-
-
-
-
-
-
+
-
-        pof2 = 1;
@@ -289 +465 @@
-               right (rank+1). */
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
-            
-            for (i=0; i<pof2; i++) {
@@ -371 +535 @@
-                                          comm);  
-
-
-
-
-
-
-
-
+
-                
-                /* tmp_recvbuf contains data received in this step.
@@ -410 +568 @@
-                                           recvcnts[rank], datatype, recvbuf,
-                                           recvcnts[rank], datatype);
-
-
-
-
-
-
-
+
-            }
-            
-            MPIU_Free(newcnts);
-            MPIU_Free(newdisps);
-        }
-
@@ -441 +591 @@
-                                      MPI_STATUS_IGNORE); 
-            }
-
-
-
-
-
-
-
-
-
-
-
+
+
-    }
-    
@@ -464 +605 @@
-                                       recvcnts[rank], datatype, recvbuf,
-                                       recvcnts[rank], datatype);
-
-
-
-
-
-
-
+
-        }
-        
-        /* allocate temporary buffer to store incoming data */
-
-
-
-
-
-
-
-
+
-        /* adjust for potential negative lower bound in datatype */
-        tmp_recvbuf = (void *)((char*)tmp_recvbuf - true_lb);
@@ -506 +634 @@
-                                          MPI_STATUS_IGNORE);
-            
-
-
-
-
-
-
-
+
-            
-            if (is_commutative || (src < rank)) {
@@ -582 +704 @@
-                                               recvcnts[rank], datatype); 
-                }
-
-
-
-
-
-
-
-
-
-
-
+
+
+
-        
-        /* if MPI_IN_PLACE, move output data to the beginning of
@@ -602 +716 @@
-                                       recvbuf, 
-                                       recvcnts[rank], datatype); 
-
-
-
-
-
-
-
+
-        }
-    }
-    
-    if (!is_commutative) {
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-                }
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-                    }
-                    /* recv only if this proc. doesn't have data and sender
-                       has data */
-                    else if ((dst < rank) && 
-                             (dst < tree_root + nprocs_completed) &&
-                             (rank >= tree_root + nprocs_completed)) {
-                        mpi_errno = MPIC_Recv(tmp_recvbuf, 1, recvtype, dst,
-                                              MPIR_REDUCE_SCATTER_TAG,
-                                              comm, MPI_STATUS_IGNORE); 
-                        received = 1;
-                        /* --BEGIN ERROR HANDLING-- */
-                        if (mpi_errno)
-                        {
-                            mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", 0);
-                            return mpi_errno;
-                        }
-                        /* --END ERROR HANDLING-- */
-                    }
-                    tmp_mask >>= 1;
-                    k--;
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-                    }
-                }
-
+
+
+
+
+
+
+
+
+
+
+
-#ifdef HAVE_CXX_BINDING
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-                    }
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-                    }
-                    /* copy result back into tmp_results */
-                    mpi_errno = MPIR_Localcopy(tmp_recvbuf, 1, recvtype, 
-                                               tmp_results, 1, recvtype);
-                    /* --BEGIN ERROR HANDLING-- */
-                    if (mpi_errno)
-                    {
-                        mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", 0);
-                        return mpi_errno;
-                    }
-                    /* --END ERROR HANDLING-- */
-                }
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    MPIR_Nest_decr();
-    /* check if multiple threads are calling this collective function */
@@ -892 +967 @@
-
-    return (mpi_errno);
+fn_fail:
+    goto fn_exit;
-}
-/* end:nested */
-
+#undef FUNCNAME
+#define FUNCNAME MPIR_Reduce_scatter
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
-/* begin:nested */
-/* not declared static because a machine-specific function may call this one in some cases */
@@ -911 +992 @@
-*/
-    
-    static const char FCNAME[] = "MPIR_Reduce_scatter_inter";
-    int rank, mpi_errno, root, local_size, total_count, i;
-    MPI_Aint true_extent, true_lb = 0, extent;
@@ -1046 +1126 @@
-#undef FUNCNAME
-#define FUNCNAME MPI_Reduce_scatter
-
+
+
-/*@
-
@@ -1081 +1162 @@
-                       MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
-{
-    static const char FCNAME[] = "MPI_Reduce_scatter";
-    int mpi_errno = MPI_SUCCESS;
-    MPID_Comm *comm_ptr = NULL;

mpich2/trunk/test/mpi/coll/redscat2.c

@@ -91 +91 @@
-        for (i=0; i<(size*block_size); i++) 
-            sendbuf[i] = rank + i;
+        for (i=0; i<block_size; i++)
+            recvbuf[i] = 0xdeadbeef;
-        recvcounts = (int *)malloc( size * sizeof(int) );
-        for (i=0; i<size; i++)