rmaps_base_registry_fns.c

MPI stands for the Message Passing Interface. Written by the MPI Forum (a large committee comprising

    }
    
    if (ORTE_SUCCESS != (rc = orte_schema.get_job_segment_name(&segment, map->job))) {
        ORTE_ERROR_LOG(rc);
        free(values);
        return rc;
    }

    /** setup the last value in the array to store the vpid start/range and update the INIT counter */
    if (ORTE_SUCCESS != (rc = orte_gpr.create_value(&(values[num_procs]),
                                            ORTE_GPR_OVERWRITE|ORTE_GPR_TOKENS_AND,
                                            segment, 4, 1))) {
        ORTE_ERROR_LOG(rc);
        free(values);
        free(segment);
        return rc;
    }
    if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(values[num_procs]->keyvals[0]), ORTE_PROC_NUM_AT_INIT, ORTE_STD_CNTR, &num_procs))) {
        ORTE_ERROR_LOG(rc);
        goto cleanup;
    }
    if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(values[num_procs]->keyvals[1]), ORTE_JOB_VPID_START_KEY, ORTE_VPID, &map->vpid_start))) {
        ORTE_ERROR_LOG(rc);
        goto cleanup;
    }
    if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(values[num_procs]->keyvals[2]), ORTE_JOB_VPID_RANGE_KEY, ORTE_VPID, &map->vpid_range))) {
        ORTE_ERROR_LOG(rc);
        goto cleanup;
    }
    if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(values[num_procs]->keyvals[3]), ORTE_JOB_MAPPING_MODE_KEY, ORTE_STRING, map->mapping_mode))) {
        ORTE_ERROR_LOG(rc);
        goto cleanup;
    }
    values[num_procs]->tokens[0] = strdup(ORTE_JOB_GLOBALS); /* counter is in the job's globals container */


    for(i=0; i<num_procs; i++) {
        if (ORTE_SUCCESS != (rc = orte_gpr.create_value(&(values[i]),
                                    ORTE_GPR_OVERWRITE|ORTE_GPR_TOKENS_AND,
                                    segment, 9, 0))) {
             ORTE_ERROR_LOG(rc);
             for(j=0; j<i; j++) {
                 OBJ_RELEASE(values[j]);
             }
             free(values);
             free(segment);
             return rc;
         }
    }

    /* iterate through all processes and initialize value array */
    for(item =  opal_list_get_first(&map->nodes);
        item != opal_list_get_end(&map->nodes);
        item =  opal_list_get_next(item)) {
        node = (orte_mapped_node_t*)item;

        for (item2 = opal_list_get_first(&node->procs);
             item2 != opal_list_get_end(&node->procs);
             item2 = opal_list_get_next(item2)) {
            proc = (orte_mapped_proc_t*)item2;
            
            value = values[index++];

            /* initialize keyvals */
            if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[0]), ORTE_PROC_RANK_KEY, ORTE_STD_CNTR, &(proc->rank)))) {
                ORTE_ERROR_LOG(rc);
                goto cleanup;
            }

            if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[1]), ORTE_PROC_NAME_KEY, ORTE_NAME, &(proc->name)))) {
                ORTE_ERROR_LOG(rc);
                goto cleanup;
            }

            if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[2]), ORTE_CELLID_KEY, ORTE_CELLID, &(node->cell)))) {
                ORTE_ERROR_LOG(rc);
                goto cleanup;
            }

            if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[3]), ORTE_NODE_NAME_KEY, ORTE_STRING, node->nodename))) {
                ORTE_ERROR_LOG(rc);
                goto cleanup;
            }

            if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[4]), ORTE_NODE_LAUNCH_ID_KEY, ORTE_INT32, &(node->launch_id)))) {
                ORTE_ERROR_LOG(rc);
                goto cleanup;
            }
            
            if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[5]), ORTE_NODE_USERNAME_KEY, ORTE_STRING, node->username))) {
                ORTE_ERROR_LOG(rc);
                goto cleanup;
            }
            
            if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[6]), ORTE_NODE_OVERSUBSCRIBED_KEY, ORTE_BOOL, &(node->oversubscribed)))) {
                ORTE_ERROR_LOG(rc);
                goto cleanup;
            }
            
            if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[7]), ORTE_PROC_APP_CONTEXT_KEY, ORTE_STD_CNTR, &(proc->app_idx)))) {
                ORTE_ERROR_LOG(rc);
                goto cleanup;
            }

            if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[8]), ORTE_PROC_STATE_KEY, ORTE_PROC_STATE, &proc_state))) {
                ORTE_ERROR_LOG(rc);
                goto cleanup;
            }

            /* set the tokens */
            if (ORTE_SUCCESS != (rc = orte_schema.get_proc_tokens(&(value->tokens), &(value->num_tokens), &(proc->name)))) {
                ORTE_ERROR_LOG(rc);
                goto cleanup;
            }
        }
    }

    /* insert all values in one call */
    if (ORTE_SUCCESS != (rc = orte_gpr.put((1+num_procs), values))) {
        ORTE_ERROR_LOG(rc);
    }

cleanup:
    for(i=0; i<=num_procs; i++) {
        if(NULL != values[i]) {
            OBJ_RELEASE(values[i]);
        }
    }
    if(NULL != segment) {
        free(segment);
    }
    if(NULL != values)
        free(values);
    return rc;
}


/* Mapping plans are associated with a job - hence, they are stored in the job's
 * container on the JOB_MASTER_SEGMENT
 */
int orte_rmaps_base_store_mapping_plan(orte_jobid_t job, opal_list_t *attr_list)
{
    int rc;
    orte_attribute_t *attr;
    orte_gpr_value_t *value;
    orte_std_cntr_t i, j, num_attrs_found, num_tokens;
    char *attrs[] = {
        ORTE_RMAPS_MAP_POLICY,
        ORTE_RMAPS_PERNODE,
        ORTE_RMAPS_NO_USE_LOCAL,
        ORTE_RMAPS_NO_OVERSUB,
        ORTE_RMAPS_DESIRED_MAPPER,
        ORTE_RMAPS_USE_PARENT_PLAN,
        ORTE_RMAPS_BOOKMARK
    };
    orte_std_cntr_t num_attrs_defd;
    
    OPAL_TRACE(2);
    
    num_attrs_defd = sizeof(attrs)/sizeof(char*);
    
    /* count the number of attributes we will need to store */
    num_attrs_found = 0;
    for (i=0; i < num_attrs_defd; i++) {
        if (NULL != orte_rmgr.find_attribute(attr_list, attrs[i])) num_attrs_found++;
    }

    /* if nothing found, then nothing to do! */
    if (0 == num_attrs_found) return ORTE_SUCCESS;
    
    /* setup to store the found values */
    if (ORTE_SUCCESS != (rc = orte_gpr.create_value(&value,
                                                    ORTE_GPR_OVERWRITE|ORTE_GPR_TOKENS_AND,
                                                    ORTE_JOBINFO_SEGMENT, num_attrs_found, 0))) {
        ORTE_ERROR_LOG(rc);
        return rc;
    }
    
    /* setup the tokens to point to this job's container */
    if (ORTE_SUCCESS != (rc = orte_schema.get_job_tokens(&(value->tokens), &num_tokens, job))) {
        ORTE_ERROR_LOG(rc);
        OBJ_RELEASE(value);
        return rc;
    }
    
    /* copy the data that is to be stored */
    for (i=0, j=0; i < num_attrs_defd; i++) {
        if (NULL != (attr = orte_rmgr.find_attribute(attr_list, attrs[i]))) {
            if (NULL != attr->value) {
                if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[j]), attr->key,
                                                                 attr->value->type, attr->value->data))) {
                    ORTE_ERROR_LOG(rc);
                    OBJ_RELEASE(value);
                    return rc;
                }
            } else {
                if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[j]), attr->key,
                                                                 ORTE_UNDEF, NULL))) {
                    ORTE_ERROR_LOG(rc);
                    OBJ_RELEASE(value);
                    return rc;
                }
            }
            j++;
        }
    }
    
    /* put the data onto the registry */
    if (ORTE_SUCCESS != (rc = orte_gpr.put(1, &value))) {
        ORTE_ERROR_LOG(rc);
    }
    
    /* cleanup memory */
    OBJ_RELEASE(value);
    
    return rc;
}


int orte_rmaps_base_get_mapping_plan(orte_jobid_t job, opal_list_t *attr_list)
{
    int rc;
    orte_gpr_value_t **values, *value;
    orte_gpr_keyval_t *kval;
    orte_std_cntr_t i, num_vals, num_tokens;
    char *attrs[] = {
        ORTE_RMAPS_MAP_POLICY,
        ORTE_RMAPS_PERNODE,
        ORTE_RMAPS_NO_USE_LOCAL,
        ORTE_RMAPS_NO_OVERSUB,
        ORTE_RMAPS_DESIRED_MAPPER,
        ORTE_RMAPS_USE_PARENT_PLAN,
        ORTE_RMAPS_BOOKMARK,
        NULL
    };
    orte_std_cntr_t num_attrs_defd;
    char **tokens;
    
    OPAL_TRACE(2);
    
    num_attrs_defd = sizeof(attrs)/sizeof(char*);
    
    /* setup the tokens to point to this job's container */
    if (ORTE_SUCCESS != (rc = orte_schema.get_job_tokens(&tokens, &num_tokens, job))) {
        ORTE_ERROR_LOG(rc);
        return rc;
    }
    
    /* query the mapping plan data from the registry */
    if (ORTE_SUCCESS != (rc = orte_gpr.get(ORTE_GPR_KEYS_OR|ORTE_GPR_TOKENS_OR,
                                           ORTE_JOBINFO_SEGMENT,
                                           tokens, attrs,
                                           &num_vals, &values))) {
        ORTE_ERROR_LOG(rc);
        return rc;
    }
    
    /*  It is okay for there to be 0 values returned as this just means a mapping plan
     * was not previously stored on the registry
     */
    if (0 == num_vals) {
        return ORTE_SUCCESS;
    }
    
    /* should only be one value returned here since there is only one
     * container/job on the segment - error otherwise.
     */
    if (1 < num_vals) {
        ORTE_ERROR_LOG(ORTE_ERR_GPR_DATA_CORRUPT);
        return ORTE_ERR_GPR_DATA_CORRUPT;
    }
    
    /* update the data on the list. This will OVERWRITE any matching data
     * on that list....USER BEWARE!
     */
    value = values[0];
    for (i=0; i < value->cnt; i++) {
        kval = value->keyvals[i];
        
        if (NULL != kval->value) {
            if (ORTE_SUCCESS != (rc = orte_rmgr.add_attribute(attr_list, kval->key,
                                                              kval->value->type,
                                                              kval->value->data,
                                                              ORTE_RMGR_ATTR_OVERRIDE))) {
                ORTE_ERROR_LOG(rc);
                OBJ_RELEASE(value);
                return rc;
            }
        } else {
            if (ORTE_SUCCESS != (rc = orte_rmgr.add_attribute(attr_list, kval->key,
                                                              ORTE_UNDEF,
                                                              NULL,
                                                              ORTE_RMGR_ATTR_OVERRIDE))) {
                ORTE_ERROR_LOG(rc);
                OBJ_RELEASE(value);
                return rc;
            }
        }
    }
    
    OBJ_RELEASE(value);
    
    return ORTE_SUCCESS;
}


int orte_rmaps_base_update_mapping_state(orte_jobid_t parent_job,
                                         opal_list_t *attrs)
{
    int rc;
    orte_attribute_t *attr;
    orte_gpr_value_t *value;
    orte_std_cntr_t num_tokens;
    
    OPAL_TRACE(2);
    
    /* see if the bookmark is present - if not, we report this as an error so
     * that the RMAPS component developer can correct it
     */
    if (NULL == (attr = orte_rmgr.find_attribute(attrs, ORTE_RMAPS_BOOKMARK))) {
        return ORTE_ERR_NOT_FOUND;
    }
    
    /* setup to store the bookmark */
    if (ORTE_SUCCESS != (rc = orte_gpr.create_value(&value,
                                                    ORTE_GPR_OVERWRITE|ORTE_GPR_TOKENS_AND,
                                                    ORTE_JOBINFO_SEGMENT, 1, 0))) {
        ORTE_ERROR_LOG(rc);
        return rc;
    }
    
    /* setup the tokens to point to this job's container */
    if (ORTE_SUCCESS != (rc = orte_schema.get_job_tokens(&(value->tokens), &num_tokens, parent_job))) {
        ORTE_ERROR_LOG(rc);
        OBJ_RELEASE(value);
        return rc;
    }
    
    /* copy the data that is to be stored */
    if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[0]), attr->key,
                                                     attr->value->type, attr->value->data))) {
        ORTE_ERROR_LOG(rc);
        OBJ_RELEASE(value);
        return rc;
    }
    
    /* put the data onto the registry */
    if (ORTE_SUCCESS != (rc = orte_gpr.put(1, &value))) {
        ORTE_ERROR_LOG(rc);
    }
    
    /* cleanup memory */
    OBJ_RELEASE(value);
    
    return rc;
}