smr_base_trig_init_fns.c

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

    /** store the number of orteds */
    if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[0]),
                                                     ORTED_NUM_TO_BE_LAUNCHED, ORTE_STD_CNTR, &num_orteds))) {
        ORTE_ERROR_LOG(rc);
        free(segment);
        OBJ_RELEASE(value);
        return rc;
    }
    /** initialize the counter to zero */
    if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[1]), ORTED_LAUNCH_STAGE_GATE_CNTR, ORTE_STD_CNTR, &zero))) {
        ORTE_ERROR_LOG(rc);
        free(segment);
        OBJ_RELEASE(value);
        return rc;
    }
    
    /* put the counters on the registry */
    if (ORTE_SUCCESS != (rc = orte_gpr.put(1, &value))) {
        ORTE_ERROR_LOG(rc);
        free(segment);
        OBJ_RELEASE(value);
        return rc;
    }
    OBJ_RELEASE(value);
    
    /* now define a trigger based on those counters */
    if (ORTE_SUCCESS != (rc = orte_schema.get_std_trigger_name(&trig_name,
                                                               ORTED_LAUNCH_STAGE_GATE_TRIGGER, job))) {
        ORTE_ERROR_LOG(rc);
        free(segment);
        return rc;
    }
    if (ORTE_SUCCESS != (rc = orte_gpr.define_trigger(&id, trig_name,
                                                      ORTE_GPR_TRIG_INCLUDE_TRIG_CNTRS | ORTE_GPR_TRIG_ONE_SHOT |
                                                      ORTE_GPR_TRIG_CMP_LEVELS,
                                                      ORTE_GPR_TOKENS_XAND | ORTE_GPR_KEYS_OR,
                                                      segment, trig_tokens, 2, trig_keys,
                                                      cbfunc, user_tag))) {
        ORTE_ERROR_LOG(rc);
        free(segment);
        free(trig_name);
        return rc;
    }
    
    free(segment);
    free(trig_name);
    return ORTE_SUCCESS;    
}


/*
 * Setup an alert monitor
 */
int orte_smr_base_define_alert_monitor(orte_jobid_t job,
                                       char *trigger_name,
                                       char *counter_key,
                                       orte_std_cntr_t init_value,
                                       orte_std_cntr_t alert_value,
                                       bool one_shot,
                                       orte_gpr_trigger_cb_fn_t cbfunc,
                                       void *user_tag)
{
    int rc;
    orte_gpr_value_t *value;
    orte_gpr_trigger_action_t trig_mode;
    char *segment, *trig_name;
    char *tokens[] = {
        ORTE_JOB_GLOBALS,
        NULL
    };
    orte_gpr_trigger_id_t id;
    
    /* get the job's segment name */
    if (ORTE_SUCCESS != (rc = orte_schema.get_job_segment_name(&segment, job))) {
        ORTE_ERROR_LOG(rc);
        return rc;
    }
    
    /* setup the counters */
    if (ORTE_SUCCESS != (rc = orte_gpr.create_value(&value,
                                                    ORTE_GPR_OVERWRITE | ORTE_GPR_TOKENS_XAND | ORTE_GPR_KEYS_OR,
                                                    segment, 1, 1))) {
        
        ORTE_ERROR_LOG(rc);
        free(segment);
        return rc;
    }    
    value->tokens[0] = strdup(ORTE_JOB_GLOBALS); /* put counters in the job's globals container */
    
    if (ORTE_SUCCESS != (rc = orte_gpr.create_keyval(&(value->keyvals[0]), counter_key, ORTE_STD_CNTR, &init_value))) {
        ORTE_ERROR_LOG(rc);
        OBJ_RELEASE(value);
        return rc;
    }
    
    /* put the counter on the registry */
    if (ORTE_SUCCESS != (rc = orte_gpr.put(1, &value))) {
        ORTE_ERROR_LOG(rc);
        OBJ_RELEASE(value);
        free(segment);
        return rc;
    }
    OBJ_RELEASE(value);
    
    
    /* define the trigger to fire at specified level */
    if (ORTE_SUCCESS != (rc = orte_schema.get_std_trigger_name(&trig_name,
                                                               trigger_name, job))) {
        ORTE_ERROR_LOG(rc);
        free(segment);
        return rc;
    }
    trig_mode = ORTE_GPR_TRIG_INCLUDE_TRIG_CNTRS | ORTE_GPR_TRIG_AT_LEVEL;
    if (one_shot) {
        trig_mode = trig_mode | ORTE_GPR_TRIG_ONE_SHOT;
    }
    
    if (ORTE_SUCCESS != (rc = orte_gpr.define_trigger_level(&id, trig_name,
                                                            trig_mode,
                                                            ORTE_GPR_TOKENS_XAND | ORTE_GPR_KEYS_OR,
                                                            segment, tokens, 1, &counter_key, &alert_value,
                                                            cbfunc, user_tag))) {
        ORTE_ERROR_LOG(rc);
        free(segment);
        free(trig_name);
        return rc;
    }
    free(segment);
    free(trig_name);
    
    return ORTE_SUCCESS;    
}


/*
 * Routine that tools such as orterun can use to subscribe
 * to events on all counters.
 */

int orte_smr_base_job_stage_gate_subscribe(orte_jobid_t job,
                                           orte_gpr_notify_cb_fn_t cbfunc, void* cbdata,
                                           orte_proc_state_t cb_conditions)
{
    orte_std_cntr_t i;
    int rc;
    char *segment, *trig_name, *tokens[2];
    orte_proc_state_t conditions;
    orte_gpr_subscription_id_t id;
    /** the order of the next three definitions MUST match */
    orte_proc_state_t state[] = {
        ORTE_PROC_STATE_INIT,
        ORTE_PROC_STATE_LAUNCHED,
        ORTE_PROC_STATE_RUNNING,
        ORTE_PROC_STATE_AT_STG1,
        ORTE_PROC_STATE_AT_STG2,
        ORTE_PROC_STATE_AT_STG3,
        ORTE_PROC_STATE_FINALIZED,
        ORTE_PROC_STATE_TERMINATED
    };
    char* keys[] = {
        ORTE_PROC_NUM_AT_INIT,
        ORTE_PROC_NUM_LAUNCHED,
        ORTE_PROC_NUM_RUNNING,
        ORTE_PROC_NUM_AT_STG1,
        ORTE_PROC_NUM_AT_STG2,
        ORTE_PROC_NUM_AT_STG3,
        ORTE_PROC_NUM_FINALIZED,
        ORTE_PROC_NUM_TERMINATED
    };
    char* trig_names[] = {
        ORTE_ALL_INIT_TRIGGER,
        ORTE_ALL_LAUNCHED_TRIGGER,
        ORTE_ALL_RUNNING_TRIGGER,
        ORTE_STG1_TRIGGER,
        ORTE_STG2_TRIGGER,
        ORTE_STG3_TRIGGER,
        ORTE_NUM_FINALIZED_TRIGGER,
        ORTE_NUM_TERMINATED_TRIGGER
    };
    orte_std_cntr_t num_counters = sizeof(keys)/sizeof(keys[0]);

    /* identify the segment for this job */
    if (ORTE_SUCCESS != (rc = orte_schema.get_job_segment_name(&segment, job))) {
        ORTE_ERROR_LOG(rc);
        return rc;
    }

    /* setup the tokens */
    tokens[0]=ORTE_JOB_GLOBALS;
    tokens[1]=NULL;

    conditions = cb_conditions;
    for (i=0; i < num_counters; i++) {
        if (state[i] & conditions) {
            /** want this one - attach ourselves to the appropriate standard trigger */
            if (ORTE_SUCCESS !=
                (rc = orte_schema.get_std_trigger_name(&trig_name, trig_names[i], job))) {
                ORTE_ERROR_LOG(rc);
                free(segment);
                return rc;
            }

            if (ORTE_SUCCESS != (rc = orte_gpr.subscribe_1(&id, trig_name, NULL,
                                        ORTE_GPR_NOTIFY_DELETE_AFTER_TRIG,
                                        ORTE_GPR_TOKENS_OR | ORTE_GPR_KEYS_OR,
                                        segment, tokens, keys[i],
                                        cbfunc, cbdata))) {
                ORTE_ERROR_LOG(rc);
                free(segment);
                free(trig_name);
                return rc;
            }
            free(trig_name);
            /* clear the spot in the conditions so we can check that all were found */
            conditions = conditions & ~(state[i]);
        }
    }
    free(segment);

    return ORTE_SUCCESS;
}