restore subscribe for connect/accept

This commit was SVN r5280.

src/mca/oob/tcp/oob_tcp.c

@@ -555,9 +555,9 @@ int mca_oob_tcp_resolve(mca_oob_tcp_peer_t* peer)
 {
     mca_oob_tcp_addr_t* addr;
     mca_oob_tcp_subscription_t* subscription;
+    orte_gpr_value_t trig, *trigs;
+    orte_gpr_subscription_t sub, *subs;
     ompi_list_item_t* item;
-     char *segment, *jobid;
-     char *keys[2];
     int rc;
   
     /* if the address is already cached - simply return it */
@@ -581,40 +581,91 @@ int mca_oob_tcp_resolve(mca_oob_tcp_peer_t* peer)
         }
     }
                                                                                                         
-     /* otherwise - need to subscribe to this registry segment
+    OBJ_CONSTRUCT(&sub, orte_gpr_subscription_t);
-      * record the subscription
+    sub.addr_mode = ORTE_GPR_TOKENS_OR | ORTE_GPR_KEYS_OR;
-     */
+    if (ORTE_SUCCESS != (rc = orte_schema.get_job_segment_name(&(sub.segment), peer->peer_name.jobid))) {
-     subscription = OBJ_NEW(mca_oob_tcp_subscription_t);
+        ORTE_ERROR_LOG(rc);
-     subscription->jobid = peer->peer_name.jobid;
-     ompi_list_append(&mca_oob_tcp_component.tcp_subscriptions, &subscription->item);
-     OMPI_THREAD_UNLOCK(&mca_oob_tcp_component.tcp_lock);
-
-     if (ORTE_SUCCESS != (rc = orte_ns.get_jobid_string(&jobid, &peer->peer_name))) {
         return rc;
     }
-     asprintf(&segment, "%s-%s", ORTE_JOB_SEGMENT, jobid);
+    sub.num_tokens= 0;
-     keys[0] = "oob-tcp";
+    sub.tokens = NULL;
-     keys[1] = NULL;
+    sub.num_keys = 1;
+    sub.keys = (char**)malloc(sizeof(char*));
+    if(NULL == sub.keys) {
+        ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
+        return ORTE_ERR_OUT_OF_RESOURCE;
+    }
+    sub.keys[0] = strdup("oob-tcp");
+    if (NULL == sub.keys[0]) {
+        ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
+        return ORTE_ERR_OUT_OF_RESOURCE;
+    }
+    sub.cbfunc = mca_oob_tcp_registry_callback;
+    sub.user_tag = NULL;
 
-#if 0
+    /* setup the trigger value */
-     rc = orte_gpr.subscribe(
+    OBJ_CONSTRUCT(&trig, orte_gpr_value_t);
-         ORTE_GPR_OR,
+    trig.addr_mode = ORTE_GPR_TOKENS_XAND | ORTE_GPR_KEYS_OR;
-         ORTE_GPR_NOTIFY_ON_STARTUP|ORTE_GPR_NOTIFY_INCLUDE_STARTUP_DATA|
+    if (ORTE_SUCCESS != (rc = orte_schema.get_job_segment_name(&(trig.segment), peer->peer_name.jobid))) {
-		 ORTE_GPR_NOTIFY_ON_SHUTDOWN|ORTE_GPR_NOTIFY_INCLUDE_SHUTDOWN_DATA|
+        ORTE_ERROR_LOG(rc);
-		 ORTE_GPR_NOTIFY_PRE_EXISTING,
+        OBJ_DESTRUCT(&trig);
-         segment,
+        return rc;
-         NULL,
+    }
-         keys,
+    trig.num_tokens = 1;
-         &subscription->subid,
+    trig.tokens = (char**)malloc(sizeof(char*));
-         mca_oob_tcp_registry_callback,
+    if (NULL == trig.tokens) {
-         NULL);
+        ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
-
+        return ORTE_ERR_OUT_OF_RESOURCE;
-     free(segment);
+    }
-     free(jobid);
+    trig.tokens[0] = strdup(ORTE_JOB_GLOBALS);
+    if (NULL == trig.tokens[0]) {
+        ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
+        return ORTE_ERR_OUT_OF_RESOURCE;
+    }
+    trig.keyvals = (orte_gpr_keyval_t**)malloc(2*sizeof(orte_gpr_keyval_t*));
+    if(NULL == trig.keyvals) {
+        ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
+        return ORTE_ERR_OUT_OF_RESOURCE;
+    }
+    trig.cnt = 2;
+    trig.keyvals[0] = OBJ_NEW(orte_gpr_keyval_t);
+    if(NULL == trig.keyvals[0]) {
+        ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
+        return ORTE_ERR_OUT_OF_RESOURCE;
+    }
+    trig.keyvals[0]->key = strdup(ORTE_JOB_SLOTS_KEY);
+    trig.keyvals[0]->type = ORTE_NULL;
+    
+    trig.keyvals[1] = OBJ_NEW(orte_gpr_keyval_t);
+    if(NULL == trig.keyvals[1]) {
+        ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
+        return ORTE_ERR_OUT_OF_RESOURCE;
+    }
+    trig.keyvals[1]->key = strdup(ORTE_PROC_NUM_AT_STG1);
+    trig.keyvals[1]->type = ORTE_NULL;
+    
+    trigs = &trig;
+    subs = ⊂
+    subscription = OBJ_NEW(mca_oob_tcp_subscription_t);
+    subscription->jobid = peer->peer_name.jobid;
+    rc = orte_gpr.subscribe(
+        ORTE_GPR_NOTIFY_ADD_ENTRY | ORTE_GPR_NOTIFY_VALUE_CHG |
+        ORTE_GPR_TRIG_CMP_LEVELS | ORTE_GPR_TRIG_ONE_SHOT,
+        1, &subs,
+        1, &trigs,
+        &subscription->subid);
+    if(rc != OMPI_SUCCESS) {
+        ORTE_ERROR_LOG(rc);
+        OBJ_DESTRUCT(&sub);
+        OBJ_DESTRUCT(&trig);
+        return rc;
+    }
+    OBJ_DESTRUCT(&sub);
+    OBJ_DESTRUCT(&trig);  /* done with these */
+
+    ompi_list_append(&mca_oob_tcp_component.tcp_subscriptions, &subscription->item);
+    OMPI_THREAD_UNLOCK(&mca_oob_tcp_component.tcp_lock);
     return rc;
-#else 
-     return ORTE_SUCCESS;
-#endif
 }