comm: add facility to store persistent requests in hash

Maintain a per-comm hash for persistent requests so we can check and
free them at MPI_Comm_disconnect or MPI_Finalize if user forget to free
them.

We need apply per-vci critical sections to prevent data race.

Reference: mpi-forum/mpi-issues#710.

src/include/mpir_comm.h

@@ -258,6 +258,8 @@ struct MPIR_Comm {
         } multiplex;
     } stream_comm;
 
+    MPIR_Request *persistent_requests;
+
     /* Other, device-specific information */
 #ifdef MPID_DEV_COMM_DECL
      MPID_DEV_COMM_DECL
@@ -369,6 +371,10 @@ int MPIR_Comm_split_filesystem(MPI_Comm comm, int key, const char *dirname, MPI_
 #define MPIR_Comm_rank(comm_ptr) ((comm_ptr)->rank)
 #define MPIR_Comm_size(comm_ptr) ((comm_ptr)->local_size)
 
+int MPIR_Comm_save_inactive_request(MPIR_Comm * comm, MPIR_Request * request);
+int MPIR_Comm_delete_inactive_request(MPIR_Comm * comm, MPIR_Request * request);
+int MPIR_Comm_free_inactive_requests(MPIR_Comm * comm);
+
 /* Comm hint registration.
  *
  * Hint function is optional. If it is NULL, MPIR_layer will set corresponding

src/include/mpir_request.h

@@ -253,6 +253,7 @@ struct MPIR_Request {
 #endif                          /* HAVE_DEBUGGER_SUPPORT */
 
     struct MPIR_Request *next, *prev;
+    UT_hash_handle hh;
 
     /* Other, device-specific information */
 #ifdef MPID_DEV_REQUEST_DECL

src/mpi/comm/commutil.c

@@ -302,6 +302,8 @@ int MPII_Comm_init(MPIR_Comm * comm_p)
     comm_p->threadcomm = NULL;
     MPIR_stream_comm_init(comm_p);
 
+    comm_p->persistent_requests = NULL;
+
     /* mutex is only used in VCI granularity. But the overhead of
      * creation is low, so we always create it. */
     {
@@ -1338,3 +1340,37 @@ int MPII_Comm_is_node_balanced(MPIR_Comm * comm, int *num_nodes, bool * node_bal
   fn_fail:
     goto fn_exit;
 }
+
+int MPIR_Comm_save_inactive_request(MPIR_Comm * comm, MPIR_Request * request)
+{
+    MPID_THREAD_CS_ENTER(VCI, comm->mutex);
+    HASH_ADD_INT(comm->persistent_requests, handle, request, MPL_MEM_COMM);
+    MPID_THREAD_CS_EXIT(VCI, comm->mutex);
+
+    return MPI_SUCCESS;
+}
+
+int MPIR_Comm_delete_inactive_request(MPIR_Comm * comm, MPIR_Request * request)
+{
+    MPID_THREAD_CS_ENTER(VCI, comm->mutex);
+    HASH_DEL(comm->persistent_requests, request);
+    MPID_THREAD_CS_EXIT(VCI, comm->mutex);
+
+    return MPI_SUCCESS;
+}
+
+int MPIR_Comm_free_inactive_requests(MPIR_Comm * comm)
+{
+    MPIR_Request *request, *tmp;
+    MPID_THREAD_CS_ENTER(VCI, comm->mutex);
+    HASH_ITER(hh, comm->persistent_requests, request, tmp) {
+        if (!MPIR_Request_is_active(request)) {
+            printf("WARNING: freeing inactive persistent request %x on communicator %x.\n",
+                   request->handle, comm->handle);
+            MPIR_Request_free_impl(request);
+        }
+    }
+    MPID_THREAD_CS_EXIT(VCI, comm->mutex);
+
+    return MPI_SUCCESS;
+}