rmaps_base_support_fns.c

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

        return ORTE_ERR_OUT_OF_RESOURCE;
    }

    /** setup the working node list to include only those nodes that were specified
     * in this mapping. We don't need to worry about nodes being fully used or not
     * since the master list only includes nodes that aren't.
     */
    for (item  = opal_list_get_first(master_node_list);
         item != opal_list_get_end(master_node_list);
         item  = opal_list_get_next(item) ) {
        node = (orte_ras_node_t*)item;
        
        if( is_mapped(item, mapped_nodes, num_mapped_nodes) ) {
            /** we can't just add this item to the mapped_node_list as it cannot be
             * on two lists at the same time, so we need to copy it first
             */
            if (ORTE_SUCCESS != (rc = orte_dss.copy((void**)&new_node, node, ORTE_RAS_NODE))) {
                ORTE_ERROR_LOG(rc);
                return rc;
            }
            opal_list_append(mapped_node_list, &new_node->super);
            num_slots += new_node->node_slots;
        }
    }

    /** check that anything is left! */
    if (0 == opal_list_get_size(mapped_node_list)) {
        opal_show_help("help-orte-rmaps-base.txt", "orte-rmaps-base:no-mapped-node",
                       true, app->num_procs, app->app);
        ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
        return ORTE_ERR_OUT_OF_RESOURCE;
    }

    *total_num_slots = num_slots;

    return ORTE_SUCCESS;
}


int orte_rmaps_base_add_proc_to_map(orte_job_map_t *map, orte_cellid_t cell, char *nodename, int32_t launch_id,
                                    char *username, bool oversubscribed, orte_mapped_proc_t *proc)
{
    opal_list_item_t *item;
    orte_mapped_node_t *node;
    
    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;
        
        if (cell == node->cell && 0 == strcmp(nodename, node->nodename)) {
            /* node was found - add this proc to that list */
            opal_list_append(&node->procs, &proc->super);
            /* set the oversubscribed flag */
            node->oversubscribed = oversubscribed;
            return ORTE_SUCCESS;
        }
    }
    
    /* node was NOT found - add this one to the list */
    node = OBJ_NEW(orte_mapped_node_t);
    if (NULL == node) {
        ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
        return ORTE_ERR_OUT_OF_RESOURCE;
    }
    
    node->cell = cell;
    node->nodename = strdup(nodename);
    if (NULL != username) {
        node->username = strdup(username);
    }
    node->launch_id = launch_id;
    node->oversubscribed = oversubscribed;
    opal_list_append(&map->nodes, &node->super);
    
    /* and add this proc to the new node's list of procs */
    opal_list_append(&node->procs, &proc->super);
    
    return ORTE_SUCCESS;
}


/*
 * Claim a slot for a specified job on a node
 */
int orte_rmaps_base_claim_slot(orte_job_map_t *map,
                               orte_ras_node_t *current_node,
                               orte_jobid_t jobid, orte_vpid_t vpid,
                               orte_std_cntr_t app_idx,
                               opal_list_t *nodes,
                               opal_list_t *fully_used_nodes,
                               bool oversubscribe)
{
    orte_process_name_t *name;
    orte_mapped_proc_t *proc;
    bool oversub;
    int rc;
    
    /* create mapped_proc object */
    proc = OBJ_NEW(orte_mapped_proc_t);
    if (NULL == proc) {
        ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
        return ORTE_ERR_OUT_OF_RESOURCE;
    }
    
    /* create the process name as an offset from the vpid-start */
    rc = orte_ns.create_process_name(&name, current_node->node_cellid,
                                     jobid, vpid);
    if (rc != ORTE_SUCCESS) {
        ORTE_ERROR_LOG(rc);
        OBJ_RELEASE(proc);
        return rc;
    }
    proc->name = *name;
    proc->rank = vpid;
    proc->app_idx = app_idx;
    
    /* Be sure to demarcate this slot as claimed for the node */
    current_node->node_slots_inuse++;
    
    /* see if this node is oversubscribed now */
    if (current_node->node_slots_inuse > current_node->node_slots) {
        oversub = true;
    } else {
        oversub = false;
    }
    
    /* add the proc to the map */
    if (ORTE_SUCCESS != (rc = orte_rmaps_base_add_proc_to_map(map, current_node->node_cellid,
                                                              current_node->node_name,
                                                              current_node->launch_id,
                                                              current_node->node_username,
                                                              oversub, proc))) {
        ORTE_ERROR_LOG(rc);
        OBJ_RELEASE(proc);
        return rc;
    }
    
    /* Remove this node if it has reached its max number of allocatable slots OR it has
     * reached the soft limit AND we are in a "no oversubscribe" state
     */
    if ((0 != current_node->node_slots_max  &&
        current_node->node_slots_inuse >= current_node->node_slots_max) ||
        (!oversubscribe && current_node->node_slots_inuse >= current_node->node_slots)) {
        opal_list_remove_item(nodes, (opal_list_item_t*)current_node);
        /* add it to the list of fully used nodes */
        opal_list_append(fully_used_nodes, &current_node->super);
        /** now return the proper code so the caller knows we removed the node! */
        return ORTE_ERR_NODE_FULLY_USED;
    }
    
    return ORTE_SUCCESS;
}


/*
 * Update the node allocations stored in the registry
 */
int orte_rmaps_base_update_node_usage(opal_list_t *nodes)
{
    opal_list_item_t* item;
    orte_gpr_value_t **values;
    int rc;
    orte_std_cntr_t num_values, i, j;
    orte_ras_node_t* node;
    
    num_values = (orte_std_cntr_t)opal_list_get_size(nodes);
    if (0 >= num_values) {
        ORTE_ERROR_LOG(ORTE_ERR_BAD_PARAM);
        return ORTE_ERR_BAD_PARAM;
    }
    
    values = (orte_gpr_value_t**)malloc(num_values * sizeof(orte_gpr_value_t*));
    if (NULL == values) {
        ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
        return ORTE_ERR_OUT_OF_RESOURCE;
    }
    
    for (i=0; i < num_values; i++) {
        if (ORTE_SUCCESS != (rc = orte_gpr.create_value(&(values[i]),
                                                        ORTE_GPR_OVERWRITE | ORTE_GPR_TOKENS_AND,
                                                        ORTE_NODE_SEGMENT, 1, 0))) {
            ORTE_ERROR_LOG(rc);
            for (j=0; j < i; j++) {
                OBJ_RELEASE(values[j]);
            }
            free(values);
            return rc;
        }
    }
    
    for(i=0, item =  opal_list_get_first(nodes);
        i < num_values && item != opal_list_get_end(nodes);
        i++, item =  opal_list_get_next(item)) {
        node = (orte_ras_node_t*)item;
        
        if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(values[i]->keyvals[0]), ORTE_NODE_SLOTS_IN_USE_KEY,
                                                         ORTE_STD_CNTR, &(node->node_slots_inuse)))) {
            ORTE_ERROR_LOG(rc);
            goto cleanup;
        }

        /* setup index/keys for this node */
        rc = orte_schema.get_node_tokens(&(values[i]->tokens), &(values[i]->num_tokens), node->node_cellid, node->node_name);
        if (ORTE_SUCCESS != rc) {
            ORTE_ERROR_LOG(rc);
            goto cleanup;
        }
    }
    
    /* try the insert */
    if (ORTE_SUCCESS != (rc = orte_gpr.put(num_values, values))) {
        ORTE_ERROR_LOG(rc);
    }
    
cleanup:
    for (j=0; j < num_values; j++) {
        OBJ_RELEASE(values[j]);
    }
    if (NULL != values) free(values);
    
    return rc;
}