coll/han: Implement MPI_Alltoallv in han using SMSC and XPMEM

Extension of the previous han MPI_Alltoall algorithm, this change adds
MPI_Alltoallv to han for a hierarchy-aware algorithm which uses XPMEM
via the SMSC module in order to directly read data from ranks on the
same host.

The provides significant speed-up over the basic implementation when
small messages are used, as many messages can be coalesced and packed
into fewer sends.

Introduces MCA parameters:
 - coll_han_alltoallv_smsc_avg_send_limit
 - coll_han_alltoallv_smsc_noncontig_limit

Signed-off-by: Luke Robison <[email protected]>

Author: Luke Robison

ompi/mca/coll/han/Makefile.am

@@ -16,6 +16,7 @@ coll_han.h \
 coll_han_trigger.h \
 coll_han_algorithms.h \
 coll_han_alltoall.c \
+coll_han_alltoallv.c \
 coll_han_dynamic.h \
 coll_han_dynamic_file.h \
 coll_han_barrier.c \

ompi/mca/coll/han/coll_han.h

@@ -199,6 +199,7 @@ typedef struct mca_coll_han_op_module_name_t {
     mca_coll_han_op_up_low_module_name_t scatter;
     mca_coll_han_op_up_low_module_name_t scatterv;
     mca_coll_han_op_up_low_module_name_t alltoall;
+    mca_coll_han_op_up_low_module_name_t alltoallv;
 } mca_coll_han_op_module_name_t;
 
 /**
@@ -260,6 +261,11 @@ typedef struct mca_coll_han_component_t {
     /* alltoall: parallel stages */
     int32_t han_alltoall_pstages;
 
+    /* low level module for alltoallv */
+    uint32_t han_alltoallv_low_module;
+    int64_t han_alltoallv_smsc_avg_send_limit;
+    double han_alltoallv_smsc_noncontig_activation_limit;
+
 
     /* name of the modules */
     mca_coll_han_op_module_name_t han_op_module_name;
@@ -286,6 +292,8 @@ typedef struct mca_coll_han_component_t {
 
     /* Define maximum dynamic errors printed by rank 0 with a 0 verbosity level */
     int max_dynamic_errors;
+
+    opal_free_list_t pack_buffers;
 } mca_coll_han_component_t;
 
 /*
@@ -297,6 +305,7 @@ typedef struct mca_coll_han_single_collective_fallback_s
     union
     {
         mca_coll_base_module_alltoall_fn_t alltoall;
+        mca_coll_base_module_alltoallv_fn_t alltoallv;
         mca_coll_base_module_allgather_fn_t allgather;
         mca_coll_base_module_allgatherv_fn_t allgatherv;
         mca_coll_base_module_allreduce_fn_t allreduce;
@@ -319,6 +328,7 @@ typedef struct mca_coll_han_single_collective_fallback_s
 typedef struct mca_coll_han_collectives_fallback_s
 {
     mca_coll_han_single_collective_fallback_t alltoall;
+    mca_coll_han_single_collective_fallback_t alltoallv;
     mca_coll_han_single_collective_fallback_t allgather;
     mca_coll_han_single_collective_fallback_t allgatherv;
     mca_coll_han_single_collective_fallback_t allreduce;
@@ -384,6 +394,9 @@ OBJ_CLASS_DECLARATION(mca_coll_han_module_t);
 #define previous_alltoall           fallback.alltoall.alltoall
 #define previous_alltoall_module    fallback.alltoall.module
 
+#define previous_alltoallv           fallback.alltoallv.alltoallv
+#define previous_alltoallv_module    fallback.alltoallv.module
+
 #define previous_allgather          fallback.allgather.allgather
 #define previous_allgather_module   fallback.allgather.module
 
@@ -440,6 +453,7 @@ OBJ_CLASS_DECLARATION(mca_coll_han_module_t);
         HAN_UNINSTALL_COLL_API(COMM, HANM, allgather);                 \
         HAN_UNINSTALL_COLL_API(COMM, HANM, allgatherv);                \
         HAN_UNINSTALL_COLL_API(COMM, HANM, alltoall);                  \
+        HAN_UNINSTALL_COLL_API(COMM, HANM, alltoallv);                  \
         han_module->enabled = false;  /* entire module set to pass-through from now on */ \
     } while(0)
 
@@ -503,6 +517,9 @@ int
 mca_coll_han_alltoall_intra_dynamic(ALLTOALL_BASE_ARGS,
                                     mca_coll_base_module_t *module);
 int
+mca_coll_han_alltoallv_intra_dynamic(ALLTOALLV_BASE_ARGS,
+                                    mca_coll_base_module_t *module);
+int
 mca_coll_han_allgather_intra_dynamic(ALLGATHER_BASE_ARGS,
                                      mca_coll_base_module_t *module);
 int
@@ -566,4 +583,7 @@ static inline struct mca_smsc_endpoint_t *mca_coll_han_get_smsc_endpoint (struct
     return (struct mca_smsc_endpoint_t *) proc->proc_endpoints[OMPI_PROC_ENDPOINT_TAG_SMSC];
 }
 
+#define COLL_HAN_PACKBUF_PAYLOAD_BYTES (128*1024)
+
+
 #endif                          /* MCA_COLL_HAN_EXPORT_H */

ompi/mca/coll/han/coll_han_algorithms.c

@@ -82,6 +82,10 @@ mca_coll_han_algorithm_value_t*  mca_coll_han_available_algorithms[COLLCOUNT] =
         {"smsc", (fnptr_t)&mca_coll_han_alltoall_using_smsc}, // 2-level
         { 0 }
     },
+    [ALLTOALLV] = (mca_coll_han_algorithm_value_t[]){
+        {"smsc", (fnptr_t)&mca_coll_han_alltoallv_using_smsc}, // 2-level
+        { 0 }
+    },
 };
 
 int

ompi/mca/coll/han/coll_han_algorithms.h

@@ -214,5 +214,10 @@ int
 mca_coll_han_alltoall_using_smsc(ALLTOALL_BASE_ARGS,
                                     mca_coll_base_module_t *module);
 
+/* Alltoallv */
+int
+mca_coll_han_alltoallv_using_smsc(ALLTOALLV_BASE_ARGS,
+                                    mca_coll_base_module_t *module);
+
 
 #endif

ompi/mca/coll/han/coll_han_alltoall.c

@@ -31,7 +31,11 @@
 #include "opal/mca/rcache/rcache.h"
 #include "ompi/mca/osc/base/base.h"
 
-
+#if 1
+#define DEBUG_PRINT(...) do{ fprintf( stdout, __VA_ARGS__ ); } while( false )
+#else
+#define DEBUG_PRINT(...) do{ } while ( false )
+#endif
 
 /* Who is the given ranks partner during the exchange?
    This function will require rounds comm_size-many rounds, and your partner
@@ -331,6 +335,22 @@ int mca_coll_han_alltoall_using_smsc(
                 /* pack the data directly into local leader's sendbuf */
                 packed_size_tmp = packed_size;
                 rc = opal_convertor_pack(&convertor, &iov, &iov_count, &packed_size_tmp);
+
+                {
+                    char tmpstr[16*4+1];
+                    char leadstr[256];
+                    char *cur = tmpstr;
+                    if (ii_push_data) {
+                        sprintf(leadstr,"push: my([%d]) data to [%d] for remote [%d]",w_rank,jlow+up_rank*low_size,remote_wrank);
+                    } else {
+                        sprintf(leadstr,"pull: from [%d] to me([%d]) for remote [%d]",jlow+up_rank*low_size,w_rank,low_size*up_partner+low_rank);
+                    }
+                    for (int jbyte=0; jbyte<(int)packed_size_tmp && jbyte<16; jbyte++) {
+                        cur += sprintf(cur,"%3d ",((char*)(to_addr))[jbyte]);
+                    }
+                    DEBUG_PRINT("%s: %s\n",leadstr, tmpstr);
+                }
+
                 opal_convertor_cleanup(&convertor);
 
                 if (1 != rc) {
@@ -348,6 +368,24 @@ int mca_coll_han_alltoall_using_smsc(
                 low_comm->c_coll->coll_barrier(low_comm, low_comm->c_coll->coll_barrier_module);
             }
 
+            {
+                char *sendbuftmp;
+                char tmpstr[16*4+1];
+                char leadstr[256];
+                char *cur = tmpstr;
+                if (use_isend) {
+                    sprintf(leadstr,"isend slot %d: from [%d] to [%d]",jfan_slot,w_rank,first_remote_wrank+low_rank);
+                    sendbuftmp = &send_bounce[send_bytes_per_fan*jfan_slot];
+                } else {
+                    sprintf(leadstr,"send: from [%d] to [%d]",w_rank,first_remote_wrank+low_rank);
+                    sendbuftmp = send_bounce;
+                }
+                for (int jbyte=0; jbyte<(int)send_bytes_per_fan && jbyte<16; jbyte++) {
+                    cur += sprintf(cur,"%3d ",sendbuftmp[jbyte]);
+                }
+                DEBUG_PRINT("%s: %s\n",leadstr, tmpstr);
+            }
+
             if (use_isend == 0) {
                 MCA_PML_CALL(send
                         (send_bounce,