Add data structure to track ACK's

This commit was SVN r9177.

ompi/mca/pml/dr/pml_dr_comm.c

@@ -32,6 +32,7 @@ static void mca_pml_dr_comm_proc_construct(mca_pml_dr_comm_proc_t* proc)
     OBJ_CONSTRUCT(&proc->frags_cant_match, opal_list_t);
     OBJ_CONSTRUCT(&proc->specific_receives, opal_list_t);
     OBJ_CONSTRUCT(&proc->unexpected_frags, opal_list_t);
+    OBJ_CONSTRUCT(&proc->acked_vfrags, opal_list_t);
 }
 
 

ompi/mca/pml/dr/pml_dr_comm.h

@@ -44,9 +44,19 @@ struct mca_pml_dr_comm_proc_t {
     opal_list_t specific_receives; /**< queues of unmatched specific receives */
     opal_list_t unexpected_frags;  /**< unexpected fragment queues */
     ompi_proc_t* ompi_proc;        /**< back pointer to ompi_proc_t */
+    opal_list_t acked_vfrags;      /**< list of vfrags id's that have been acked */
+    opal_list_item_t* acked_vfrags_ptr; /**< a pointer to the last place we were in the list */
 };
 typedef struct mca_pml_dr_comm_proc_t mca_pml_dr_comm_proc_t;
 
+struct mca_pml_dr_acked_item_t{ 
+    opal_list_item_t super; 
+    int32_t vfrag_id_high;
+    int32_t vfrag_id_low;
+}; 
+typedef struct mca_pml_dr_acked_item_t mca_pml_dr_acked_item_t;
+
+OMPI_DECLSPEC OBJ_CLASS_DECLARATION(mca_pml_dr_acked_item_t);
 
 /**
  *  Cached on ompi_communicator_t to hold queues/state
@@ -79,6 +89,125 @@ OMPI_DECLSPEC OBJ_CLASS_DECLARATION(mca_pml_dr_comm_t);
 
 OMPI_DECLSPEC extern int mca_pml_dr_comm_init(mca_pml_dr_comm_t* dr_comm, ompi_communicator_t* ompi_comm);
 
+static inline bool mac_pml_dr_comm_proc_check_acked(mca_pml_dr_comm_proc_t* proc, int32_t vfrag_id) { 
+    mca_pml_dr_acked_item_t* item = (mca_pml_dr_acked_item_t*) proc->acked_vfrags_ptr;
+    int8_t direction = 0; /* 1 is next, -1 is previous */
+    while(true) { 
+        if(NULL == item) { 
+            return false;
+        } else if(item->vfrag_id_high >= vfrag_id && item->vfrag_id_low <= vfrag_id) { 
+            proc->acked_vfrags_ptr = (opal_list_item_t*) item;
+            return true; 
+        } else if(vfrag_id > item->vfrag_id_high && direction != -1) { 
+            direction = 1; 
+            item = (mca_pml_dr_acked_item_t*) opal_list_get_next(item); 
+        } else if(vfrag_id < item->vfrag_id_low && direction != 1) { 
+            direction = -1;
+            item = (mca_pml_dr_acked_item_t*) opal_list_get_prev(item); 
+        } else { 
+            return false;
+        }
+    }
+}
+
+static inline void mac_pml_dr_comm_proc_set_acked(mca_pml_dr_comm_proc_t* proc, int32_t vfrag_id) { 
+    mca_pml_dr_acked_item_t* item = (mca_pml_dr_acked_item_t*) proc->acked_vfrags_ptr;
+    int8_t direction = 0; /* 1 is next, -1 is previous */
+    mca_pml_dr_acked_item_t *new_item, *next_item, *prev_item;
+    while(true) { 
+        if(NULL == item && direction == 0) { 
+
+            new_item = OBJ_NEW(mca_pml_dr_acked_item_t);
+            new_item->vfrag_id_low = new_item->vfrag_id_high = vfrag_id; 
+            opal_list_append(&proc->acked_vfrags, (opal_list_item_t*) item);
+            proc->acked_vfrags_ptr = (opal_list_item_t*) item;
+            return;
+        
+        } else if (NULL == item && direction == 1) { 
+
+            opal_list_append(&proc->acked_vfrags, (opal_list_item_t*) item);
+            proc->acked_vfrags_ptr = (opal_list_item_t*) item;
+            return;
+            
+        } else if (NULL == item && direction == -1) { 
+
+            opal_list_prepend(&proc->acked_vfrags, (opal_list_item_t*) item); 
+            proc->acked_vfrags_ptr = (opal_list_item_t*) item;
+            return; 
+            
+        } else if(item->vfrag_id_high >= vfrag_id && item->vfrag_id_low <= vfrag_id ) { 
+
+            proc->acked_vfrags_ptr = (opal_list_item_t*) item;
+            return; 
+
+        } else if((item->vfrag_id_high + 1) == vfrag_id) { 
+            
+            next_item = (mca_pml_dr_acked_item_t*) opal_list_get_next(item); 
+            /* try to consolidate */ 
+            if(next_item && next_item->vfrag_id_low == (vfrag_id+1)) { 
+                item->vfrag_id_high = next_item->vfrag_id_high;
+                opal_list_remove_item(&proc->acked_vfrags, (opal_list_item_t*) next_item);
+                OBJ_RELEASE(next_item);
+            } else { 
+                item->vfrag_id_high = vfrag_id;
+            }    
+            proc->acked_vfrags_ptr = (opal_list_item_t*) item;
+            return; 
+            
+        } else if((item->vfrag_id_low - 1) == vfrag_id) { 
+            
+            prev_item = (mca_pml_dr_acked_item_t*) opal_list_get_prev(item);
+            /* try to consolidate */
+            if(prev_item && prev_item->vfrag_id_high == (vfrag_id-1)) {  
+                item->vfrag_id_low = prev_item->vfrag_id_low;
+                opal_list_remove_item(&proc->acked_vfrags, (opal_list_item_t*) prev_item);
+                OBJ_RELEASE(prev_item);
+            } else { 
+                item->vfrag_id_low = vfrag_id; 
+            }
+            proc->acked_vfrags_ptr = (opal_list_item_t*) item;
+            return; 
+            
+        } else if(vfrag_id > item->vfrag_id_high ) { 
+            if(direction == -1) { 
+                /* we have gone back in the list, and we went one item too far */ 
+                new_item = OBJ_NEW(mca_pml_dr_acked_item_t);
+                new_item->vfrag_id_low = new_item->vfrag_id_high = vfrag_id; 
+                /* insert new_item directly before item */
+                opal_list_insert_pos(&proc->acked_vfrags, 
+                                     (opal_list_item_t*) item, 
+                                     (opal_list_item_t*) new_item); 
+                proc->acked_vfrags_ptr = (opal_list_item_t*) item;
+                return;
+            } else { 
+                direction = 1;
+                item = (mca_pml_dr_acked_item_t*) opal_list_get_next(item);
+            }
+        } else if(vfrag_id < item->vfrag_id_low) { 
+            if(direction == 1) { 
+                /* we have gone forward in the list, and we went one item too far */ 
+                new_item = OBJ_NEW(mca_pml_dr_acked_item_t); 
+                next_item = (mca_pml_dr_acked_item_t*) opal_list_get_next(item);
+                if(NULL == next_item) { 
+                    opal_list_append(&proc->acked_vfrags, (opal_list_item_t*) new_item);
+                } else { 
+                    opal_list_insert_pos(&proc->acked_vfrags, 
+                                         (opal_list_item_t*) next_item, 
+                                         (opal_list_item_t*) new_item); 
+                }
+                proc->acked_vfrags_ptr = (opal_list_item_t*) item;
+                return;
+            } else { 
+                direction = -1;
+                item = (mca_pml_dr_acked_item_t*) opal_list_get_prev(item); 
+            }
+        } else { 
+            return;
+        }
+    }
+}
+
+
 #if defined(c_plusplus) || defined(__cplusplus)
 }
 #endif