Redo the non contiguous datatype part in SOLO module.

Known problems:
1. bcast: has bug if a datatype is from MPI_Bottom
2. reduce \ allreduce : fix or remove the non contiguous support.

Author: dycz0fx

ompi/mca/coll/solo/coll_solo.h

@@ -59,12 +59,6 @@ typedef struct mca_coll_solo_module_t {
     /* Whether this module has been lazily initialized or not yet */
     bool enabled;
 
-    /**
-     * osc alrogithms attach memory blocks to this bynamic window and use it to perform one-sided 
-     * communications. 
-     */
-    MPI_Win dynamic_win;
-
     /**
      * This window is created by ompi_win_allocate_shared such that each process contains a shared 
      * memory data buffer, and this data buffer is divided into two parts - ctrl_bufs and data_bufs.
@@ -101,18 +95,6 @@ mca_coll_base_module_t *mca_coll_solo_comm_query(struct ompi_communicator_t *com
 /* Lazily enable a module (since it involves expensive memory allocation, etc.) */
 int mca_coll_solo_lazy_enable(mca_coll_base_module_t * module, struct ompi_communicator_t *comm);
 
-/* Attach a memory block to the dynamic_win of a communicator */
-char **mca_coll_solo_attach_buf(mca_coll_solo_module_t * solo_module,
-                                struct ompi_communicator_t *comm,
-                                char *local_buf, 
-                                size_t local_buf_size);
-
-/* Detach a memory block from the dynamic_win of a communicator */
-void mca_coll_solo_detach_buf(mca_coll_solo_module_t * solo_module,
-                              struct ompi_communicator_t *comm,
-                              char *local_buf, 
-                              char ***attached_bufs);
-
 /* Setup and initialize the static_win of a communicator */
 void mca_coll_solo_setup_static_win(mca_coll_solo_module_t *solo_module,
                                     struct ompi_communicator_t *comm, 
@@ -135,26 +117,13 @@ int mca_coll_solo_bcast_linear_intra_memcpy(void *buff, int count,
                                             struct ompi_communicator_t *comm, 
                                             mca_coll_base_module_t * module);
 
-int mca_coll_solo_bcast_linear_intra_osc(void *buff, int count,
-                                         struct ompi_datatype_t *dtype,
-                                         int root, 
-                                         struct ompi_communicator_t *comm, 
-                                         mca_coll_base_module_t * module);
-
 int mca_coll_solo_bcast_pipeline_intra_memcpy(void *buff, int count,
                                               struct ompi_datatype_t *dtype, 
                                               int root,
                                               struct ompi_communicator_t *comm, 
                                               mca_coll_base_module_t * module,
                                               size_t seg_size);
 
-int mca_coll_solo_bcast_pipeline_intra_osc(void *buff, int count,
-                                           struct ompi_datatype_t *dtype, 
-                                           int root,
-                                           struct ompi_communicator_t *comm, 
-                                           mca_coll_base_module_t * module,
-                                           size_t seg_size);
-
 /* MPI_Reduce algorithms */
 int mca_coll_solo_reduce_intra(const void *sbuf, void *rbuf, int count,
                                struct ompi_datatype_t *dtype,
@@ -163,24 +132,13 @@ int mca_coll_solo_reduce_intra(const void *sbuf, void *rbuf, int count,
                                struct ompi_communicator_t *comm, 
                                mca_coll_base_module_t * module);
 
-int mca_coll_solo_reduce_ring_intra(const void *sbuf, void *rbuf, int count,
-                                    struct ompi_datatype_t *dtype,
-                                    struct ompi_op_t *op, int root,
-                                    struct ompi_communicator_t *comm,
-                                    mca_coll_base_module_t * module);
-
 int mca_coll_solo_reduce_ring_intra_memcpy(const void *sbuf, void *rbuf, int count,
                                            struct ompi_datatype_t *dtype,
                                            struct ompi_op_t *op,
                                            int root, 
                                            struct ompi_communicator_t
                                            *comm, mca_coll_base_module_t * module);
 
-int mca_coll_solo_reduce_ring_intra_osc(const void *sbuf, void *rbuf, int count,
-                                        struct ompi_datatype_t *dtype,
-                                        struct ompi_op_t *op, int root,
-                                        struct ompi_communicator_t *comm,
-                                        mca_coll_base_module_t * module);
 
 /* MPI_Allreduce algorithms */
 int mca_coll_solo_allreduce_intra(const void *sbuf, void *rbuf, int count,
@@ -195,10 +153,39 @@ int mca_coll_solo_allreduce_ring_intra_memcpy(const void *sbuf, void *rbuf, int
                                               struct ompi_communicator_t *comm, 
                                               mca_coll_base_module_t * module);
 
-int mca_coll_solo_allreduce_ring_intra_osc(const void *sbuf, void *rbuf, int count, 
-                                           struct ompi_datatype_t *dtype, 
-                                           struct ompi_op_t *op, 
-                                           struct ompi_communicator_t *comm, 
-                                           mca_coll_base_module_t * module);
+
+/* Solo pack to shared memory */
+static inline void mca_coll_solo_pack_to_shared(void *local_buf, void *shared_buf, struct ompi_datatype_t *dtype, int count, ptrdiff_t extent) {
+    if (ompi_datatype_is_predefined(dtype)) {
+        memcpy((char *) shared_buf, (char *) local_buf, count * extent);
+    }
+    else {
+        MPI_Aint pos = 0;
+        ompi_datatype_pack_external("external32", local_buf, count, dtype, shared_buf, count * extent, &pos);
+    }
+}
+
+/* Solo unpack from shared memory */
+static inline void mca_coll_solo_unpack_from_shared(void *local_buf, void *shared_buf, struct ompi_datatype_t *dtype, int count, ptrdiff_t extent) {
+    if (ompi_datatype_is_predefined(dtype)) {
+        memcpy((char *) local_buf, (char *) shared_buf, count * extent);
+    }
+    else {
+        MPI_Aint pos = 0;
+        ompi_datatype_unpack_external("external32", shared_buf, count * extent, &pos, local_buf, count, dtype);
+    }
+}
+
+/* Solo copy from source to target */
+static inline void mca_coll_solo_copy(void *source, void *target, struct ompi_datatype_t *dtype, int count, ptrdiff_t extent) {
+    if (ompi_datatype_is_predefined(dtype)) {
+        memcpy(target, source, count * extent);
+    }
+    else {
+        ompi_datatype_copy_content_same_ddt(dtype, count, target, source);
+    }
+    return;
+} 
+
 END_C_DECLS
 #endif                          /* MCA_COLL_SOLO_EXPORT_H */

ompi/mca/coll/solo/coll_solo_allreduce.c

@@ -18,14 +18,7 @@ int mca_coll_solo_allreduce_intra(const void *sbuf, void *rbuf,
                                     struct ompi_communicator_t *comm,
                                     mca_coll_base_module_t * module)
 {
-    if (ompi_datatype_is_contiguous_memory_layout(dtype, count)) {
-        mca_coll_solo_allreduce_ring_intra_memcpy(sbuf, rbuf, count, dtype, op, comm, module);
-    }
-    else {
-        mca_coll_solo_allreduce_ring_intra_osc(sbuf, rbuf, count, dtype, op, comm, module);
-    }
-    return OMPI_SUCCESS;
-
+    return mca_coll_solo_allreduce_ring_intra_memcpy(sbuf, rbuf, count, dtype, op, comm, module);
 }
 
 
@@ -91,10 +84,7 @@ int mca_coll_solo_allreduce_ring_intra_memcpy(const void *sbuf, void *rbuf, int
         }
     } else {
         /* For the messages which are greater than mpool_large_block_size*np, invoke this reduce multiple times */
-        int seg_count = count;
-        size_t typelng;
-        ompi_datatype_type_size(dtype, &typelng);
-        COLL_BASE_COMPUTED_SEGCOUNT(mca_coll_solo_component.mpool_large_block_size, typelng, seg_count);
+        int seg_count = mca_coll_solo_component.mpool_large_block_size / extent;
         int num_segments = (count + seg_count - 1) / seg_count;
         int last_count = count - seg_count * (num_segments - 1);
         for (int i = 0; i < num_segments; i++) {
@@ -122,7 +112,7 @@ int mca_coll_solo_allreduce_ring_intra_memcpy(const void *sbuf, void *rbuf, int
         }
         /* At first iteration, copy local data to the solo data buffer */
         if (cur == rank) {
-            memcpy(data_bufs[cur], (char *) sbuf + cur * l_seg_count * extent, seg_count * extent);
+            mca_coll_solo_copy((void *) ((char *) sbuf + cur * l_seg_count * extent), (void *) data_bufs[cur], dtype, seg_count, extent);
             mac_coll_solo_barrier_intra(comm, module);
 
         }
@@ -147,7 +137,7 @@ int mca_coll_solo_allreduce_ring_intra_memcpy(const void *sbuf, void *rbuf, int
         } else {
             seg_count = count - i * l_seg_count;
         }
-        memcpy((char *) c, data_bufs[i], seg_count * extent);
+        mca_coll_solo_copy((void *) data_bufs[i], (void *) c, dtype, seg_count, extent);
         c = c + seg_count * extent;
     }
     mac_coll_solo_barrier_intra(comm, module);
@@ -167,127 +157,4 @@ int mca_coll_solo_allreduce_ring_intra_memcpy(const void *sbuf, void *rbuf, int
 
     }
     return OMPI_SUCCESS;
-}
-
-int mca_coll_solo_allreduce_ring_intra_osc(const void *sbuf, void *rbuf, int count,
-                                             struct ompi_datatype_t *dtype,
-                                             struct ompi_op_t *op, 
-                                             struct ompi_communicator_t *comm, 
-                                             mca_coll_base_module_t * module)
-{
-    mca_coll_solo_module_t *solo_module = (mca_coll_solo_module_t *) module;
-    int size = ompi_comm_size(comm);
-    int rank = ompi_comm_rank(comm);
-    int i;
-    ptrdiff_t extent, lower_bound;
-    ompi_datatype_get_extent(dtype, &lower_bound, &extent);
-
-    /* Enable solo module if necessary */
-    if (!solo_module->enabled) {
-        mca_coll_solo_lazy_enable(module, comm);
-    }
-
-    /* Set up segment count */
-    int seg_count, l_seg_count;
-    seg_count = count / size;
-    l_seg_count = seg_count;
-    if (rank == size - 1) {
-        seg_count = count - rank * l_seg_count;
-    }
-
-    char **data_bufs = NULL;
-    int id;
-    MPI_Win cur_win;
-    char *local_buf = NULL;
-    if ((size_t) l_seg_count * extent <= mca_coll_solo_component.static_block_size) {
-        data_bufs = (char **) malloc(sizeof(char *) * size);
-        for (i = 0; i < size; i++) {
-            data_bufs[i] = (char *) 0 + 4 * opal_cache_line_size;
-        }
-        cur_win = solo_module->static_win;
-    } else if ((size_t) l_seg_count * extent <= mca_coll_solo_component.mpool_large_block_size) {
-        id = mca_coll_solo_mpool_request(mca_coll_solo_component.solo_mpool, l_seg_count * extent);
-        local_buf =
-            mca_coll_solo_mpool_calculate(mca_coll_solo_component.solo_mpool, id,
-                                          l_seg_count * extent);
-        data_bufs = mca_coll_solo_attach_buf(solo_module, comm, local_buf, l_seg_count * extent);
-        cur_win = solo_module->dynamic_win;
-    } else {
-        /* For the messages which are greater than mpool_large_block_size*np, invoke this reduce multiple times */
-        int seg_count = count;
-        size_t typelng;
-        ompi_datatype_type_size(dtype, &typelng);
-        COLL_BASE_COMPUTED_SEGCOUNT(mca_coll_solo_component.mpool_large_block_size, typelng, seg_count);
-        int num_segments = (count + seg_count - 1) / seg_count;
-        int last_count = count - seg_count * (num_segments - 1);
-        for (int i = 0; i < num_segments; i++) {
-            char *temp_sbuf = (char *)sbuf + seg_count * extent * i;
-            char *temp_rbuf = (char *)rbuf + seg_count * extent * i;
-            int temp_count = seg_count;
-            if (i == num_segments - 1) {
-                temp_count = last_count;
-            }
-            mca_coll_solo_allreduce_ring_intra_osc(temp_sbuf, temp_rbuf, temp_count, dtype, op, 
-                                                   comm, module);
-        }
-        return MPI_SUCCESS;
-    }
-
-    *(int *) (solo_module->ctrl_bufs[rank]) = rank;
-    mac_coll_solo_barrier_intra(comm, module);
-
-    int cur = rank;
-    for (i = 0; i < size; i++) {
-        if (cur != size - 1) {
-            seg_count = l_seg_count;
-        } else {
-            seg_count = count - cur * l_seg_count;
-        }
-        /* At first iteration, copy local data to the solo data buffer */
-        if (cur == rank) {
-            cur_win->w_osc_module->osc_fence(0, cur_win);
-            cur_win->w_osc_module->osc_put((char *) sbuf +
-                                           cur * l_seg_count * extent,
-                                           seg_count, dtype, cur,
-                                           (ptrdiff_t) data_bufs[cur], seg_count, dtype, cur_win);
-            cur_win->w_osc_module->osc_fence(0, cur_win);
-        }
-        /* For other iterations, do operations on the solo data buffer */
-        else {
-            cur_win->w_osc_module->osc_accumulate((char *) sbuf +
-                                                  cur * l_seg_count *
-                                                  extent, seg_count, dtype, cur, (ptrdiff_t)
-                                                  data_bufs[cur], seg_count, dtype, op, cur_win);
-            cur_win->w_osc_module->osc_fence(0, cur_win);
-        }
-        cur = (cur - 1 + size) % size;
-        *(int *) (solo_module->ctrl_bufs[rank]) =
-            (*(int *) (solo_module->ctrl_bufs[rank]) + 1) % size;
-        mac_coll_solo_barrier_intra(comm, module);
-
-    }
-    /* At last, all the processes copies data from the solo data buffer */
-    char *c;
-    c = rbuf;
-    for (i = 0; i < size; i++) {
-        if (i != size - 1) {
-            seg_count = l_seg_count;
-        } else {
-            seg_count = count - i * l_seg_count;
-        }
-        cur_win->w_osc_module->osc_get(c, seg_count, dtype, i,
-                                       (ptrdiff_t) data_bufs[i], seg_count, dtype, cur_win);
-        c = c + seg_count * extent;
-    }
-    cur_win->w_osc_module->osc_fence(0, cur_win);
-    if ((size_t) l_seg_count * extent <= mca_coll_solo_component.static_block_size) {
-        if (data_bufs != NULL) {
-            free(data_bufs);
-            data_bufs = NULL;
-        }
-    } else if ((size_t) l_seg_count * extent <= mca_coll_solo_component.mpool_large_block_size) {
-        mca_coll_solo_detach_buf(solo_module, comm, local_buf, &data_bufs);
-        mca_coll_solo_mpool_return(mca_coll_solo_component.solo_mpool, id, l_seg_count * extent);
-    } 
-    return OMPI_SUCCESS;
-}
+}