Control no. of active RMA requests in the runtime.

When there are too many active requests in the runtime,
the internal memory might be used up. This patch
prevents such situation by triggering blocking
wait loop in operation routines when no. of active
requests reaches certain threshold value.

Signed-off-by: Pavan Balaji <balaji@anl.gov>

src/mpid/ch3/include/mpid_rma_oplist.h

@@ -375,6 +375,8 @@ static inline int MPIDI_CH3I_RMA_Cleanup_ops_target(MPID_Win * win_ptr, MPIDI_RM
             /* dequeue the operation and free it */
             MPL_LL_DELETE(*op_list, *op_list_tail, curr_op);
             MPIDI_CH3I_Win_op_free(win_ptr, curr_op);
+            win_ptr->active_req_cnt--;
+
             if (*op_list == NULL) {
                 if (read_flag == 1) {
                     read_flag = 0;

src/mpid/ch3/include/mpidpre.h

@@ -380,6 +380,9 @@ struct MPIDI_Win_target_state {
                            in current epoch (accumulated value, not      \
                            current value) to control when to poke        \
                            progress engine in RMA operation routines. */ \
+    int active_req_cnt; /* keep track of number of active requests in    \
+                           current epoch, i.e., number of issued but     \
+                           incomplete RMA operations. */                 \
 
 #ifdef MPIDI_CH3_WIN_DECL
 #define MPID_DEV_WIN_DECL \

src/mpid/ch3/src/ch3u_rma_oplist.c

@@ -21,6 +21,24 @@ cvars:
       description : >-
         Use the immediate accumulate optimization
 
+    - name        : MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD
+      category    : CH3
+      type        : int
+      default     : 2097152
+      class       : none
+      verbosity   : MPI_T_VERBOSITY_USER_BASIC
+      scope       : MPI_T_SCOPE_ALL_EQ
+      description : >-
+         Threshold of number of active requests to trigger
+         blocking waiting in operation routines. When the
+         value is negative, we never blockingly wait in
+         operation routines. When the value is zero, we always
+         trigger blocking waiting in operation routines to
+         wait until no. of active requests becomes zero. When the
+         value is positive, we do blocking waiting in operation
+         routines to wait until no. of active requests being
+         reduced to this value.
+
 === END_MPI_T_CVAR_INFO_BLOCK ===
 */
 
@@ -356,6 +374,7 @@ static inline int issue_ops_target(MPID_Win * win_ptr, MPIDI_RMA_Target_t *targe
                 MPIDI_CH3I_RMA_Ops_append(&(target->read_op_list),
                                           &(target->read_op_list_tail), curr_op);
             }
+            win_ptr->active_req_cnt++;
         }
 
         curr_op = target->next_op_to_issue;

src/mpid/ch3/src/ch3u_rma_ops.c

@@ -143,6 +143,20 @@ int MPIDI_Put(const void *origin_addr, int origin_count, MPI_Datatype
             if (mpi_errno != MPI_SUCCESS)
                 MPIU_ERR_POP(mpi_errno);
         }
+
+        if (MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD >= 0 &&
+            win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+            while (win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+                int local_completed = 0, remote_completed = 0;
+                mpi_errno = MPIDI_CH3I_RMA_Cleanup_ops_win(win_ptr, &local_completed,
+                                                           &remote_completed);
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+                mpi_errno = poke_progress_engine();
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+            }
+        }
     }
 
   fn_exit:
@@ -265,6 +279,20 @@ int MPIDI_Get(void *origin_addr, int origin_count, MPI_Datatype
             if (mpi_errno != MPI_SUCCESS)
                 MPIU_ERR_POP(mpi_errno);
         }
+
+        if (MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD >= 0 &&
+            win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+            while (win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+                int local_completed = 0, remote_completed = 0;
+                mpi_errno = MPIDI_CH3I_RMA_Cleanup_ops_win(win_ptr, &local_completed,
+                                                           &remote_completed);
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+                mpi_errno = poke_progress_engine();
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+            }
+        }
     }
 
   fn_exit:
@@ -424,6 +452,20 @@ int MPIDI_Accumulate(const void *origin_addr, int origin_count, MPI_Datatype
             if (mpi_errno != MPI_SUCCESS)
                 MPIU_ERR_POP(mpi_errno);
         }
+
+        if (MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD >= 0 &&
+            win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+            while (win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+                int local_completed = 0, remote_completed = 0;
+                mpi_errno = MPIDI_CH3I_RMA_Cleanup_ops_win(win_ptr, &local_completed,
+                                                           &remote_completed);
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+                mpi_errno = poke_progress_engine();
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+            }
+        }
     }
 
   fn_exit:
@@ -579,6 +621,20 @@ int MPIDI_Get_accumulate(const void *origin_addr, int origin_count,
             if (mpi_errno != MPI_SUCCESS)
                 MPIU_ERR_POP(mpi_errno);
         }
+
+        if (MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD >= 0 &&
+            win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+            while (win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+                int local_completed = 0, remote_completed = 0;
+                mpi_errno = MPIDI_CH3I_RMA_Cleanup_ops_win(win_ptr, &local_completed,
+                                                           &remote_completed);
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+                mpi_errno = poke_progress_engine();
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+            }
+        }
     }
 
   fn_exit:
@@ -685,6 +741,20 @@ int MPIDI_Compare_and_swap(const void *origin_addr, const void *compare_addr,
             if (mpi_errno != MPI_SUCCESS)
                 MPIU_ERR_POP(mpi_errno);
         }
+
+        if (MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD >= 0 &&
+            win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+            while (win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+                int local_completed = 0, remote_completed = 0;
+                mpi_errno = MPIDI_CH3I_RMA_Cleanup_ops_win(win_ptr, &local_completed,
+                                                           &remote_completed);
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+                mpi_errno = poke_progress_engine();
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+            }
+        }
     }
 
   fn_exit:
@@ -788,6 +858,20 @@ int MPIDI_Fetch_and_op(const void *origin_addr, void *result_addr,
             if (mpi_errno != MPI_SUCCESS)
                 MPIU_ERR_POP(mpi_errno);
         }
+
+        if (MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD >= 0 &&
+            win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+            while (win_ptr->active_req_cnt >= MPIR_CVAR_CH3_RMA_ACTIVE_REQ_THRESHOLD) {
+                int local_completed = 0, remote_completed = 0;
+                mpi_errno = MPIDI_CH3I_RMA_Cleanup_ops_win(win_ptr, &local_completed,
+                                                           &remote_completed);
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+                mpi_errno = poke_progress_engine();
+                if (mpi_errno != MPI_SUCCESS)
+                    MPIU_ERR_POP(mpi_errno);
+            }
+        }
     }
 
   fn_exit:

src/mpid/ch3/src/mpid_rma.c

@@ -333,6 +333,7 @@ static int win_init(MPI_Aint size, int disp_unit, int create_flavor, int model,
     (*win_ptr)->states.exposure_state = MPIDI_RMA_NONE;
     (*win_ptr)->non_empty_slots = 0;
     (*win_ptr)->posted_ops_cnt = 0;
+    (*win_ptr)->active_req_cnt = 0;
 
     /* Initialize the passive target lock state */
     MPIU_CHKPMEM_MALLOC((*win_ptr)->targets, struct MPIDI_Win_target_state *,