Add support for dynamic allow_overtake

This support only covers the case when allow_overtake is enabled. For
the opposite there is no simple solution, as described in the comment
in pml_ob1.c

Signed-off-by: George Bosilca <[email protected]>

Author: bosilca

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_recvfrag.c

@@ -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)

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);