ports/openmpi
ports/openmpi
1
1
Форкнуть 0
Код Релизы Активность

Add a job_info segment to the system that holds a container for each job. Within each container is a keyval indicating the job state (i.e., all procs at stage1, finalized, etc.). This provides a rough state-of-health for the job.

Просмотр исходного кода 
This required a little fiddling with a number of areas. Biggest problem was that it uncovered a potential for an infinite loop to be created in the registry. If a callback function modified the registry, the registry checked the triggers to see if anything had fired. Well, if the original callback was due to a trigger firing, that condition hadn't changed - so the trigger fired again....which caused the callback to be called, which modified the registry, which checked the triggers, etc. etc.

Triggers are now checked and then "flagged" as being "in process" so that the registry will NOT recheck that trigger until all callbacks have been processed. Tried doing this with subscriptions as well, but that caused a problem - when we release processes from a stagegate, they (at the moment) immediately place data on the registry that should cause a subscription to fire. Unfortunately, the system will just hang if that subscription doesn't get processed. So, I have left the subscription system alone - any callback function that modifies the registry in a fashion that will fire a subscription will indeed fire that subscription. We'll have to see if this causes problems - it shouldn't, but a careless user could lock things up if the callback generates a callback to itself.

Also fixed the code that placed a process' RML contact info on the registry to eliminate the leading '/' from the string.

This commit was SVN r6684.
Этот коммит содержится в:
Ralph Castain 2005-07-29 14:11:19 +00:00
родитель 6528ee4692
Коммит 4e79a51395
29 изменённых файлов: 547 добавлений и 118 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

11
ompi/runtime/ompi_mpi_finalize.c
Просмотреть файл

@ -209,6 +209,17 @@ int ompi_mpi_finalize(void)
ORTE_ERROR_LOG(ret);
}
/*
* Wait for everyone to get here. This is necessary to allow the soh
* to update the job state for singletons. Otherwise, we finalize
* the RTE while the soh is trying to do the update - which causes
* an ugly race condition
*/
if (ORTE_SUCCESS != (ret = orte_rml.xcast(NULL, NULL, 0, NULL, NULL))) {
ORTE_ERROR_LOG(ret);
return ret;
}
/* Leave the RTE */
if (OMPI_SUCCESS != (ret = orte_finalize())) {

31
orte/include/orte_types.h
Просмотреть файл

@ -75,23 +75,24 @@ typedef uint8_t orte_data_type_t ;
/* SOH types */
#define ORTE_NODE_STATE (orte_data_type_t) 31 /**< node status flag */
#define ORTE_PROC_STATE (orte_data_type_t) 32 /**< process/resource status */
#define ORTE_EXIT_CODE (orte_data_type_t) 33 /**< process exit code */
#define ORTE_JOB_STATE (orte_data_type_t) 33 /**< job status flag */
#define ORTE_EXIT_CODE (orte_data_type_t) 34 /**< process exit code */
/* GPR types */
#define ORTE_KEYVAL (orte_data_type_t) 34 /**< registry key-value pair */
#define ORTE_GPR_NOTIFY_ACTION (orte_data_type_t) 35 /**< registry notify action */
#define ORTE_GPR_TRIGGER_ACTION (orte_data_type_t) 36 /**< registry trigger action */
#define ORTE_GPR_CMD (orte_data_type_t) 37 /**< registry command */
#define ORTE_GPR_SUBSCRIPTION_ID (orte_data_type_t) 38 /**< registry notify id tag */
#define ORTE_GPR_TRIGGER_ID (orte_data_type_t) 39 /**< registry notify id tag */
#define ORTE_GPR_VALUE (orte_data_type_t) 40 /**< registry return value */
#define ORTE_GPR_ADDR_MODE (orte_data_type_t) 41 /**< Addressing mode for registry cmds */
#define ORTE_GPR_SUBSCRIPTION (orte_data_type_t) 42 /**< describes data returned by subscription */
#define ORTE_GPR_TRIGGER (orte_data_type_t) 43 /**< describes trigger conditions */
#define ORTE_GPR_NOTIFY_DATA (orte_data_type_t) 44 /**< data returned from a subscription */
#define ORTE_GPR_NOTIFY_MSG (orte_data_type_t) 45 /**< notify message containing notify_data objects */
#define ORTE_KEYVAL (orte_data_type_t) 35 /**< registry key-value pair */
#define ORTE_GPR_NOTIFY_ACTION (orte_data_type_t) 36 /**< registry notify action */
#define ORTE_GPR_TRIGGER_ACTION (orte_data_type_t) 37 /**< registry trigger action */
#define ORTE_GPR_CMD (orte_data_type_t) 38 /**< registry command */
#define ORTE_GPR_SUBSCRIPTION_ID (orte_data_type_t) 39 /**< registry notify id tag */
#define ORTE_GPR_TRIGGER_ID (orte_data_type_t) 40 /**< registry notify id tag */
#define ORTE_GPR_VALUE (orte_data_type_t) 41 /**< registry return value */
#define ORTE_GPR_ADDR_MODE (orte_data_type_t) 42 /**< Addressing mode for registry cmds */
#define ORTE_GPR_SUBSCRIPTION (orte_data_type_t) 43 /**< describes data returned by subscription */
#define ORTE_GPR_TRIGGER (orte_data_type_t) 44 /**< describes trigger conditions */
#define ORTE_GPR_NOTIFY_DATA (orte_data_type_t) 45 /**< data returned from a subscription */
#define ORTE_GPR_NOTIFY_MSG (orte_data_type_t) 46 /**< notify message containing notify_data objects */
/* Resource Manager types */
#define ORTE_APP_CONTEXT (orte_data_type_t) 46 /**< argv and enviro arrays */
#define ORTE_APP_CONTEXT_MAP (orte_data_type_t) 47 /**< application context mapping array */
#define ORTE_APP_CONTEXT (orte_data_type_t) 47 /**< argv and enviro arrays */
#define ORTE_APP_CONTEXT_MAP (orte_data_type_t) 48 /**< application context mapping array */
/* define the starting point for dynamically assigning data types */
#define ORTE_DPS_ID_DYNAMIC 50

84
orte/mca/gpr/base/gpr_base_xfer_payload.c
Просмотреть файл

@ -34,18 +34,6 @@ int orte_gpr_base_xfer_payload(orte_gpr_value_union_t *dest,
switch(type) {
case ORTE_BOOL:
dest->tf_flag = src->tf_flag;
break;
case ORTE_SIZE:
dest->size = src->size;
break;
case ORTE_PID:
dest->pid = src->pid;
break;
case ORTE_STRING:
dest->strptr = strdup(src->strptr);
if (NULL == dest->strptr) {
@ -54,6 +42,18 @@ int orte_gpr_base_xfer_payload(orte_gpr_value_union_t *dest,
}
break;
case ORTE_SIZE:
dest->size = src->size;
break;
case ORTE_BOOL:
dest->tf_flag = src->tf_flag;
break;
case ORTE_PID:
dest->pid = src->pid;
break;
case ORTE_UINT8:
dest->ui8 = src->ui8;
break;
@ -90,34 +90,6 @@ int orte_gpr_base_xfer_payload(orte_gpr_value_union_t *dest,
break;
#endif
case ORTE_NAME:
dest->proc = src->proc;;
break;
case ORTE_JOBID:
dest->jobid = src->jobid;
break;
case ORTE_CELLID:
dest->cellid = src->cellid;
break;
case ORTE_VPID:
dest->vpid = src->vpid;
break;
case ORTE_NODE_STATE:
dest->node_state = src->node_state;
break;
case ORTE_PROC_STATE:
dest->proc_state = src->proc_state;
break;
case ORTE_EXIT_CODE:
dest->exit_code = src->exit_code;
break;
case ORTE_BYTE_OBJECT:
(dest->byteobject).size = (src->byteobject).size;
(dest->byteobject).bytes = (uint8_t*)malloc((dest->byteobject).size);
@ -128,6 +100,38 @@ int orte_gpr_base_xfer_payload(orte_gpr_value_union_t *dest,
memcpy((dest->byteobject).bytes, (src->byteobject).bytes, (dest->byteobject).size);
break;
case ORTE_NAME:
dest->proc = src->proc;;
break;
case ORTE_VPID:
dest->vpid = src->vpid;
break;
case ORTE_JOBID:
dest->jobid = src->jobid;
break;
case ORTE_CELLID:
dest->cellid = src->cellid;
break;
case ORTE_NODE_STATE:
dest->node_state = src->node_state;
break;
case ORTE_PROC_STATE:
dest->proc_state = src->proc_state;
break;
case ORTE_JOB_STATE:
dest->job_state = src->job_state;
break;
case ORTE_EXIT_CODE:
dest->exit_code = src->exit_code;
break;
case ORTE_APP_CONTEXT:
if(NULL == src->app_context) {
dest->app_context = NULL;

3
orte/mca/gpr/gpr_types.h
Просмотреть файл

@ -140,8 +140,9 @@ typedef union { /* shared storage for the value */
orte_cellid_t cellid;
orte_node_state_t node_state;
orte_proc_state_t proc_state;
orte_app_context_t *app_context;
orte_job_state_t job_state;
orte_exit_code_t exit_code;
orte_app_context_t *app_context;
} orte_gpr_value_union_t;
/*

4
orte/mca/gpr/replica/functional_layer/gpr_replica_dump_fn.c
Просмотреть файл

@ -807,6 +807,10 @@ void orte_gpr_replica_dump_itagval_value(orte_buffer_t *buffer,
sprintf(tmp, "\t\tData type: ORTE_PROC_STATE\tValue: %d", (int)iptr->value.proc_state);
break;
case ORTE_JOB_STATE:
sprintf(tmp, "\t\tData type: ORTE_JOB_STATE\tValue: %d", (int)iptr->value.job_state);
break;
case ORTE_EXIT_CODE:
sprintf(tmp, "\t\tData type: ORTE_EXIT_CODE\tValue: %d", (int)iptr->value.exit_code);
break;

11
orte/mca/gpr/replica/functional_layer/gpr_replica_messaging_fn.c
Просмотреть файл

@ -136,7 +136,9 @@ CLEANUP:
OBJ_RELEASE(cb);
}
/* cleanup any one-shot triggers that fired */
/* cleanup any one-shot triggers that fired and set processing to
* false on all others
*/
trigs = (orte_gpr_replica_trigger_t**)((orte_gpr_replica.triggers)->addr);
for (i=0, k=0, m=0; k < orte_gpr_replica.num_trigs &&
i < (orte_gpr_replica.triggers)->size; i++) {
@ -146,13 +148,16 @@ CLEANUP:
OBJ_RELEASE(trigs[i]);
orte_pointer_array_set_item(orte_gpr_replica.triggers, i, NULL);
m++;
} else {
trigs[i]->processing = false;
}
}
}
orte_gpr_replica.num_trigs -= m;
/* cleanup any subscriptions that are supposed to be
* removed based on a trigger having fired
* removed based on a trigger having fired - set processing to false
* on all others
*/
subs = (orte_gpr_replica_subscription_t**)(orte_gpr_replica.subscriptions)->addr;
for (i=0, k=0; k < orte_gpr_replica.num_subs &&
@ -173,6 +178,8 @@ CLEANUP:
}
}
}
} else {
subs[i]->processing = false;
}
}
}

24
orte/mca/gpr/replica/functional_layer/gpr_replica_segment_fn.c
Просмотреть файл

@ -422,6 +422,10 @@ int orte_gpr_replica_get_value(void *value, orte_gpr_replica_itagval_t *ival)
*((orte_proc_state_t*)value) = src->proc_state;
break;
case ORTE_JOB_STATE:
*((orte_job_state_t*)value) = src->job_state;
break;
case ORTE_EXIT_CODE:
*((orte_exit_code_t*)value) = src->exit_code;
break;
@ -564,9 +568,7 @@ int orte_gpr_replica_compare_values(int *cmp, orte_gpr_replica_itagval_t *ival1,
*cmp = 1;
}
break;
break;
case ORTE_CELLID:
if (ival1->value.cellid == ival2->value.cellid) {
*cmp = 0;
@ -577,8 +579,6 @@ int orte_gpr_replica_compare_values(int *cmp, orte_gpr_replica_itagval_t *ival1,
}
break;
break;
case ORTE_VPID:
if (ival1->value.vpid == ival2->value.vpid) {
*cmp = 0;
@ -589,8 +589,6 @@ int orte_gpr_replica_compare_values(int *cmp, orte_gpr_replica_itagval_t *ival1,
}
break;
break;
case ORTE_NODE_STATE:
if (ival1->value.node_state == ival2->value.node_state) {
*cmp = 0;
@ -601,8 +599,6 @@ int orte_gpr_replica_compare_values(int *cmp, orte_gpr_replica_itagval_t *ival1,
}
break;
break;
case ORTE_PROC_STATE:
if (ival1->value.proc_state == ival2->value.proc_state) {
*cmp = 0;
@ -613,6 +609,14 @@ int orte_gpr_replica_compare_values(int *cmp, orte_gpr_replica_itagval_t *ival1,
}
break;
case ORTE_JOB_STATE:
if (ival1->value.job_state == ival2->value.job_state) {
*cmp = 0;
} else if (ival1->value.job_state < ival2->value.job_state) {
*cmp = -1;
} else {
*cmp = 1;
}
break;
case ORTE_EXIT_CODE:
@ -625,8 +629,6 @@ int orte_gpr_replica_compare_values(int *cmp, orte_gpr_replica_itagval_t *ival1,
}
break;
break;
case ORTE_NULL:
*cmp = 0;
break;

11
orte/mca/gpr/replica/functional_layer/gpr_replica_trig_ops_fn.c
Просмотреть файл

@ -814,7 +814,7 @@ int orte_gpr_replica_check_events(void)
trigs = (orte_gpr_replica_trigger_t**)((orte_gpr_replica.triggers)->addr);
for (i=0, j=0; j < orte_gpr_replica.num_trigs &&
i < (orte_gpr_replica.triggers)->size; i++) {
if (NULL != trigs[i]) {
if (NULL != trigs[i] && !trigs[i]->processing) {
j++;
/* check the trigger */
if (ORTE_SUCCESS != (rc = orte_gpr_replica_check_trig(trigs[i]))) {
@ -939,6 +939,11 @@ FIRED:
}
}
/* set the processing flag so we don't go into infinite loop if
* any callback functions modify the registry
*/
trig->processing = true;
/* if this trigger was a one-shot, set flag to indicate it has fired
* so it can be cleaned up later
*/
@ -1056,6 +1061,10 @@ int orte_gpr_replica_check_subscription(orte_gpr_replica_subscription_t *sub)
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
/* register that this subscription is being processed
* to avoid potential infinite loops
*/
sub->processing = true;
}
}
}

10
orte/mca/gpr/replica/gpr_replica.h
Просмотреть файл

@ -273,6 +273,11 @@ typedef struct {
char *name;
/* boolean indicating if this subscription is active or not */
bool active;
/* boolean indicating that this subscription is already being
* processed - required to prevent infinite loops should a
* callback function modify the registry
*/
bool processing;
/* boolean indicating that this subscription
* should be removed after processing
* is completed
@ -328,6 +333,11 @@ struct orte_gpr_replica_trigger_t {
orte_gpr_replica_trigger_requestor_t *master;
/* the action that causes the trigger to be fired */
orte_gpr_trigger_action_t action;
/* boolean indicating that this trigger is already being
* processed - required to prevent infinite loops should a
* callback function modify the registry
*/
bool processing;
/* flag that indicates this trigger is a one-shot, has fired and
* now should be cleaned up
*/

2
orte/mca/gpr/replica/gpr_replica_class_instances.h
Просмотреть файл

@ -288,6 +288,7 @@ static void orte_gpr_replica_subscription_construct(orte_gpr_replica_subscriptio
sub->index = 0;
sub->name = NULL;
sub->active = false;
sub->processing = false;
sub->cleanup = false;
sub->action = ORTE_GPR_REPLICA_NO_ACTION;
@ -383,6 +384,7 @@ static void orte_gpr_replica_trigger_construct(orte_gpr_replica_trigger_t* trig)
trig->action = ORTE_GPR_REPLICA_NO_ACTION;
trig->one_shot_fired = false;
trig->processing = false;
trig->num_counters = 0;
orte_pointer_array_init(&(trig->counters), orte_gpr_array_block_size,

2
orte/mca/oob/tcp/oob_tcp.c
Просмотреть файл

@ -974,7 +974,7 @@ int mca_oob_tcp_init(void)
(value->keyvals[1])->type = ORTE_STRING;
(value->keyvals[1])->key = strdup(ORTE_PROC_RML_IP_ADDRESS_KEY);
tmp = mca_oob.oob_get_addr();
tmp2 = strrchr(tmp, '/');
tmp2 = strrchr(tmp, '/') + 1;
tmp3 = strrchr(tmp, ':');
if(NULL == tmp2 || NULL == tmp3) {
opal_output(0, "[%lu,%lu,%lu] mca_oob_tcp_init: invalid address \'%s\' "

132
orte/mca/rmgr/base/rmgr_base_stage_gate.c
Просмотреть файл

@ -23,17 +23,18 @@
#include <string.h>
#include "include/orte_constants.h"
#include "include/orte_types.h"
#include "orte/include/orte_constants.h"
#include "orte/include/orte_types.h"
#include "opal/util/output.h"
#include "dps/dps.h"
#include "mca/gpr/gpr.h"
#include "mca/errmgr/errmgr.h"
#include "mca/rml/rml.h"
#include "orte/dps/dps.h"
#include "orte/mca/gpr/gpr.h"
#include "orte/mca/errmgr/errmgr.h"
#include "orte/mca/rml/rml.h"
#include "orte/mca/soh/soh.h"
#include "mca/rmgr/base/base.h"
#include "orte/mca/rmgr/base/base.h"
int orte_rmgr_base_proc_stage_gate_init(orte_jobid_t job)
@ -375,6 +376,12 @@ int orte_rmgr_base_proc_stage_gate_init(orte_jobid_t job)
if (ORTE_SUCCESS != rc) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
/* set the job state to "launched" */
if (ORTE_SUCCESS != (rc = orte_soh.set_job_soh(job, ORTE_JOB_STATE_LAUNCHED))) {
ORTE_ERROR_LOG(rc);
}
CLEANUP:
@ -394,26 +401,20 @@ void orte_rmgr_base_proc_stage_gate_mgr(orte_gpr_notify_data_t *data,
{
orte_gpr_value_t **values;
orte_gpr_keyval_t **kvals;
orte_process_name_t *recipients;
orte_process_name_t *recipients=NULL;
size_t i, j, m, n=0;
orte_vpid_t k=0;
int rc;
bool found_slots=false, found_start=false;
bool found_stg1=false, found_stg2=false;
bool found_stg3=false, found_finalized=false;
orte_buffer_t msg;
orte_jobid_t job;
char **tokens=NULL;
size_t num_tokens;
/* check to see if this came from one of the stage gates as opposed
* to either terminate or finalize - if the latter, we ignore it
*/
values = (orte_gpr_value_t**)(data->values)->addr;
kvals = values[0]->keyvals;
for (i=0; i < values[0]->cnt; i++) {
if (0 == strcmp(kvals[i]->key, ORTE_PROC_NUM_FINALIZED) ||
0 == strcmp(kvals[i]->key, ORTE_PROC_NUM_TERMINATED)) {
return;
}
}
/* get the jobid from the segment name
* we setup the stage gate triggers to return at least one value
* to us. we use that value to extract the jobid for the returned
@ -426,12 +427,34 @@ void orte_rmgr_base_proc_stage_gate_mgr(orte_gpr_notify_data_t *data,
return;
}
if (ORTE_SUCCESS != (rc = orte_schema.get_job_tokens(&tokens, &num_tokens, job))) {
ORTE_ERROR_LOG(rc);
return;
}
/* check to see if this came from one of the stage gates as opposed
* to either terminate or finalize - if the latter, we set the job
* state as appropriate and then return - no message needs to be
* sent to the processes themselves
*/
kvals = values[0]->keyvals;
for (i=0; i < values[0]->cnt; i++) {
if (0 == strcmp(kvals[i]->key, ORTE_PROC_NUM_TERMINATED)) {
if (ORTE_SUCCESS != (rc = orte_soh.set_job_soh(job, ORTE_JOB_STATE_TERMINATED))) {
ORTE_ERROR_LOG(rc);
}
goto CLEANUP;
}
}
/* value returned will contain the counter, which contains the number of
* procs in this job
* procs in this job. We need to know which counter is included as this
* tells us the job state we have reached.
*/
for (i=0, m=0; m < data->cnt &&
i < (data->values)->size &&
(!found_slots || !found_start); i++) {
(!found_slots || !found_start ||
(!found_stg1 && !found_stg2 && !found_stg3 && !found_finalized)); i++) {
if (NULL != values[i]) {
m++;
kvals = values[i]->keyvals;
@ -439,7 +462,9 @@ void orte_rmgr_base_proc_stage_gate_mgr(orte_gpr_notify_data_t *data,
if (NULL != values[i]->tokens &&
0 == strcmp(ORTE_JOB_GLOBALS, values[i]->tokens[0])) {
/* find the ORTE_JOB_SLOTS_KEY and the ORTE_JOB_VPID_START_KEY keyval */
for (j=0; j < values[i]->cnt && (!found_slots || !found_start); j++) {
for (j=0; j < values[i]->cnt &&
(!found_slots || !found_start ||
(!found_stg1 && !found_stg2 && !found_stg3 && !found_finalized)); j++) {
if (NULL != kvals[j] && !found_slots &&
0 == strcmp(ORTE_JOB_SLOTS_KEY, kvals[j]->key)) {
n = kvals[j]->value.size;
@ -450,6 +475,19 @@ void orte_rmgr_base_proc_stage_gate_mgr(orte_gpr_notify_data_t *data,
k = kvals[j]->value.vpid;
found_start = true;
}
if (NULL != kvals[j] &&
0 == strcmp(ORTE_PROC_NUM_AT_STG1, kvals[j]->key)) {
found_stg1 = true;
} else if (NULL != kvals[j] &&
0 == strcmp(ORTE_PROC_NUM_AT_STG2, kvals[j]->key)) {
found_stg2 = true;
} else if (NULL != kvals[j] &&
0 == strcmp(ORTE_PROC_NUM_AT_STG3, kvals[j]->key)) {
found_stg3 = true;
} else if (NULL != kvals[j] &&
0 == strcmp(ORTE_PROC_NUM_FINALIZED, kvals[j]->key)) {
found_finalized = true;
}
}
}
}
@ -457,13 +495,37 @@ void orte_rmgr_base_proc_stage_gate_mgr(orte_gpr_notify_data_t *data,
if (!found_slots) {
ORTE_ERROR_LOG(ORTE_ERR_GPR_DATA_CORRUPT);
return;
goto CLEANUP;
}
if (!found_start) {
ORTE_ERROR_LOG(ORTE_ERR_GPR_DATA_CORRUPT);
return;
goto CLEANUP;
}
/* set the job state to the appropriate level */
if (found_stg1) {
if (ORTE_SUCCESS != (rc = orte_soh.set_job_soh(job, ORTE_JOB_STATE_AT_STG1))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
} else if (found_stg2) {
if (ORTE_SUCCESS != (rc = orte_soh.set_job_soh(job, ORTE_JOB_STATE_AT_STG2))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
} else if (found_stg3) {
if (ORTE_SUCCESS != (rc = orte_soh.set_job_soh(job, ORTE_JOB_STATE_AT_STG3))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
} else if (found_finalized) {
if (ORTE_SUCCESS != (rc = orte_soh.set_job_soh(job, ORTE_JOB_STATE_FINALIZED))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
}
/* now can generate the list of recipients */
recipients = (orte_process_name_t*)malloc(n * sizeof(orte_process_name_t));
for (i=0; i < n; i++) {
@ -483,17 +545,27 @@ void orte_rmgr_base_proc_stage_gate_mgr(orte_gpr_notify_data_t *data,
if (ORTE_SUCCESS != (rc = orte_dps.pack(&msg, &job, 1, ORTE_JOBID))) {
ORTE_ERROR_LOG(rc);
OBJ_DESTRUCT(&msg);
return;
goto CLEANUP;
}
if (ORTE_SUCCESS != (rc = orte_rml.xcast(orte_process_info.my_name, recipients,
n, &msg, NULL))) {
ORTE_ERROR_LOG(rc);
OBJ_DESTRUCT(&msg);
return;
goto CLEANUP;
}
OBJ_DESTRUCT(&msg);
free(recipients);
CLEANUP:
for (j=0; j < num_tokens; j++) {
free(tokens[j]);
tokens[j] = NULL;
}
if (NULL != tokens) free(tokens);
if (NULL != recipients) free(recipients);
return;
}
void orte_rmgr_base_proc_stage_gate_mgr_abort(orte_gpr_notify_data_t *data,
@ -515,6 +587,12 @@ void orte_rmgr_base_proc_stage_gate_mgr_abort(orte_gpr_notify_data_t *data,
ORTE_ERROR_LOG(rc);
return;
}
/* set the job status to "aborted" */
if (ORTE_SUCCESS != (rc = orte_soh.set_job_soh(job, ORTE_JOB_STATE_ABORTED))) {
ORTE_ERROR_LOG(rc);
}
orte_errmgr.incomplete_start(job);
}

1
orte/mca/schema/base/base.h
Просмотреть файл

@ -56,6 +56,7 @@ extern "C" {
/* general usage functions */
int orte_schema_base_get_proc_tokens(char ***tokens, size_t* num_tokens, orte_process_name_t *proc);
int orte_schema_base_get_node_tokens(char ***tokens, size_t* num_tokens, orte_cellid_t cellid, char *nodename);
int orte_schema_base_get_job_tokens(char ***tokens, size_t* num_tokens, orte_jobid_t jobid);
int orte_schema_base_get_cell_tokens(char ***tokens, size_t* num_tokens, orte_cellid_t cellid);
int orte_schema_base_get_job_segment_name(char **name, orte_jobid_t jobid);
int orte_schema_base_extract_jobid_from_segment_name(orte_jobid_t *jobid, char *name);

32
orte/mca/schema/base/schema_base_fns.c
Просмотреть файл

@ -113,6 +113,38 @@ CLEANUP:
return rc;
}
int orte_schema_base_get_job_tokens(char ***job_tokens, size_t* num_tokens, orte_jobid_t jobid)
{
int rc;
char** tokens;
char* jobid_string;
tokens = (char**)malloc(2 * sizeof(char*));
if (NULL == tokens) {
ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
return ORTE_ERR_OUT_OF_RESOURCE;
}
if (ORTE_SUCCESS != (rc = orte_ns.convert_jobid_to_string(&jobid_string, jobid))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
asprintf(&tokens[0], "%s-%s", ORTE_JOBID_KEY, jobid_string);
free(jobid_string);
tokens[1] = NULL;
*job_tokens = tokens;
if(num_tokens != NULL)
*num_tokens = 1;
return ORTE_SUCCESS;
CLEANUP:
if (NULL != tokens) {
if (NULL != tokens[0]) free(tokens[0]);
free(tokens);
}
return rc;
}
int orte_schema_base_get_cell_tokens(char ***cell_tokens, size_t* num_tokens, orte_cellid_t cellid)
{
int rc;

1
orte/mca/schema/base/schema_base_open.c
Просмотреть файл

@ -39,6 +39,7 @@
OMPI_DECLSPEC orte_schema_base_module_t orte_schema = {
orte_schema_base_get_proc_tokens,
orte_schema_base_get_node_tokens,
orte_schema_base_get_job_tokens,
orte_schema_base_get_cell_tokens,
orte_schema_base_get_job_segment_name,
orte_schema_base_extract_jobid_from_segment_name,

6
orte/mca/schema/schema.h
Просмотреть файл

@ -36,6 +36,11 @@ typedef int (*orte_schema_get_proc_tokens_fn_t)(
size_t* num_tokens,
orte_process_name_t *proc);
typedef int (*orte_schema_get_job_tokens_fn_t)(
char ***tokens,
size_t* num_tokens,
orte_jobid_t jobid);
typedef int (*orte_schema_get_node_tokens_fn_t)(
char ***tokens,
size_t* num_tokens,
@ -68,6 +73,7 @@ typedef int (*orte_schema_get_std_subscription_name_fn_t)(char **name,
struct orte_schema_base_module_1_0_0_t {
orte_schema_get_proc_tokens_fn_t get_proc_tokens;
orte_schema_get_node_tokens_fn_t get_node_tokens;
orte_schema_get_job_tokens_fn_t get_job_tokens;
orte_schema_get_cell_tokens_fn_t get_cell_tokens;
orte_schema_get_job_segment_name_fn_t get_job_segment_name;
orte_schema_extract_jobid_from_segment_name_fn_t extract_jobid_from_segment_name;

2
orte/mca/schema/schema_types.h
Просмотреть файл

@ -40,6 +40,7 @@ extern char *orte_error_strings[];
*/
#define ORTE_JOB_SEGMENT "orte-job"
#define ORTE_NODE_SEGMENT "orte-node"
#define ORTE_JOBINFO_SEGMENT "orte_active_jobs"
#define ORTE_RESOURCE_SEGMENT "orte-resources"
/*
@ -68,6 +69,7 @@ extern char *orte_error_strings[];
#define ORTE_JOB_VPID_START_KEY "orte-job-vpid-start"
#define ORTE_JOB_VPID_RANGE_KEY "orte-job-vpid-range"
#define ORTE_JOB_IOF_KEY "orte-job-iof"
#define ORTE_JOB_STATE_KEY "orte-job-state"
#define ORTE_PROC_NAME_KEY "orte-proc-name"
#define ORTE_PROC_RANK_KEY "orte-proc-rank"
#define ORTE_PROC_PID_KEY "orte-proc-pid"

4
orte/mca/soh/base/Makefile.am
Просмотреть файл

@ -36,8 +36,10 @@ libmca_soh_base_la_SOURCES = \
soh_base_local_functions.c \
soh_base_get_proc_soh.c \
soh_base_set_proc_soh.c \
soh_base_get_job_soh.c \
soh_base_set_job_soh.c \
soh_base_open.c \
data_type_support/soh_data_type_packing_fns.c \
data_type_support/soh_data_type_packing_fns.c \
data_type_support/soh_data_type_unpacking_fns.c

12
orte/mca/soh/base/base.h
Просмотреть файл

@ -60,6 +60,12 @@ int orte_soh_base_set_node_soh_not_available(orte_cellid_t cell,
char *nodename,
orte_node_state_t state);
int orte_soh_base_get_job_soh(orte_job_state_t *state,
orte_jobid_t jobid);
int orte_soh_base_set_job_soh(orte_jobid_t jobid,
orte_job_state_t state);
int orte_soh_base_begin_monitoring_not_available(orte_jobid_t job);
@ -77,6 +83,9 @@ int orte_soh_base_pack_node_state(orte_buffer_t *buffer, void *src,
int orte_soh_base_pack_proc_state(orte_buffer_t *buffer, void *src,
size_t num_vals, orte_data_type_t type);
int orte_soh_base_pack_job_state(orte_buffer_t *buffer, void *src,
size_t num_vals, orte_data_type_t type);
/*
* DATA TYPE UNPACKING FUNCTIONS
*/
@ -89,6 +98,9 @@ int orte_soh_base_unpack_node_state(orte_buffer_t *buffer, void *dest,
int orte_soh_base_unpack_proc_state(orte_buffer_t *buffer, void *dest,
size_t *num_vals, orte_data_type_t type);
int orte_soh_base_unpack_job_state(orte_buffer_t *buffer, void *dest,
size_t *num_vals, orte_data_type_t type);
/*
* globals that might be needed
*/

15
orte/mca/soh/base/data_type_support/soh_data_type_packing_fns.c
Просмотреть файл

@ -70,3 +70,18 @@ int orte_soh_base_pack_proc_state(orte_buffer_t *buffer, void *src,
return rc;
}
/*
* JOB STATE
*/
int orte_soh_base_pack_job_state(orte_buffer_t *buffer, void *src,
size_t num_vals, orte_data_type_t type)
{
int rc;
if (ORTE_SUCCESS != (rc = orte_dps_pack_buffer(buffer, src, num_vals, ORTE_INT8))) {
ORTE_ERROR_LOG(rc);
}
return rc;
}

15
orte/mca/soh/base/data_type_support/soh_data_type_unpacking_fns.c
Просмотреть файл

@ -70,3 +70,18 @@ int orte_soh_base_unpack_proc_state(orte_buffer_t *buffer, void *dest,
return rc;
}
/*
* JOB STATE
*/
int orte_soh_base_unpack_job_state(orte_buffer_t *buffer, void *dest,
size_t *num_vals, orte_data_type_t type)
{
int rc;
if (ORTE_SUCCESS != (rc = orte_dps_unpack_buffer(buffer, dest, num_vals, ORTE_INT8))) {
ORTE_ERROR_LOG(rc);
}
return rc;
}

90
orte/mca/soh/base/soh_base_get_job_soh.c Обычный файл
Просмотреть файл

@ -0,0 +1,90 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University.
* All rights reserved.
* Copyright (c) 2004-2005 The Trustees of the University of Tennessee.
* All rights reserved.
* Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
* University of Stuttgart. All rights reserved.
* Copyright (c) 2004-2005 The Regents of the University of California.
* All rights reserved.
* $COPYRIGHT$
*
* Additional copyrights may follow
*
* $HEADER$
*/
/** @file:
*/
/*
* includes
*/
#include "orte_config.h"
#include <string.h>
#include "mca/schema/schema.h"
#include "mca/errmgr/errmgr.h"
#include "mca/gpr/gpr.h"
#include "mca/ns/ns.h"
#include "mca/soh/base/base.h"
int orte_soh_base_get_job_soh(orte_job_state_t *state,
orte_jobid_t jobid)
{
orte_gpr_value_t **values;
orte_gpr_keyval_t **keyvals;
int rc;
size_t cnt, num_tokens, i, j;
char **tokens, *keys[2];
if (ORTE_SUCCESS != (rc = orte_schema.get_job_tokens(&tokens, &num_tokens, jobid))) {
ORTE_ERROR_LOG(rc);
return rc;
}
keys[0] = strdup(ORTE_JOB_STATE_KEY);
keys[1] = NULL;
if (ORTE_SUCCESS != (rc = orte_gpr.get(ORTE_GPR_TOKENS_XAND, ORTE_JOBINFO_SEGMENT,
tokens, keys, &cnt, &values))) {
ORTE_ERROR_LOG(rc);
goto CLEANUP;
}
for (i=0; i < cnt; i++) {
keyvals = values[i]->keyvals;
if (NULL != keyvals) {
for (j=0; j < values[i]->cnt; j++) {
if (ORTE_JOB_STATE == keyvals[j]->type) {
*state = keyvals[j]->value.job_state;
goto CLEANUP;
} else {
ORTE_ERROR_LOG(ORTE_ERR_GPR_DATA_CORRUPT);
rc = ORTE_ERR_GPR_DATA_CORRUPT;
}
}
}
}
CLEANUP:
for (i=0; i < 2; i++) {
if (NULL != keys[i]) free(keys[i]);
}
for (i=0; i < num_tokens; i++) {
if (NULL != tokens[i]) free(tokens[i]);
}
free(tokens);
if (NULL != values) {
for (i=0; i < cnt; i++) {
OBJ_RELEASE(values[i]);
}
free(values);
}
return rc;
}

10
orte/mca/soh/base/soh_base_open.c
Просмотреть файл

@ -56,6 +56,8 @@ orte_soh_base_module_t orte_soh = {
orte_soh_base_set_proc_soh,
orte_soh_base_get_node_soh_not_available,
orte_soh_base_set_node_soh_not_available,
orte_soh_base_get_job_soh,
orte_soh_base_set_job_soh,
orte_soh_base_begin_monitoring_not_available,
orte_soh_base_module_finalize_not_available
};
@ -101,6 +103,14 @@ int orte_soh_base_open(void)
return rc;
}
tmp = ORTE_JOB_STATE;
if (ORTE_SUCCESS != (rc = orte_dps.register_type(orte_soh_base_pack_job_state,
orte_soh_base_unpack_job_state,
"ORTE_JOB_STATE", &tmp))) {
ORTE_ERROR_LOG(rc);
return rc;
}
tmp = ORTE_EXIT_CODE;
if (ORTE_SUCCESS != (rc = orte_dps.register_type(orte_soh_base_pack_exit_code,
orte_soh_base_unpack_exit_code,

84
orte/mca/soh/base/soh_base_set_job_soh.c Обычный файл
Просмотреть файл

@ -0,0 +1,84 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University.
* All rights reserved.
* Copyright (c) 2004-2005 The Trustees of the University of Tennessee.
* All rights reserved.
* Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
* University of Stuttgart. All rights reserved.
* Copyright (c) 2004-2005 The Regents of the University of California.
* All rights reserved.
* $COPYRIGHT$
*
* Additional copyrights may follow
*
* $HEADER$
*/
/** @file:
*/
/*
* includes
*/
#include "orte_config.h"
#include <string.h>
#include "mca/schema/schema.h"
#include "mca/errmgr/errmgr.h"
#include "mca/gpr/gpr.h"
#include "mca/ns/ns.h"
#include "mca/soh/base/base.h"
int orte_soh_base_set_job_soh(orte_jobid_t jobid,
orte_job_state_t state)
{
orte_gpr_value_t *value;
int rc;
value = OBJ_NEW(orte_gpr_value_t);
if (NULL == value) {
ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
return ORTE_ERR_OUT_OF_RESOURCE;
}
value->addr_mode = ORTE_GPR_OVERWRITE | ORTE_GPR_TOKENS_AND;
value->segment = strdup(ORTE_JOBINFO_SEGMENT);
if (ORTE_JOBID_MAX == jobid) { /* wildcard case - set everyone on job segment to this status */
value->tokens = NULL;
} else {
if (ORTE_SUCCESS != (rc = orte_schema.get_job_tokens(&(value->tokens),
&(value->num_tokens), jobid))) {
ORTE_ERROR_LOG(rc);
return rc;
}
}
value->cnt = 1;
value->keyvals = (orte_gpr_keyval_t**)malloc(value->cnt * sizeof(orte_gpr_keyval_t*));
if (NULL == value->keyvals) {
ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
OBJ_RELEASE(value);
return ORTE_ERR_OUT_OF_RESOURCE;
}
value->keyvals[0] = OBJ_NEW(orte_gpr_keyval_t);
if (NULL == value->keyvals[0]) {
ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
OBJ_RELEASE(value);
return ORTE_ERR_OUT_OF_RESOURCE;
}
(value->keyvals[0])->key = strdup(ORTE_JOB_STATE_KEY);
(value->keyvals[0])->type = ORTE_JOB_STATE;
(value->keyvals[0])->value.job_state = state;
if (ORTE_SUCCESS != (rc = orte_gpr.put(1, &value))) {
ORTE_ERROR_LOG(rc);
}
OBJ_RELEASE(value);
/* all done */
return rc;
}

2
orte/mca/soh/bproc/soh_bproc.c
Просмотреть файл

@ -340,6 +340,8 @@ orte_soh_base_module_t orte_soh_bproc_module = {
orte_soh_bproc_set_proc_soh,
orte_soh_base_get_node_soh_not_available,
orte_soh_base_set_node_soh_not_available,
orte_soh_base_get_job_soh_not_available,
orte_soh_base_set_job_soh_not_available,
orte_soh_base_begin_monitoring_not_available,
orte_soh_bproc_finalize
};

14
orte/mca/soh/soh.h
Просмотреть файл

@ -65,6 +65,18 @@ typedef int (*orte_soh_base_module_set_node_soh_fn_t)(orte_cellid_t cell,
char *nodename,
orte_node_state_t state);
/*
* Query the state-of-health of a job
*/
typedef int (*orte_soh_base_module_get_job_soh_fn_t)(orte_job_state_t *state,
orte_jobid_t jobid);
/*
* Set the state-of-health of a job
*/
typedef int (*orte_soh_base_module_set_job_soh_fn_t)(orte_jobid_t jobid,
orte_job_state_t state);
/*
* Initiate monitoring of a job
* This function notifies the soh that it should initiate monitoring of the specified
@ -92,6 +104,8 @@ struct orte_soh_base_module_1_0_0_t {
orte_soh_base_module_set_proc_soh_fn_t set_proc_soh;
orte_soh_base_module_get_node_soh_fn_t get_node_soh;
orte_soh_base_module_set_node_soh_fn_t set_node_soh;
orte_soh_base_module_get_job_soh_fn_t get_job_soh;
orte_soh_base_module_set_job_soh_fn_t set_job_soh;
orte_soh_base_module_begin_monitoring_fn_t begin_monitoring_job;
orte_soh_base_module_finalize_fn_t finalize;
};

18
orte/mca/soh/soh_types.h
Просмотреть файл

@ -41,8 +41,24 @@ typedef int8_t orte_proc_state_t;
#define ORTE_PROC_STATE_TERMINATED 0x09 /* process has terminated and is no longer running */
#define ORTE_PROC_STATE_ABORTED 0x0A /* process aborted */
/*
* Job state codes
*/
typedef int8_t orte_job_state_t;
#define ORTE_JOB_STATE_INIT 0x01 /* job entry has been created by rmaps */
#define ORTE_JOB_STATE_LAUNCHED 0x02 /* job has been launched by pls */
#define ORTE_JOB_STATE_AT_STG1 0x03 /* all processes are at Stage Gate 1 barrier in orte_init */
#define ORTE_JOB_STATE_AT_STG2 0x04 /* all processes are at Stage Gate 2 barrier in orte_init */
#define ORTE_JOB_STATE_RUNNING 0x06 /* all processes have exited orte_init and is running */
#define ORTE_JOB_STATE_AT_STG3 0x07 /* all processes are at Stage Gate 3 barrier in orte_finalize */
#define ORTE_JOB_STATE_FINALIZED 0x08 /* all processes have completed orte_finalize and is running */
#define ORTE_JOB_STATE_TERMINATED 0x09 /* all processes have terminated and is no longer running */
#define ORTE_JOB_STATE_ABORTED 0x0A /* at least one process aborted, causing job to abort */
/**
* Node State, correspondinding to the ORTE_NODE_STATE_* #defines,
* Node State, corresponding to the ORTE_NODE_STATE_* #defines,
* below. These are #defines instead of an enum because the thought
* is that we may have lots and lots of entries of these in the
* registry and by making this an int8_t, it's only 1 byte, whereas an

17
orte/runtime/orte_init_stage1.c
Просмотреть файл

@ -291,18 +291,33 @@ int orte_init_stage1(void)
free(contact_path);
}
/* open/load rmgr/soh */
/*
* setup the resource manager
*/
if (ORTE_SUCCESS != (ret = orte_rmgr_base_open())) {
ORTE_ERROR_LOG(ret);
return ret;
}
if (ORTE_SUCCESS != (ret = orte_rmgr_base_select())) {
ORTE_ERROR_LOG(ret);
return ret;
}
/*
* setup the state-of-health monitor
*/
if (ORTE_SUCCESS != (ret = orte_soh_base_open())) {
ORTE_ERROR_LOG(ret);
return ret;
}
if (ORTE_SUCCESS != (ret = orte_soh_base_select())) {
ORTE_ERROR_LOG(ret);
return ret;
}
/* if we are a singleton or the seed, setup the infrastructure for our job */
if(orte_process_info.singleton || orte_process_info.seed) {

17
orte/runtime/orte_init_stage2.c
Просмотреть файл

@ -55,23 +55,6 @@ int orte_init_stage2(void)
return ret;
}
/*
* setup the resource manager
*/
if (ORTE_SUCCESS != (ret = orte_rmgr_base_select())) {
ORTE_ERROR_LOG(ret);
return ret;
}
/*
* setup the state-of-health monitor
*/
if (ORTE_SUCCESS != (ret = orte_soh_base_select())) {
ORTE_ERROR_LOG(ret);
return ret;
}
/*
* setup I/O forwarding system
*/