Merge pull request open-mpi#9843 from devreal/allow-overtake-snafu

Fix issues with mpi_assert_allow_overtaking

Author: devreal

ompi/mca/pml/ob1/pml_ob1.c

@@ -3,7 +3,7 @@
  * Copyright (c) 2004-2010 The Trustees of Indiana University and Indiana
  *                         University Research and Technology
  *                         Corporation.  All rights reserved.
- * Copyright (c) 2004-2020 The University of Tennessee and The University
+ * Copyright (c) 2004-2022 The University of Tennessee and The University
  *                         of Tennessee Research Foundation.  All rights
  *                         reserved.
  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
@@ -202,6 +202,43 @@ int mca_pml_ob1_enable(bool enable)
     return OMPI_SUCCESS;
 }
 
+static const char*
+mca_pml_ob1_set_allow_overtake(opal_infosubscriber_t* obj,
+                               const char* key,
+                               const char* value)
+{
+    ompi_communicator_t *ompi_comm = (ompi_communicator_t *) obj;
+    bool allow_overtake_was_set = OMPI_COMM_CHECK_ASSERT_ALLOW_OVERTAKE(ompi_comm);
+
+    /* As we keep the out-of-sequence messages ordered by their sequence, as a receiver we
+     * can just move the previously considered out-of-order messages into the unexpected queue,
+     * and we maintain some form of logical consistency with the message order.
+     */
+    if (opal_str_to_bool(value)) {
+        if (!allow_overtake_was_set) {
+            ompi_comm->c_flags |= OMPI_COMM_ASSERT_ALLOW_OVERTAKE;
+            mca_pml_ob1_merge_cant_match(ompi_comm);
+        }
+        return "true";
+    }
+    if (allow_overtake_was_set) {
+        /* However, in the case we are trying to turn off allow_overtake, it is not clear what
+         * should be done with the previous messages that are pending on our peers, nor with
+         * the messages currently in the network. Similarly, if one process turns off allow
+         * overtake, before any potential sender start sending valid sequence numbers there
+         * is no way to order the messages in a sensible order.
+         * The possible solution is cumbersome, it would force a network quiescence followed by
+         * a synchronization of all processes in the communicator, and then all peers will
+         * start sending messages starting with sequence number 0.
+         * A lot of code for minimal benefit, especially taking in account that the MPI standard
+         * does not define this. Instead, refuse to disable allow overtake, and at least the
+         * user has the opportunity to check if we accepted to change it.
+         */
+        return "true";
+    }
+    return "false";
+}
+
 int mca_pml_ob1_add_comm(ompi_communicator_t* comm)
 {
     /* allocate pml specific comm data */
@@ -221,11 +258,14 @@ int mca_pml_ob1_add_comm(ompi_communicator_t* comm)
     }
 
     ompi_comm_assert_subscribe (comm, OMPI_COMM_ASSERT_NO_ANY_SOURCE);
-    ompi_comm_assert_subscribe (comm, OMPI_COMM_ASSERT_ALLOW_OVERTAKE);
 
     mca_pml_ob1_comm_init_size(pml_comm, comm->c_remote_group->grp_proc_count);
     comm->c_pml_comm = pml_comm;
 
+    /* Register the subscriber alert for the mpi_assert_allow_overtaking info. */
+    opal_infosubscribe_subscribe (&comm->super, "mpi_assert_allow_overtaking",
+                                  "false", mca_pml_ob1_set_allow_overtake);
+
     /* Grab all related messages from the non_existing_communicator pending queue */
     OPAL_LIST_FOREACH_SAFE(frag, next_frag, &mca_pml_ob1.non_existing_communicator_pending, mca_pml_ob1_recv_frag_t) {
         hdr = &frag->hdr.hdr_match;

ompi/mca/pml/ob1/pml_ob1_isend.c

@@ -3,7 +3,7 @@
  * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
  *                         University Research and Technology
  *                         Corporation.  All rights reserved.
- * Copyright (c) 2004-2018 The University of Tennessee and The University
+ * Copyright (c) 2004-2022 The University of Tennessee and The University
  *                         of Tennessee Research Foundation.  All rights
  *                         reserved.
  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
@@ -170,7 +170,7 @@ int mca_pml_ob1_isend(const void *buf,
         return OMPI_ERR_UNREACH;
     }
 
-    if (!OMPI_COMM_CHECK_ASSERT_ALLOW_OVERTAKE(comm)) {
+    if (!OMPI_COMM_CHECK_ASSERT_ALLOW_OVERTAKE(comm) || 0 > tag) {
         seqn = (uint16_t) OPAL_THREAD_ADD_FETCH32(&ob1_proc->send_sequence, 1);
     }
 
@@ -274,7 +274,7 @@ int mca_pml_ob1_send(const void *buf,
         return OMPI_SUCCESS;
     }
 
-    if (!OMPI_COMM_CHECK_ASSERT_ALLOW_OVERTAKE(comm)) {
+    if (!OMPI_COMM_CHECK_ASSERT_ALLOW_OVERTAKE(comm) || 0 > tag) {
         seqn = (uint16_t) OPAL_THREAD_ADD_FETCH32(&ob1_proc->send_sequence, 1);
     }
 

ompi/mca/pml/ob1/pml_ob1_recvfrag.c

@@ -3,7 +3,7 @@
  * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
  *                         University Research and Technology
  *                         Corporation.  All rights reserved.
- * Copyright (c) 2004-2019 The University of Tennessee and The University
+ * Copyright (c) 2004-2022 The University of Tennessee and The University
  *                         of Tennessee Research Foundation.  All rights
  *                         reserved.
  * Copyright (c) 2004-2007 High Performance Computing Center Stuttgart,
@@ -370,7 +370,7 @@ int mca_pml_ob1_revoke_comm( struct ompi_communicator_t* ompi_comm, bool coll_on
         /* note this is not an ompi_proc, but a ob1_comm_proc, thus we don't
          * use ompi_proc_is_sentinel to verify if initialized. */
         if( NULL == proc ) continue;
-        /* remove the frag from the unexpected list, add to the nack list 
+        /* remove the frag from the unexpected list, add to the nack list
          * so that we can send the nack as needed to remote cancel the send
          * from outside the match lock.
          */
@@ -385,7 +385,7 @@ int mca_pml_ob1_revoke_comm( struct ompi_communicator_t* ompi_comm, bool coll_on
             }
         }
         /* same for the cantmatch queue/heap; this list is more complicated
-         * Keep it simple: we pop all of the complex list, put the bad items 
+         * Keep it simple: we pop all of the complex list, put the bad items
          * in the nack_list, and keep the good items in the keep_list;
          * then we reinsert the good items in the cantmatch heaplist */
         mca_pml_ob1_recv_frag_t* frag;
@@ -519,7 +519,7 @@ void mca_pml_ob1_recv_frag_callback_match (mca_btl_base_module_t *btl,
     }
 #endif
 
-    if (!OMPI_COMM_CHECK_ASSERT_ALLOW_OVERTAKE(comm_ptr)) {
+    if (!OMPI_COMM_CHECK_ASSERT_ALLOW_OVERTAKE(comm_ptr) || 0 > hdr->hdr_tag) {
         /* get sequence number of next message that can be processed.
          * If this frag is out of sequence, queue it up in the list
          * now as we still have the lock.
@@ -637,6 +637,38 @@ void mca_pml_ob1_recv_frag_callback_match (mca_btl_base_module_t *btl,
     }
 }
 
+/**
+ * Merge all out of sequence fragments into the matching queue, as if they were received now.
+ */
+int mca_pml_ob1_merge_cant_match( ompi_communicator_t * ompi_comm )
+{
+    mca_pml_ob1_comm_t * pml_comm = (mca_pml_ob1_comm_t *)ompi_comm->c_pml_comm;
+    mca_pml_ob1_recv_frag_t *frag, *frags_cant_match;
+    mca_pml_ob1_comm_proc_t* proc;
+    int cnt = 0;
+
+    for (uint32_t i = 0; i < pml_comm->num_procs; i++) {
+        if ((NULL == (proc = pml_comm->procs[i])) || (NULL != proc->frags_cant_match)) {
+            continue;
+        }
+
+        OB1_MATCHING_LOCK(&pml_comm->matching_lock);
+        /* Acquire all cant_match frags from the peer */
+        frags_cant_match = proc->frags_cant_match;
+        proc->frags_cant_match = NULL;
+        while(NULL != (frag = remove_head_from_ordered_list(&frags_cant_match))) {
+            /* mca_pml_ob1_recv_frag_match_proc() will release the lock. */
+            mca_pml_ob1_recv_frag_match_proc(frag->btl, ompi_comm, proc,
+                                             &frag->hdr.hdr_match,
+                                             frag->segments, frag->num_segments,
+                                             frag->hdr.hdr_match.hdr_common.hdr_type, frag);
+            OB1_MATCHING_LOCK(&pml_comm->matching_lock);
+            cnt++;
+        }
+    }
+    OB1_MATCHING_UNLOCK(&pml_comm->matching_lock);
+    return cnt;
+}
 
 void mca_pml_ob1_recv_frag_callback_rndv (mca_btl_base_module_t *btl,
                                           const mca_btl_base_receive_descriptor_t *descriptor)
@@ -1092,7 +1124,7 @@ static int mca_pml_ob1_recv_frag_match (mca_btl_base_module_t *btl,
     frag_msg_seq = hdr->hdr_seq;
     next_msg_seq_expected = (uint16_t)proc->expected_sequence;
 
-    if (!OMPI_COMM_CHECK_ASSERT_ALLOW_OVERTAKE(comm_ptr)) {
+    if (!OMPI_COMM_CHECK_ASSERT_ALLOW_OVERTAKE(comm_ptr) || 0 > hdr->hdr_tag) {
         /* If the sequence number is wrong, queue it up for later. */
         if(OPAL_UNLIKELY(frag_msg_seq != next_msg_seq_expected)) {
             mca_pml_ob1_recv_frag_t* frag;

ompi/mca/pml/ob1/pml_ob1_recvfrag.h

@@ -172,6 +172,12 @@ extern void mca_pml_ob1_recv_frag_callback_cid( mca_btl_base_module_t *btl,
 extern mca_pml_ob1_recv_frag_t*
 check_cantmatch_for_match(mca_pml_ob1_comm_proc_t *proc);
 
+/**
+ * Move for all peers all pending cant_match fragments into the matching queues. This
+ * function is necessary when allow_overtake info key is transition to set.
+ */
+int mca_pml_ob1_merge_cant_match( ompi_communicator_t * ompi_comm );
+
 void append_frag_to_ordered_list(mca_pml_ob1_recv_frag_t** queue,
                                  mca_pml_ob1_recv_frag_t* frag,
                                  uint16_t seq);