Merge pull request #1080 from kawashima-fj/pr/v2.x/parallel-rma-fix

v2.x: osc/pt2pt: Fix tag conflicts on parallel RMA communications

Author: hppritcha

ompi/mca/osc/pt2pt/osc_pt2pt.h

@@ -14,6 +14,7 @@
  * Copyright (c) 2012-2013 Sandia National Laboratories.  All rights reserved.
  * Copyright (c) 2015      Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
+ * Copyright (c) 2016      FUJITSU LIMITED.  All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -150,7 +151,6 @@ struct ompi_osc_pt2pt_module_t {
 
     /** cyclic counter for a unique tage for long messages. */
     uint32_t tag_counter;
-    uint32_t rtag_counter;
 
     /* Number of outgoing fragments that have completed since the
        begining of time */
@@ -637,13 +637,15 @@ static inline void osc_pt2pt_add_pending (ompi_osc_pt2pt_pending_t *pending)
 /**
  * get_tag:
  *
- * @short Get a send/recv tag for large memory operations.
+ * @short Get a send/recv base tag for large memory operations.
  *
  * @param[in] module - OSC PT2PT module
  *
- * @long This function aquires a 16-bit tag for use with large memory operations. The
+ * @long This function acquires a 16-bit tag for use with large memory operations. The
  *       tag will be odd or even depending on if this is in a passive target access
- *       or not.
+ *       or not. An actual tag that will be passed to PML send/recv function is given
+ *       by tag_to_target or tag_to_origin function depending on the communication
+ *       direction.
  */
 static inline int get_tag(ompi_osc_pt2pt_module_t *module)
 {
@@ -654,14 +656,32 @@ static inline int get_tag(ompi_osc_pt2pt_module_t *module)
     return (tmp & OSC_PT2PT_FRAG_MASK) | !!(module->passive_target_access_epoch);
 }
 
-static inline int get_rtag(ompi_osc_pt2pt_module_t *module)
+/**
+ * tag_to_target:
+ *
+ * @short Get a tag used for PML send/recv communication from an origin to a target.
+ *
+ * @param[in] tag - base tag given by get_tag function.
+ */
+static inline int tag_to_target(int tag)
 {
-    /* the LSB of the tag is used be the receiver to determine if the
-       message is a passive or active target (ie, where to mark
-       completion). */
-    int32_t tmp = OPAL_THREAD_ADD32((volatile int32_t *) &module->rtag_counter, 4);
-    return (tmp & OSC_PT2PT_FRAG_MASK) | !!(module->passive_target_access_epoch);
+    /* (returned_tag >> 1) & 0x1 == 0 */
+    return tag + 0;
 }
+
+/**
+ * tag_to_origin:
+ *
+ * @short Get a tag used for PML send/recv communication from a target to an origin.
+ *
+ * @param[in] tag - base tag given by get_tag function.
+ */
+static inline int tag_to_origin(int tag)
+{
+    /* (returned_tag >> 1) & 0x1 == 1 */
+    return tag + 2;
+}
+
 /**
  * ompi_osc_pt2pt_accumulate_lock:
  *

ompi/mca/osc/pt2pt/osc_pt2pt_comm.c

@@ -14,6 +14,7 @@
  * Copyright (c) 2012-2013 Sandia National Laboratories.  All rights reserved.
  * Copyright (c) 2015      Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
+ * Copyright (c) 2016      FUJITSU LIMITED.  All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -363,9 +364,9 @@ static inline int ompi_osc_pt2pt_put_w_req (const void *origin_addr, int origin_
 
             OBJ_RETAIN(target_dt);
 
-            ret = ompi_osc_pt2pt_isend_w_cb ((void *) packed_ddt, ddt_len, MPI_BYTE, target,
-                                            tag, module->comm, ompi_osc_pt2pt_dt_send_complete,
-                                            target_dt);
+            ret = ompi_osc_pt2pt_isend_w_cb ((void *) packed_ddt, ddt_len, MPI_BYTE,
+                                            target, tag_to_target(tag), module->comm,
+                                            ompi_osc_pt2pt_dt_send_complete, target_dt);
             if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
                 break;
             }
@@ -394,8 +395,8 @@ static inline int ompi_osc_pt2pt_put_w_req (const void *origin_addr, int origin_
             header->tag = tag;
             osc_pt2pt_hton(header, proc);
 
-            ret = ompi_osc_pt2pt_data_isend (module,origin_addr, origin_count, origin_dt, target, tag,
-                                             request);
+            ret = ompi_osc_pt2pt_data_isend (module,origin_addr, origin_count, origin_dt,
+                                            target, tag_to_target(tag), request);
         }
     } while (0);
 
@@ -491,7 +492,7 @@ ompi_osc_pt2pt_accumulate_w_req (const void *origin_addr, int origin_count,
          }
 
         is_long_msg = true;
-        tag = get_rtag (module);
+        tag = get_tag (module);
     }
 
     if (is_long_msg) {
@@ -519,9 +520,9 @@ ompi_osc_pt2pt_accumulate_w_req (const void *origin_addr, int origin_count,
 
             OBJ_RETAIN(target_dt);
 
-            ret = ompi_osc_pt2pt_isend_w_cb ((void *) packed_ddt, ddt_len, MPI_BYTE, target,
-                                            tag, module->comm, ompi_osc_pt2pt_dt_send_complete,
-                                            target_dt);
+            ret = ompi_osc_pt2pt_isend_w_cb ((void *) packed_ddt, ddt_len, MPI_BYTE,
+                                            target, tag_to_target(tag), module->comm,
+                                            ompi_osc_pt2pt_dt_send_complete, target_dt);
             if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
                 break;
             }
@@ -553,8 +554,8 @@ ompi_osc_pt2pt_accumulate_w_req (const void *origin_addr, int origin_count,
             OPAL_OUTPUT_VERBOSE((25, ompi_osc_base_framework.framework_output,
                                  "acc: starting long accumulate with tag %d", tag));
 
-            ret = ompi_osc_pt2pt_data_isend (module, origin_addr, origin_count, origin_dt, target, tag,
-                                             request);
+            ret = ompi_osc_pt2pt_data_isend (module, origin_addr, origin_count, origin_dt,
+                                            target, tag_to_target(tag), request);
         }
     } while (0);
 
@@ -663,7 +664,8 @@ int ompi_osc_pt2pt_compare_and_swap (const void *origin_addr, const void *compar
     osc_pt2pt_copy_for_send (ptr, dt->super.size, compare_addr, proc, 1, dt);
 
     request->outstanding_requests = 1;
-    ret = ompi_osc_pt2pt_irecv_w_cb (result_addr, 1, dt, target, tag, module->comm,
+    ret = ompi_osc_pt2pt_irecv_w_cb (result_addr, 1, dt,
+                                    target, tag_to_origin(tag), module->comm,
                                     NULL, ompi_osc_pt2pt_req_comm_complete, request);
     if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
         return ret;
@@ -786,7 +788,7 @@ static inline int ompi_osc_pt2pt_rget_internal (void *origin_addr, int origin_co
     ret = ompi_osc_pt2pt_frag_alloc(module, target, frag_len, &frag, &ptr, false, release_req);
     if (OMPI_SUCCESS != ret) {
         /* allocate space for the header plus space to store ddt_len */
-        frag_len = sizeof(ompi_osc_pt2pt_header_put_t) + 8;
+        frag_len = sizeof(ompi_osc_pt2pt_header_get_t) + 8;
         ret = ompi_osc_pt2pt_frag_alloc(module, target, frag_len, &frag, &ptr, false, release_req);
         if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
             return OMPI_ERR_OUT_OF_RESOURCE;
@@ -827,9 +829,9 @@ static inline int ompi_osc_pt2pt_rget_internal (void *origin_addr, int origin_co
 
             OBJ_RETAIN(target_dt);
 
-            ret = ompi_osc_pt2pt_isend_w_cb ((void *) packed_ddt, ddt_len, MPI_BYTE, target,
-                                            tag, module->comm, ompi_osc_pt2pt_dt_send_complete,
-                                            target_dt);
+            ret = ompi_osc_pt2pt_isend_w_cb ((void *) packed_ddt, ddt_len, MPI_BYTE,
+                                            target, tag_to_target(tag), module->comm,
+                                            ompi_osc_pt2pt_dt_send_complete, target_dt);
             if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
                 break;
             }
@@ -843,8 +845,9 @@ static inline int ompi_osc_pt2pt_rget_internal (void *origin_addr, int origin_co
 
         /* TODO -- store the request somewhere so we can cancel it on error */
         pt2pt_request->outstanding_requests = 1;
-        ret = ompi_osc_pt2pt_irecv_w_cb (origin_addr, origin_count, origin_dt, target, tag,
-                                        module->comm, NULL, ompi_osc_pt2pt_req_comm_complete, pt2pt_request);
+        ret = ompi_osc_pt2pt_irecv_w_cb (origin_addr, origin_count, origin_dt,
+                                        target, tag_to_origin(tag), module->comm,
+                                        NULL, ompi_osc_pt2pt_req_comm_complete, pt2pt_request);
     } while (0);
 
     if (OMPI_SUCCESS == ret) {
@@ -1045,9 +1048,9 @@ int ompi_osc_pt2pt_rget_accumulate_internal (const void *origin_addr, int origin
 
             OBJ_RETAIN(target_datatype);
 
-            ret = ompi_osc_pt2pt_isend_w_cb ((void *) packed_ddt, ddt_len, MPI_BYTE, target_rank,
-                                            tag, module->comm, ompi_osc_pt2pt_dt_send_complete,
-                                            target_datatype);
+            ret = ompi_osc_pt2pt_isend_w_cb ((void *) packed_ddt, ddt_len, MPI_BYTE,
+                                            target_rank, tag_to_target(tag), module->comm,
+                                            ompi_osc_pt2pt_dt_send_complete, target_datatype);
             if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
                 break;
             }
@@ -1059,8 +1062,9 @@ int ompi_osc_pt2pt_rget_accumulate_internal (const void *origin_addr, int origin
             ptr += ddt_len;
         }
 
-        ret = ompi_osc_pt2pt_irecv_w_cb (result_addr, result_count, result_datatype, target_rank, tag,
-                                        module->comm, NULL, ompi_osc_pt2pt_req_comm_complete, pt2pt_request);
+        ret = ompi_osc_pt2pt_irecv_w_cb (result_addr, result_count, result_datatype,
+                                        target_rank, tag_to_origin(tag), module->comm,
+                                        NULL, ompi_osc_pt2pt_req_comm_complete, pt2pt_request);
         if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
             break;
         }
@@ -1077,8 +1081,9 @@ int ompi_osc_pt2pt_rget_accumulate_internal (const void *origin_addr, int origin
             header->base.type = OMPI_OSC_PT2PT_HDR_TYPE_GET_ACC_LONG;
             osc_pt2pt_hton(header, proc);
 
-            ret = ompi_osc_pt2pt_isend_w_cb (origin_addr, origin_count, origin_datatype, target_rank,
-                                            tag, module->comm, ompi_osc_pt2pt_req_comm_complete, pt2pt_request);
+            ret = ompi_osc_pt2pt_isend_w_cb (origin_addr, origin_count, origin_datatype,
+                                            target_rank, tag_to_target(tag), module->comm,
+                                            ompi_osc_pt2pt_req_comm_complete, pt2pt_request);
         }
     } while (0);
 

ompi/mca/osc/pt2pt/osc_pt2pt_component.c

@@ -290,7 +290,6 @@ component_select(struct ompi_win_t *win, void **base, size_t size, int disp_unit
     /* fill in the function pointer part */
     memcpy(module, &ompi_osc_pt2pt_module_template,
            sizeof(ompi_osc_base_module_t));
-    module->rtag_counter = 2;
 
     /* initialize the objects, so that always free in cleanup */
     OBJ_CONSTRUCT(&module->lock, opal_mutex_t);

ompi/mca/osc/pt2pt/osc_pt2pt_data_move.c

@@ -14,6 +14,7 @@
  * Copyright (c) 2012-2013 Sandia National Laboratories.  All rights reserved.
  * Copyright (c) 2014-2015 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
+ * Copyright (c) 2016      FUJITSU LIMITED.  All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -398,7 +399,7 @@ static inline int process_put_long(ompi_osc_pt2pt_module_t* module, int source,
     ret = ompi_osc_pt2pt_component_irecv (module, target,
                                          put_header->count,
                                          datatype, source,
-                                         put_header->tag,
+                                         tag_to_target(put_header->tag),
                                          module->comm);
     if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
         OPAL_OUTPUT_VERBOSE((1, ompi_osc_base_framework.framework_output,
@@ -534,7 +535,8 @@ static inline int process_get (ompi_osc_pt2pt_module_t* module, int target,
     }
 
     /* send get data */
-    ret = osc_pt2pt_get_post_send (module, source, get_header->count, datatype, target, get_header->tag);
+    ret = osc_pt2pt_get_post_send (module, source, get_header->count, datatype,
+                                  target, tag_to_origin(get_header->tag));
 
     OBJ_RELEASE(datatype);
 
@@ -847,9 +849,9 @@ static int ompi_osc_pt2pt_acc_long_start (ompi_osc_pt2pt_module_t *module, int s
 
     do {
         if (op == &ompi_mpi_op_replace.op) {
-            ret = ompi_osc_pt2pt_irecv_w_cb (target, acc_header->count, datatype, source,
-                                            acc_header->tag, module->comm, NULL,
-                                            replace_cb, module);
+            ret = ompi_osc_pt2pt_irecv_w_cb (target, acc_header->count, datatype,
+                                            source, tag_to_target(acc_header->tag), module->comm,
+                                            NULL, replace_cb, module);
             break;
         }
 
@@ -876,8 +878,9 @@ static int ompi_osc_pt2pt_acc_long_start (ompi_osc_pt2pt_module_t *module, int s
             break;
         }
 
-        ret = ompi_osc_pt2pt_irecv_w_cb (buffer, primitive_count, primitive_datatype, source,
-                                         acc_header->tag, module->comm, NULL, accumulate_cb, acc_data);
+        ret = ompi_osc_pt2pt_irecv_w_cb (buffer, primitive_count, primitive_datatype,
+                                        source, tag_to_target(acc_header->tag), module->comm,
+                                        NULL, accumulate_cb, acc_data);
         if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
             OBJ_RELEASE(acc_data);
         }
@@ -925,8 +928,9 @@ static int ompi_osc_pt2pt_gacc_start (ompi_osc_pt2pt_module_t *module, int sourc
             break;
         }
 
-        ret = ompi_osc_pt2pt_isend_w_cb (target, acc_header->count, datatype, source, acc_header->tag,
-                                        module->comm, accumulate_cb, acc_data);
+        ret = ompi_osc_pt2pt_isend_w_cb (target, acc_header->count, datatype,
+                                        source, tag_to_origin(acc_header->tag), module->comm,
+                                        accumulate_cb, acc_data);
         if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
             OBJ_RELEASE(acc_data);
         }
@@ -994,15 +998,17 @@ static int ompi_osc_gacc_long_start (ompi_osc_pt2pt_module_t *module, int source
             break;
         }
 
-        ret = ompi_osc_pt2pt_irecv_w_cb (buffer, acc_header->count, datatype, source, acc_header->tag,
-                                        module->comm, &recv_request, accumulate_cb, acc_data);
+        ret = ompi_osc_pt2pt_irecv_w_cb (buffer, acc_header->count, datatype,
+                                        source, tag_to_target(acc_header->tag), module->comm,
+                                        &recv_request, accumulate_cb, acc_data);
         if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
             OBJ_RELEASE(acc_data);
             break;
         }
 
-        ret = ompi_osc_pt2pt_isend_w_cb (target, primitive_count, primitive_datatype, source, acc_header->tag,
-                                        module->comm, accumulate_cb, acc_data);
+        ret = ompi_osc_pt2pt_isend_w_cb (target, primitive_count, primitive_datatype,
+                                        source, tag_to_origin(acc_header->tag), module->comm,
+                                        accumulate_cb, acc_data);
         if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
             /* cancel the receive and free the accumulate data */
             ompi_request_cancel (recv_request);
@@ -1054,8 +1060,8 @@ static int ompi_osc_pt2pt_cswap_start (ompi_osc_pt2pt_module_t *module, int sour
 
     do {
         /* no reason to do a non-blocking send here */
-        ret = MCA_PML_CALL(send(target, 1, datatype, source, cswap_header->tag, MCA_PML_BASE_SEND_STANDARD,
-                                module->comm));
+        ret = MCA_PML_CALL(send(target, 1, datatype, source, tag_to_origin(cswap_header->tag),
+                                MCA_PML_BASE_SEND_STANDARD, module->comm));
         if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
             break;
         }
@@ -1503,8 +1509,9 @@ static int process_large_datatype_request (ompi_osc_pt2pt_module_t *module, int
     memcpy (ddt_buffer->header, header, header_len);
 
     ret = ompi_osc_pt2pt_irecv_w_cb ((void *)((uintptr_t) ddt_buffer->header + header_len),
-                                    ddt_len, MPI_BYTE, source, tag, module->comm, NULL,
-                                    process_large_datatype_request_cb, ddt_buffer);
+                                    ddt_len, MPI_BYTE,
+                                    source, tag_to_target(tag), module->comm,
+                                    NULL, process_large_datatype_request_cb, ddt_buffer);
     if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
         OBJ_RELEASE(ddt_buffer);
         return ret;