SST,MPI,DP: soft handle peer error

source/adios2/toolkit/sst/dp/mpi_dp.c

@@ -846,6 +846,32 @@ static void MpiDestroyWriterPerReader(CP_Services Svcs, DP_WSR_Stream WSR_Stream
 
     const int CohortSize = StreamWPR->Link.CohortSize;
 
+    printf("WPR soft Destroy called %d (rank: %d)\n", CohortSize, StreamWR->Stream.Rank);
+
+    for (int i = 0; i < CohortSize; i++)
+    {
+        if (StreamWPR->CohortMpiComms[i] != MPI_COMM_NULL)
+        {
+            MPI_Comm_disconnect(&StreamWPR->CohortMpiComms[i]);
+        }
+    }
+}
+
+/**
+ * MpiDestroyWriterPerReader.
+ *
+ * This is called whenever a reader disconnect from a writer. This function
+ * also removes the StreamWPR from its own StreamWR.
+ */
+static void MpiFullyDestroyWriterPerReader(CP_Services Svcs, DP_WSR_Stream WSR_Stream_v)
+{
+    MpiStreamWPR StreamWPR = (MpiStreamWPR)WSR_Stream_v;
+    MpiStreamWR StreamWR = StreamWPR->StreamWR;
+
+    const int CohortSize = StreamWPR->Link.CohortSize;
+
+    printf("WPR Destroy called %d (rank: %d)\n", CohortSize, StreamWR->Stream.Rank);
+
     for (int i = 0; i < CohortSize; i++)
     {
         if (StreamWPR->CohortMpiComms[i] != MPI_COMM_NULL)
@@ -875,7 +901,7 @@ static void MpiDestroyWriter(CP_Services Svcs, DP_WS_Stream WS_Stream_v)
     while (!TAILQ_EMPTY(&StreamWR->Readers))
     {
         MpiStreamWPR Stream = TAILQ_FIRST(&StreamWR->Readers);
-        MpiDestroyWriterPerReader(Svcs, Stream);
+        MpiFullyDestroyWriterPerReader(Svcs, Stream);
     }
     pthread_mutex_unlock(&StreamWR->MutexReaders);