ka-ching

This commit was SVN r2829.

src/mca/gpr/replica/gpr_replica_internals.c

@@ -21,8 +21,6 @@
 
 #include "include/constants.h"
 
-#include "threads/mutex.h"
-
 #include "util/output.h"
 #include "util/printf.h"
 #include "util/proc_info.h"
@@ -42,11 +40,9 @@ mca_gpr_replica_segment_t *gpr_replica_define_segment(char *segment)
     mca_gpr_replica_segment_t *seg;
     mca_gpr_replica_key_t key;
 
-    /* OMPI_THREAD_LOCK(&mca_gpr_replica_internals_mutex); */
 
     key = gpr_replica_define_key(segment, NULL);
     if (MCA_GPR_REPLICA_KEY_MAX == key) {  /* got some kind of error code */
-	/* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
  	return NULL;
     }
 
@@ -55,7 +51,6 @@ mca_gpr_replica_segment_t *gpr_replica_define_segment(char *segment)
     seg->segment = key;
     ompi_list_append(&mca_gpr_replica_head.registry, &seg->item);
 
-    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 
     return seg;
 }
@@ -66,7 +61,6 @@ mca_gpr_replica_segment_t *gpr_replica_find_seg(bool create, char *segment)
     mca_gpr_replica_keytable_t *ptr_seg;
     mca_gpr_replica_segment_t *seg;
 
-    /* OMPI_THREAD_LOCK(&mca_gpr_replica_internals_mutex); */
 
     /* search the registry segments to find which one is being referenced */
     for (ptr_seg = (mca_gpr_replica_keytable_t*)ompi_list_get_first(&mca_gpr_replica_head.segment_dict);
@@ -78,14 +72,12 @@ mca_gpr_replica_segment_t *gpr_replica_find_seg(bool create, char *segment)
 		 seg != (mca_gpr_replica_segment_t*)ompi_list_get_end(&mca_gpr_replica_head.registry);
 		 seg = (mca_gpr_replica_segment_t*)ompi_list_get_next(seg)) {
 		if(seg->segment == ptr_seg->key) {
-		    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 		    return(seg);
 		}
 	    }
 	}
     }
 
-    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 
     if (create) {
 	/* didn't find the dictionary entry - create it */
@@ -100,7 +92,6 @@ mca_gpr_replica_keytable_t *gpr_replica_find_dict_entry(char *segment, char *tok
     mca_gpr_replica_keytable_t *ptr_key;
     mca_gpr_replica_segment_t *seg;
 
-    /* OMPI_THREAD_LOCK(&mca_gpr_replica_internals_mutex); */
 
     /* search the registry segments to find which one is being referenced */
     for (ptr_seg = (mca_gpr_replica_keytable_t*)ompi_list_get_first(&mca_gpr_replica_head.segment_dict);
@@ -108,7 +99,6 @@ mca_gpr_replica_keytable_t *gpr_replica_find_dict_entry(char *segment, char *tok
 	 ptr_seg = (mca_gpr_replica_keytable_t*)ompi_list_get_next(ptr_seg)) {
 	if (0 == strcmp(segment, ptr_seg->token)) {
 	    if (NULL == token) { /* just want segment token-key pair */
-		/* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 		return(ptr_seg);
 	    }
 	    /* search registry to find segment */
@@ -121,19 +111,15 @@ mca_gpr_replica_keytable_t *gpr_replica_find_dict_entry(char *segment, char *tok
 			 ptr_key != (mca_gpr_replica_keytable_t*)ompi_list_get_end(&seg->keytable);
 			 ptr_key = (mca_gpr_replica_keytable_t*)ompi_list_get_next(ptr_key)) {
 			if (0 == strcmp(token, ptr_key->token)) {
-			    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 			    return(ptr_key);
 			}
 		    }
-		    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 		    return(NULL); /* couldn't find the specified entry */
 		}
 	    }
-	    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 	    return(NULL); /* couldn't find segment, even though we found entry in registry dict */
 	}
     }
-    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
     return(NULL); /* couldn't find segment token-key pair */
 }
 
@@ -174,7 +160,6 @@ char *gpr_replica_get_token(char *segment, mca_gpr_replica_key_t key)
 	return NULL;
     }
 
-    /* OMPI_THREAD_LOCK(&mca_gpr_replica_internals_mutex); */
 
     /* find the matching key */
     for (ptr_key = (mca_gpr_replica_keytable_t*)ompi_list_get_first(&seg->keytable);
@@ -182,11 +167,9 @@ char *gpr_replica_get_token(char *segment, mca_gpr_replica_key_t key)
 	 ptr_key = (mca_gpr_replica_keytable_t*)ompi_list_get_next(ptr_key)) {
 	if (key == ptr_key->key) {
 	    answer = strdup(ptr_key->token);
-	    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 	    return answer;
 	}
     }
-    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
     return(NULL); /* couldn't find the specified entry */
 }
 
@@ -227,7 +210,6 @@ mca_gpr_replica_key_t gpr_replica_define_key(char *segment, char *token)
     /* if token is NULL, then this is defining a segment name. Check dictionary to ensure uniqueness */
     if (NULL == token) {
 
-	/* OMPI_THREAD_LOCK(&mca_gpr_replica_internals_mutex); */
 
 	for (ptr_seg = (mca_gpr_replica_keytable_t*)ompi_list_get_first(&mca_gpr_replica_head.segment_dict);
 	     ptr_seg != (mca_gpr_replica_keytable_t*)ompi_list_get_end(&mca_gpr_replica_head.segment_dict);
@@ -245,7 +227,6 @@ mca_gpr_replica_key_t gpr_replica_define_key(char *segment, char *token)
 		mca_gpr_replica_head.lastkey++;
 		new->key = mca_gpr_replica_head.lastkey;
 	    } else {  /* out of keys */
-		/* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 		return MCA_GPR_REPLICA_KEY_MAX;
 	    }
 	} else {
@@ -253,10 +234,8 @@ mca_gpr_replica_key_t gpr_replica_define_key(char *segment, char *token)
 	    new->key = ptr_key->key;
 	}
 	ompi_list_append(&mca_gpr_replica_head.segment_dict, &new->item);
-	/* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 	return new->key;
     }
-    /* OMPI_THREAD_LOCK(&mca_gpr_replica_internals_mutex); */
 
     /* okay, token is specified */
     /* search the registry segments to find which one is being referenced */
@@ -267,7 +246,6 @@ mca_gpr_replica_key_t gpr_replica_define_key(char *segment, char *token)
 	     ptr_key != (mca_gpr_replica_keytable_t*)ompi_list_get_end(&seg->keytable);
 	     ptr_key = (mca_gpr_replica_keytable_t*)ompi_list_get_next(ptr_key)) {
 	    if (0 == strcmp(token, ptr_key->token)) {
-		/* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 		return ptr_key->key; /* already taken, report value */
 	    }
 	}
@@ -282,7 +260,6 @@ mca_gpr_replica_key_t gpr_replica_define_key(char *segment, char *token)
 	    new->key = ptr_key->key;
 	}
 	ompi_list_append(&seg->keytable, &new->item);
-	/* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 	return new->key;
     }
     /* couldn't find segment */
@@ -302,7 +279,6 @@ int gpr_replica_delete_key(char *segment, char *token)
 	return(OMPI_ERROR);
     }
 
-    /* OMPI_THREAD_LOCK(&mca_gpr_replica_internals_mutex); */
 
     /* find the segment */
     seg = gpr_replica_find_seg(false, segment);
@@ -311,13 +287,11 @@ int gpr_replica_delete_key(char *segment, char *token)
 	/* if specified token is NULL, then this is deleting a segment name.*/
 	if (NULL == token) {
 	    if (OMPI_SUCCESS != gpr_replica_empty_segment(seg)) { /* couldn't empty segment */
-		/* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 		return OMPI_ERROR;
 	    }
 	    /* now remove the dictionary entry from the global registry dictionary*/
 	    ptr_seg = gpr_replica_find_dict_entry(segment, NULL);
 	    if (NULL == ptr_seg) { /* failed to find dictionary entry */
-		/* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 		return OMPI_ERROR;
 	    }
 
@@ -330,7 +304,6 @@ int gpr_replica_delete_key(char *segment, char *token)
 	    /* remove the dictionary entry */
 	    ompi_list_remove_item(&mca_gpr_replica_head.segment_dict, &ptr_seg->item);
 
-	    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 
 	    return(OMPI_SUCCESS);
 
@@ -358,15 +331,12 @@ int gpr_replica_delete_key(char *segment, char *token)
 
 		    /* now remove the dictionary entry from the segment's dictionary */
 		    ompi_list_remove_item(&seg->keytable, &ptr_key->item);
-		    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 		    return(OMPI_SUCCESS);
 		}
 	    }
-	    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 	    return(OMPI_ERROR); /* if we get here, then we couldn't find token in dictionary */
 	}
     }
-    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
     return(OMPI_ERROR); /* if we get here, then we couldn't find segment */
 }
 
@@ -378,7 +348,6 @@ int gpr_replica_empty_segment(mca_gpr_replica_segment_t *seg)
 
     /* need to free memory from each entry - remove_last returns pointer to the entry */
 
-    /* OMPI_THREAD_LOCK(&mca_gpr_replica_internals_mutex); */
 
     /* empty the segment's registry */
     while (!ompi_list_is_empty(&seg->registry_entries)) {
@@ -401,7 +370,6 @@ int gpr_replica_empty_segment(mca_gpr_replica_segment_t *seg)
     ompi_list_remove_item(&mca_gpr_replica_head.registry, &seg->item);
     OBJ_RELEASE(seg);
 
-    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 
     return OMPI_SUCCESS;
 }
@@ -500,11 +468,9 @@ gpr_replica_construct_trigger(ompi_registry_synchro_mode_t synchro_mode,
     mca_gpr_replica_key_t *keyptr;
     int i, num_tokens;
 
-    /* OMPI_THREAD_LOCK(&mca_gpr_replica_internals_mutex); */
 
     seg = gpr_replica_find_seg(true, segment);
     if (NULL == seg) { /* couldn't find or create segment */
-	/* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 	return NULL;
     }
 
@@ -572,7 +538,6 @@ gpr_replica_construct_trigger(ompi_registry_synchro_mode_t synchro_mode,
 
     ompi_list_append(&seg->triggers, &trig->item);
 
-    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 
     return trig;
 
@@ -591,11 +556,9 @@ mca_gpr_notify_id_t gpr_replica_remove_trigger(ompi_registry_synchro_mode_t sync
     int i=0, num_tokens=0;
     bool found=false, mismatch=false;
 
-    /* OMPI_THREAD_LOCK(&mca_gpr_replica_internals_mutex); */
 
     seg = gpr_replica_find_seg(false, segment);
     if (NULL == seg) { /* couldn't find segment */
-	/* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 	return MCA_GPR_NOTIFY_ID_MAX;
     }
 
@@ -655,11 +618,9 @@ mca_gpr_notify_id_t gpr_replica_remove_trigger(ompi_registry_synchro_mode_t sync
 	id_tag = trig->id_tag;
 	ompi_list_remove_item(&seg->triggers, &trig->item);
 	OBJ_RELEASE(trig);
-	/* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 	return id_tag;
     }
 
-    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 
     return MCA_GPR_NOTIFY_ID_MAX;
 }
@@ -828,10 +789,66 @@ bool gpr_replica_process_triggers(char *segment,
 
     return false;
 
-    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 
 }
 
+mca_gpr_notify_id_t gpr_replica_enter_notify_request(ompi_process_name_t *requestor,
+						     mca_gpr_notify_id_t idtag,
+						     ompi_registry_notify_cb_fn_t cb_func,
+						     void *user_tag)
+{
+    mca_gpr_notify_request_tracker_t *trackptr;
+    mca_gpr_idtag_list_t *ptr_free_id;
+
+    trackptr = OBJ_NEW(mca_gpr_notify_request_tracker_t);
+    trackptr->requestor = ompi_name_server.copy_process_name(requestor);
+    trackptr->req_tag = idtag;
+    trackptr->callback = cb_func;
+    trackptr->user_tag = user_tag;
+    if (ompi_list_is_empty(&mca_gpr_replica_free_notify_id_tags)) {
+	trackptr->id_tag = mca_gpr_replica_last_notify_id_tag;
+	mca_gpr_replica_last_notify_id_tag++;
+    } else {
+	ptr_free_id = (mca_gpr_idtag_list_t*)ompi_list_remove_first(&mca_gpr_replica_free_notify_id_tags);
+	trackptr->id_tag = ptr_free_id->id_tag;
+    }
+    ompi_list_append(&mca_gpr_replica_notify_request_tracker, &trackptr->item);
+
+    return trackptr->id_tag;
+}
+
+
+mca_gpr_notify_id_t gpr_replica_remove_notify_request(mca_gpr_notify_id_t idtag)
+{
+    mca_gpr_notify_request_tracker_t *trackptr;
+    mca_gpr_idtag_list_t *ptr_free_id;
+    mca_gpr_notify_id_t ret_value;
+
+    /* find request on replica notify tracking system */
+    for (trackptr = (mca_gpr_notify_request_tracker_t*)ompi_list_get_first(&mca_gpr_replica_notify_request_tracker);
+	 trackptr != (mca_gpr_notify_request_tracker_t*)ompi_list_get_end(&mca_gpr_replica_notify_request_tracker) &&
+	     trackptr->id_tag != idtag;
+	 trackptr = (mca_gpr_notify_request_tracker_t*)ompi_list_get_next(trackptr));
+
+    if (trackptr != (mca_gpr_notify_request_tracker_t*)ompi_list_get_end(&mca_gpr_replica_notify_request_tracker)) {
+	/* save the requestor's tag */
+	ret_value = trackptr->req_tag;
+
+	/* ...and remove the request */
+	ompi_list_remove_item(&mca_gpr_replica_notify_request_tracker, &trackptr->item);
+	/* put local id tag on free list */
+	ptr_free_id = OBJ_NEW(mca_gpr_idtag_list_t);
+	ptr_free_id->id_tag = trackptr->id_tag;
+	ompi_list_append(&mca_gpr_replica_free_notify_id_tags, &ptr_free_id->item);
+	/* release tracker item */
+	OBJ_RELEASE(trackptr);
+
+	return ret_value;
+    }
+    /* error condition if reach here */
+    return MCA_GPR_NOTIFY_ID_MAX;
+}
+ 
 ompi_list_t *gpr_replica_test_internals(int level)
 {
     ompi_list_t *test_results, *keylist;
@@ -844,7 +861,6 @@ ompi_list_t *gpr_replica_test_internals(int level)
     mca_gpr_replica_keytable_t *dict_entry;
     bool success;
 
-    /* OMPI_THREAD_LOCK(&mca_gpr_replica_internals_mutex); */
 
     test_results = OBJ_NEW(ompi_list_t);
 
@@ -1038,7 +1054,6 @@ ompi_list_t *gpr_replica_test_internals(int level)
 
     /* check ability to empty segment */
 
-    /* OMPI_THREAD_UNLOCK(&mca_gpr_replica_internals_mutex); */
 
     return test_results;
 }