opal/mca/pmix/s1/pmix

/* [<][>][^][v][top][bottom][index][help] */
This source file includes following definitions.
kvs_get
kvs_put
s1_init
s1_fini
s1_initialized
s1_abort
s1_spawn
s1_put
s1_commit
fencenb
s1_fencenb
fence_release
s1_fence
s1_get
s1_publish
s1_lookup
s1_unpublish
s1_job_connect
s1_job_disconnect
s1_store_local
s1_get_nspace
s1_register_jobid
pmix_error
   1 /* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
   2 /*
   3  * Copyright (c) 2014-2018 Intel, Inc. All rights reserved.
   4  * Copyright (c) 2014-2016 Research Organization for Information Science
   5  *                         and Technology (RIST). All rights reserved.
   6  * Copyright (c) 2016 Cisco Systems, Inc.  All rights reserved.
   7  * Copyright (c) 2017      Los Alamos National Security, LLC. All
   8  *                         rights reserved.
   9  * $COPYRIGHT$
  10  *
  11  * Additional copyrights may follow
  12  *
  13  * $HEADER$
  14  */
  15 
  16 #include "opal_config.h"
  17 #include "opal/constants.h"
  18 #include "opal/types.h"
  19 
  20 #include "opal_stdint.h"
  21 #include "opal/mca/hwloc/base/base.h"
  22 #include "opal/util/argv.h"
  23 #include "opal/util/opal_environ.h"
  24 #include "opal/util/output.h"
  25 #include "opal/util/proc.h"
  26 #include "opal/util/show_help.h"
  27 
  28 #include <string.h>
  29 #include <pmi.h>
  30 
  31 #include "opal/mca/pmix/base/base.h"
  32 #include "opal/mca/pmix/base/pmix_base_fns.h"
  33 #include "opal/mca/pmix/base/pmix_base_hash.h"
  34 #include "pmix_s1.h"
  35 
  36 static int s1_init(opal_list_t *ilist);
  37 static int s1_fini(void);
  38 static int s1_initialized(void);
  39 static int s1_abort(int flag, const char msg[],
  40                     opal_list_t *procs);
  41 static int s1_commit(void);
  42 static int s1_fencenb(opal_list_t *procs, int collect_data,
  43                       opal_pmix_op_cbfunc_t cbfunc, void *cbdata);
  44 static int s1_fence(opal_list_t *procs, int collect_data);
  45 static int s1_put(opal_pmix_scope_t scope,
  46                   opal_value_t *kv);
  47 static int s1_get(const opal_process_name_t *id,
  48                   const char *key, opal_list_t *info,
  49                   opal_value_t **kv);
  50 static int s1_publish(opal_list_t *info);
  51 static int s1_lookup(opal_list_t *data, opal_list_t *info);
  52 static int s1_unpublish(char **keys, opal_list_t *info);
  53 static int s1_spawn(opal_list_t *jobinfo, opal_list_t *apps, opal_jobid_t *jobid);
  54 static int s1_job_connect(opal_list_t *procs);
  55 static int s1_job_disconnect(opal_list_t *procs);
  56 static int s1_store_local(const opal_process_name_t *proc,
  57                           opal_value_t *val);
  58 static const char *s1_get_nspace(opal_jobid_t jobid);
  59 static void s1_register_jobid(opal_jobid_t jobid, const char *nspace);
  60 
  61 const opal_pmix_base_module_t opal_pmix_s1_module = {
  62     .init = s1_init,
  63     .finalize = s1_fini,
  64     .initialized = s1_initialized,
  65     .abort = s1_abort,
  66     .commit = s1_commit,
  67     .fence_nb = s1_fencenb,
  68     .fence = s1_fence,
  69     .put = s1_put,
  70     .get = s1_get,
  71     .publish = s1_publish,
  72     .lookup = s1_lookup,
  73     .unpublish = s1_unpublish,
  74     .spawn = s1_spawn,
  75     .connect = s1_job_connect,
  76     .disconnect = s1_job_disconnect,
  77     .register_evhandler = opal_pmix_base_register_handler,
  78     .deregister_evhandler = opal_pmix_base_deregister_handler,
  79     .store_local = s1_store_local,
  80     .get_nspace = s1_get_nspace,
  81     .register_jobid = s1_register_jobid
  82 };
  83 
  84 // usage accounting
  85 static int pmix_init_count = 0;
  86 
  87 // local object
  88 typedef struct {
  89     opal_object_t super;
  90     opal_event_t ev;
  91     opal_pmix_op_cbfunc_t opcbfunc;
  92     void *cbdata;
  93 } pmi_opcaddy_t;
  94 static OBJ_CLASS_INSTANCE(pmi_opcaddy_t,
  95                           opal_object_t,
  96                           NULL, NULL);
  97 
  98 // PMI constant values:
  99 static int pmix_kvslen_max = 0;
 100 static int pmix_keylen_max = 0;
 101 static int pmix_vallen_max = 0;
 102 static int pmix_vallen_threshold = INT_MAX;
 103 
 104 // Job environment description
 105 static char *pmix_kvs_name = NULL;
 106 static bool s1_committed = false;
 107 static char* pmix_packed_data = NULL;
 108 static int pmix_packed_data_offset = 0;
 109 static char* pmix_packed_encoded_data = NULL;
 110 static int pmix_packed_encoded_data_offset = 0;
 111 static int pmix_pack_key = 0;
 112 static opal_process_name_t s1_pname;
 113 static int *lranks = NULL, nlranks;
 114 static bool got_modex_data = false;
 115 
 116 static char* pmix_error(int pmix_err);
 117 #define OPAL_PMI_ERROR(pmi_err, pmi_func)                       \
 118     do {                                                        \
 119         opal_output(0, "%s [%s:%d:%s]: %s\n",                   \
 120             pmi_func, __FILE__, __LINE__, __func__,     \
 121             pmix_error(pmi_err));                        \
 122     } while(0);
 123 
 124 static int kvs_get(const char key[], char value [], int maxvalue)
 125 {
 126     int rc;
 127     rc = PMI_KVS_Get(pmix_kvs_name, key, value, maxvalue);
 128     if( PMI_SUCCESS != rc ){
 129         /* silently return an error - might be okay */
 130         return OPAL_ERROR;
 131     }
 132     return OPAL_SUCCESS;
 133 }
 134 
 135 static int kvs_put(const char key[], const char value[])
 136 {
 137     int rc;
 138     rc = PMI_KVS_Put(pmix_kvs_name, key, value);
 139     if( PMI_SUCCESS != rc ){
 140         OPAL_PMI_ERROR(rc, "PMI_KVS_Put");
 141         return OPAL_ERROR;
 142     }
 143     return rc;
 144 }
 145 
 146 static int s1_init(opal_list_t *ilist)
 147 {
 148     PMI_BOOL initialized;
 149     int spawned;
 150     int rc, ret = OPAL_ERROR;
 151     int i, rank, lrank, nrank;
 152     char *pmix_id, tmp[64];
 153     opal_value_t kv;
 154     char *str;
 155     uint32_t ui32;
 156     opal_process_name_t ldr;
 157     char **localranks=NULL;
 158     opal_process_name_t wildcard_rank;
 159 
 160     if (0 < pmix_init_count) {
 161         ++pmix_init_count;
 162         return OPAL_SUCCESS;
 163     }
 164 
 165     if (PMI_SUCCESS != (rc = PMI_Initialized(&initialized))) {
 166         OPAL_PMI_ERROR(rc, "PMI_Initialized");
 167         return OPAL_ERROR;
 168     }
 169 
 170     if (PMI_TRUE != initialized && PMI_SUCCESS != (rc = PMI_Init(&spawned))) {
 171         OPAL_PMI_ERROR(rc, "PMI_Init");
 172         return OPAL_ERROR;
 173     }
 174 
 175     // setup hash table
 176     opal_pmix_base_hash_init();
 177 
 178     // Initialize space demands
 179     rc = PMI_KVS_Get_value_length_max(&pmix_vallen_max);
 180     if (PMI_SUCCESS != rc) {
 181         OPAL_PMI_ERROR(rc, "PMI_KVS_Get_value_length_max");
 182         goto err_exit;
 183     }
 184     pmix_vallen_threshold = pmix_vallen_max * 3;
 185     pmix_vallen_threshold >>= 2;
 186 
 187     rc = PMI_KVS_Get_name_length_max(&pmix_kvslen_max);
 188     if (PMI_SUCCESS != rc) {
 189         OPAL_PMI_ERROR(rc, "PMI_KVS_Get_name_length_max");
 190         goto err_exit;
 191     }
 192 
 193     rc = PMI_KVS_Get_key_length_max(&pmix_keylen_max);
 194     if (PMI_SUCCESS != rc) {
 195         OPAL_PMI_ERROR(rc, "PMI_KVS_Get_key_length_max");
 196         goto err_exit;
 197     }
 198 
 199     // Initialize job environment information
 200     pmix_id = (char*)malloc(pmix_vallen_max);
 201     if (pmix_id == NULL) {
 202         ret = OPAL_ERR_OUT_OF_RESOURCE;
 203         goto err_exit;
 204     }
 205     /* Get domain id */
 206     if (PMI_SUCCESS != (rc = PMI_Get_kvs_domain_id(pmix_id, pmix_vallen_max))) {
 207         free(pmix_id);
 208         goto err_exit;
 209     }
 210 
 211     /* get our rank */
 212     ret = PMI_Get_rank(&rank);
 213     if( PMI_SUCCESS != ret ) {
 214         OPAL_PMI_ERROR(ret, "PMI_Get_rank");
 215         free(pmix_id);
 216         goto err_exit;
 217     }
 218 
 219     /* Slurm PMI provides the job id as an integer followed
 220      * by a '.', followed by essentially a stepid. The first integer
 221      * defines an overall job number. The second integer is the number of
 222      * individual jobs we have run within that allocation. */
 223     s1_pname.jobid = strtoul(pmix_id, &str, 10);
 224     s1_pname.jobid = (s1_pname.jobid << 16) & 0xffff0000;
 225     if (NULL != str) {
 226         ui32 = strtoul(str+1, NULL, 10);
 227         s1_pname.jobid |= (ui32 & 0x0000ffff);
 228     }
 229     ldr.jobid = s1_pname.jobid;
 230     s1_pname.vpid = rank;
 231     /* store our name in the opal_proc_t so that
 232      * debug messages will make sense - an upper
 233      * layer will eventually overwrite it, but that
 234      * won't do any harm */
 235     opal_proc_set_name(&s1_pname);
 236     opal_output_verbose(2, opal_pmix_base_framework.framework_output,
 237                         "%s pmix:s1: assigned tmp name",
 238                         OPAL_NAME_PRINT(s1_pname));
 239 
 240     /* setup wildcard rank*/
 241     wildcard_rank = OPAL_PROC_MY_NAME;
 242     wildcard_rank.vpid = OPAL_VPID_WILDCARD;
 243 
 244     OBJ_CONSTRUCT(&kv, opal_value_t);
 245     kv.key = strdup(OPAL_PMIX_JOBID);
 246     kv.type = OPAL_UINT32;
 247     kv.data.uint32 = s1_pname.jobid;
 248     if (OPAL_SUCCESS != (ret = opal_pmix_base_store(&wildcard_rank, &kv))) {
 249         OPAL_ERROR_LOG(ret);
 250         OBJ_DESTRUCT(&kv);
 251         goto err_exit;
 252     }
 253     OBJ_DESTRUCT(&kv);
 254 
 255     /* save it */
 256     OBJ_CONSTRUCT(&kv, opal_value_t);
 257     kv.key = strdup(OPAL_PMIX_RANK);
 258     kv.type = OPAL_UINT32;
 259     kv.data.uint32 = rank;
 260     if (OPAL_SUCCESS != (ret = opal_pmix_base_store(&OPAL_PROC_MY_NAME, &kv))) {
 261         OPAL_ERROR_LOG(ret);
 262         OBJ_DESTRUCT(&kv);
 263         goto err_exit;
 264     }
 265     OBJ_DESTRUCT(&kv);
 266 
 267     pmix_kvs_name = (char*)malloc(pmix_kvslen_max);
 268     if (pmix_kvs_name == NULL) {
 269         ret = OPAL_ERR_OUT_OF_RESOURCE;
 270         goto err_exit;
 271     }
 272 
 273     rc = PMI_KVS_Get_my_name(pmix_kvs_name, pmix_kvslen_max);
 274     if (PMI_SUCCESS != rc) {
 275         OPAL_PMI_ERROR(rc, "PMI_KVS_Get_my_name");
 276         goto err_exit;
 277     }
 278 
 279     /* get our local proc info to find our local rank */
 280     if (PMI_SUCCESS != (rc = PMI_Get_clique_size(&nlranks))) {
 281         OPAL_PMI_ERROR(rc, "PMI_Get_clique_size");
 282         return rc;
 283     }
 284     /* save the local size */
 285     OBJ_CONSTRUCT(&kv, opal_value_t);
 286     kv.key = strdup(OPAL_PMIX_LOCAL_SIZE);
 287     kv.type = OPAL_UINT32;
 288     kv.data.uint32 = nlranks;
 289     if (OPAL_SUCCESS != (ret = opal_pmix_base_store(&wildcard_rank, &kv))) {
 290         OPAL_ERROR_LOG(ret);
 291         OBJ_DESTRUCT(&kv);
 292         goto err_exit;
 293     }
 294     OBJ_DESTRUCT(&kv);
 295     lrank = 0;
 296     nrank = 0;
 297     ldr.vpid = rank;
 298     if (0 < nlranks) {
 299         /* now get the specific ranks */
 300         lranks = (int*)calloc(nlranks, sizeof(int));
 301         if (NULL == lranks) {
 302             rc = OPAL_ERR_OUT_OF_RESOURCE;
 303             OPAL_ERROR_LOG(rc);
 304             return rc;
 305         }
 306         if (PMI_SUCCESS != (rc = PMI_Get_clique_ranks(lranks, nlranks))) {
 307             OPAL_PMI_ERROR(rc, "PMI_Get_clique_ranks");
 308             free(lranks);
 309             return rc;
 310         }
 311         /* note the local ldr */
 312         ldr.vpid = lranks[0];
 313         /* save this */
 314         memset(tmp, 0, 64);
 315         for (i=0; i < nlranks; i++) {
 316             (void)snprintf(tmp, 64, "%d", lranks[i]);
 317             opal_argv_append_nosize(&localranks, tmp);
 318             if (rank == lranks[i]) {
 319                 lrank = i;
 320                 nrank = i;
 321             }
 322         }
 323         str = opal_argv_join(localranks, ',');
 324         opal_argv_free(localranks);
 325         OBJ_CONSTRUCT(&kv, opal_value_t);
 326         kv.key = strdup(OPAL_PMIX_LOCAL_PEERS);
 327         kv.type = OPAL_STRING;
 328         kv.data.string = str;
 329         if (OPAL_SUCCESS != (ret = opal_pmix_base_store(&wildcard_rank, &kv))) {
 330             OPAL_ERROR_LOG(ret);
 331             OBJ_DESTRUCT(&kv);
 332             goto err_exit;
 333         }
 334         OBJ_DESTRUCT(&kv);
 335     }
 336 
 337     /* save the local leader */
 338     OBJ_CONSTRUCT(&kv, opal_value_t);
 339     kv.key = strdup(OPAL_PMIX_LOCALLDR);
 340     kv.type = OPAL_UINT64;
 341     kv.data.uint64 = *(uint64_t*)&ldr;
 342     if (OPAL_SUCCESS != (ret = opal_pmix_base_store(&wildcard_rank, &kv))) {
 343         OPAL_ERROR_LOG(ret);
 344         OBJ_DESTRUCT(&kv);
 345         goto err_exit;
 346     }
 347     OBJ_DESTRUCT(&kv);
 348     /* save our local rank */
 349     OBJ_CONSTRUCT(&kv, opal_value_t);
 350     kv.key = strdup(OPAL_PMIX_LOCAL_RANK);
 351     kv.type = OPAL_UINT16;
 352     kv.data.uint16 = lrank;
 353     if (OPAL_SUCCESS != (ret = opal_pmix_base_store(&OPAL_PROC_MY_NAME, &kv))) {
 354         OPAL_ERROR_LOG(ret);
 355         OBJ_DESTRUCT(&kv);
 356         goto err_exit;
 357     }
 358     OBJ_DESTRUCT(&kv);
 359     /* and our node rank */
 360     OBJ_CONSTRUCT(&kv, opal_value_t);
 361     kv.key = strdup(OPAL_PMIX_NODE_RANK);
 362     kv.type = OPAL_UINT16;
 363     kv.data.uint16 = nrank;
 364     if (OPAL_SUCCESS != (ret = opal_pmix_base_store(&OPAL_PROC_MY_NAME, &kv))) {
 365         OPAL_ERROR_LOG(ret);
 366         OBJ_DESTRUCT(&kv);
 367         goto err_exit;
 368     }
 369     OBJ_DESTRUCT(&kv);
 370 
 371     /* get universe size */
 372     ret = PMI_Get_universe_size(&i);
 373     if (PMI_SUCCESS != ret) {
 374         OPAL_PMI_ERROR(ret, "PMI_Get_universe_size");
 375         goto err_exit;
 376     }
 377     /* push this into the dstore for subsequent fetches */
 378     OBJ_CONSTRUCT(&kv, opal_value_t);
 379     kv.key = strdup(OPAL_PMIX_UNIV_SIZE);
 380     kv.type = OPAL_UINT32;
 381     kv.data.uint32 = i;
 382     if (OPAL_SUCCESS != (ret = opal_pmix_base_store(&wildcard_rank, &kv))) {
 383         OPAL_ERROR_LOG(ret);
 384         OBJ_DESTRUCT(&kv);
 385         goto err_exit;
 386     }
 387     OBJ_DESTRUCT(&kv);
 388     /* push this into the dstore for subsequent fetches */
 389     OBJ_CONSTRUCT(&kv, opal_value_t);
 390     kv.key = strdup(OPAL_PMIX_MAX_PROCS);
 391     kv.type = OPAL_UINT32;
 392     kv.data.uint32 = i;
 393     if (OPAL_SUCCESS != (ret = opal_pmix_base_store(&wildcard_rank, &kv))) {
 394         OPAL_ERROR_LOG(ret);
 395         OBJ_DESTRUCT(&kv);
 396         goto err_exit;
 397     }
 398     OBJ_DESTRUCT(&kv);
 399 
 400 
 401     /* get job size */
 402     ret = PMI_Get_size(&i);
 403     if (PMI_SUCCESS != ret) {
 404         OPAL_PMI_ERROR(ret, "PMI_Get_size");
 405         goto err_exit;
 406     }
 407     OBJ_CONSTRUCT(&kv, opal_value_t);
 408     kv.key = strdup(OPAL_PMIX_JOB_SIZE);
 409     kv.type = OPAL_UINT32;
 410     kv.data.uint32 = i;
 411     if (OPAL_SUCCESS != (ret = opal_pmix_base_store(&wildcard_rank, &kv))) {
 412         OPAL_ERROR_LOG(ret);
 413         OBJ_DESTRUCT(&kv);
 414         goto err_exit;
 415     }
 416     OBJ_DESTRUCT(&kv);
 417 
 418     /* get appnum */
 419     ret = PMI_Get_appnum(&i);
 420     if (PMI_SUCCESS != ret) {
 421         OPAL_PMI_ERROR(ret, "PMI_Get_appnum");
 422         goto err_exit;
 423     }
 424     OBJ_CONSTRUCT(&kv, opal_value_t);
 425     kv.key = strdup(OPAL_PMIX_APPNUM);
 426     kv.type = OPAL_UINT32;
 427     kv.data.uint32 = i;
 428     if (OPAL_SUCCESS != (ret = opal_pmix_base_store(&OPAL_PROC_MY_NAME, &kv))) {
 429         OPAL_ERROR_LOG(ret);
 430         OBJ_DESTRUCT(&kv);
 431         goto err_exit;
 432     }
 433     OBJ_DESTRUCT(&kv);
 434 
 435     /* increment the init count */
 436     ++pmix_init_count;
 437 
 438     return OPAL_SUCCESS;
 439 
 440 err_exit:
 441     PMI_Finalize();
 442     return ret;
 443 }
 444 
 445 static int s1_fini(void) {
 446     if (0 == pmix_init_count) {
 447         return OPAL_SUCCESS;
 448     }
 449 
 450     if (0 == --pmix_init_count) {
 451         PMI_Finalize ();
 452         // teardown hash table
 453         opal_pmix_base_hash_finalize();
 454     }
 455 
 456     return OPAL_SUCCESS;
 457 }
 458 
 459 static int s1_initialized(void)
 460 {
 461     if (0 < pmix_init_count) {
 462         return 1;
 463     }
 464     return 0;
 465 }
 466 
 467 static int s1_abort(int flag, const char msg[],
 468                     opal_list_t *procs)
 469 {
 470     PMI_Abort(flag, msg);
 471     return OPAL_SUCCESS;
 472 }
 473 
 474 static int s1_spawn(opal_list_t *jobinfo, opal_list_t *apps, opal_jobid_t *jobid)
 475 {
 476     /*
 477     int rc;
 478     size_t preput_vector_size;
 479     const int info_keyval_sizes[1];
 480     info_keyval_sizes[0] = (int)opal_list_get_size(info_keyval_vector);
 481     //FIXME what's the size of array of lists?
 482     preput_vector_size = opal_list_get_size(preput_keyval_vector);
 483     rc = PMI_Spawn_multiple(count, cmds, argcs, argvs, maxprocs, info_keyval_sizes, info_keyval_vector, (int)preput_vector_size, preput_keyval_vector);
 484     if( PMI_SUCCESS != rc ) {
 485         OPAL_PMI_ERROR(rc, "PMI_Spawn_multiple");
 486         return OPAL_ERROR;
 487     }*/
 488     return OPAL_ERR_NOT_IMPLEMENTED;
 489 }
 490 
 491 static int s1_put(opal_pmix_scope_t scope,
 492                   opal_value_t *kv)
 493 {
 494     int rc;
 495 
 496     opal_output_verbose(2, opal_pmix_base_framework.framework_output,
 497                         "%s pmix:s1 put for key %s",
 498                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME), kv->key);
 499 
 500     if (OPAL_SUCCESS != (rc = opal_pmix_base_store_encoded (kv->key, (void*)&kv->data, kv->type, &pmix_packed_data, &pmix_packed_data_offset))) {
 501         OPAL_ERROR_LOG(rc);
 502         return rc;
 503     }
 504 
 505     if (pmix_packed_data_offset == 0) {
 506         /* nothing to write */
 507         return OPAL_SUCCESS;
 508     }
 509 
 510     if (((pmix_packed_data_offset/3)*4) + pmix_packed_encoded_data_offset < pmix_vallen_max) {
 511         /* this meta-key is still being filled,
 512          * nothing to put yet
 513          */
 514         return OPAL_SUCCESS;
 515     }
 516 
 517     rc = opal_pmix_base_partial_commit_packed (&pmix_packed_data, &pmix_packed_data_offset,
 518                                                &pmix_packed_encoded_data, &pmix_packed_encoded_data_offset,
 519                                                pmix_vallen_max, &pmix_pack_key, kvs_put);
 520 
 521     s1_committed = false;
 522     return rc;
 523 }
 524 
 525 static int s1_commit(void)
 526 {
 527     int rc;
 528 
 529     /* check if there is partially filled meta key and put them */
 530     opal_pmix_base_commit_packed (&pmix_packed_data, &pmix_packed_data_offset,
 531                                   &pmix_packed_encoded_data, &pmix_packed_encoded_data_offset,
 532                                   pmix_vallen_max, &pmix_pack_key, kvs_put);
 533 
 534     if (PMI_SUCCESS != (rc = PMI_KVS_Commit(pmix_kvs_name))) {
 535         OPAL_PMI_ERROR(rc, "PMI_KVS_Commit");
 536         return OPAL_ERROR;
 537     }
 538     return OPAL_SUCCESS;
 539 }
 540 
 541 static void fencenb(int sd, short args, void *cbdata)
 542 {
 543     pmi_opcaddy_t *op = (pmi_opcaddy_t*)cbdata;
 544     int rc = OPAL_SUCCESS;
 545     int32_t i;
 546     opal_value_t *kp, kvn;
 547     opal_hwloc_locality_t locality;
 548     opal_process_name_t s1_pname;
 549 
 550     opal_output_verbose(2, opal_pmix_base_framework.framework_output,
 551                         "%s pmix:s1 called fence",
 552                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));
 553 
 554     /* use the PMI barrier function */
 555     if (PMI_SUCCESS != (rc = PMI_Barrier())) {
 556         OPAL_PMI_ERROR(rc, "PMI_Barrier");
 557         rc = OPAL_ERROR;
 558         goto cleanup;
 559     }
 560 
 561     opal_output_verbose(2, opal_pmix_base_framework.framework_output,
 562                         "%s pmix:s1 barrier complete",
 563                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));
 564 
 565     /* get the modex data from each local process and set the
 566      * localities to avoid having the MPI layer fetch data
 567      * for every process in the job */
 568     s1_pname.jobid = OPAL_PROC_MY_NAME.jobid;
 569     if (!got_modex_data) {
 570         got_modex_data = true;
 571         /* we only need to set locality for each local rank as "not found"
 572      * equates to "non-local" */
 573         for (i=0; i < nlranks; i++) {
 574             s1_pname.vpid = lranks[i];
 575             rc = opal_pmix_base_cache_keys_locally(&s1_pname, OPAL_PMIX_CPUSET,
 576                                                    &kp, pmix_kvs_name, pmix_vallen_max, kvs_get);
 577             if (OPAL_SUCCESS != rc) {
 578                 OPAL_ERROR_LOG(rc);
 579                 goto cleanup;
 580             }
 581             if (NULL == kp || NULL == kp->data.string) {
 582                 /* if we share a node, but we don't know anything more, then
 583          * mark us as on the node as this is all we know
 584          */
 585                 locality = OPAL_PROC_ON_CLUSTER | OPAL_PROC_ON_CU | OPAL_PROC_ON_NODE;
 586             } else {
 587                 /* determine relative location on our node */
 588                 locality = opal_hwloc_base_get_relative_locality(opal_hwloc_topology,
 589                                                                  opal_process_info.cpuset,
 590                                                                  kp->data.string);
 591             }
 592             if (NULL != kp) {
 593                 OBJ_RELEASE(kp);
 594             }
 595             OPAL_OUTPUT_VERBOSE((1, opal_pmix_base_framework.framework_output,
 596                                  "%s pmix:s1 proc %s locality %s",
 597                                  OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
 598                                  OPAL_NAME_PRINT(s1_pname),
 599                                  opal_hwloc_base_print_locality(locality)));
 600 
 601             OBJ_CONSTRUCT(&kvn, opal_value_t);
 602             kvn.key = strdup(OPAL_PMIX_LOCALITY);
 603             kvn.type = OPAL_UINT16;
 604             kvn.data.uint16 = locality;
 605             opal_pmix_base_store(&s1_pname, &kvn);
 606             OBJ_DESTRUCT(&kvn);
 607         }
 608     }
 609 
 610 cleanup:
 611     if (NULL != op->opcbfunc) {
 612         op->opcbfunc(rc, op->cbdata);
 613     }
 614     OBJ_RELEASE(op);
 615     return;
 616 }
 617 
 618 static int s1_fencenb(opal_list_t *procs, int collect_data,
 619                       opal_pmix_op_cbfunc_t cbfunc, void *cbdata)
 620 {
 621     pmi_opcaddy_t *op;
 622 
 623     /* thread-shift this so we don't block in SLURM's barrier */
 624     op = OBJ_NEW(pmi_opcaddy_t);
 625     op->opcbfunc = cbfunc;
 626     op->cbdata = cbdata;
 627     event_assign(&op->ev, opal_pmix_base.evbase, -1,
 628                  EV_WRITE, fencenb, op);
 629     event_active(&op->ev, EV_WRITE, 1);
 630 
 631     return OPAL_SUCCESS;
 632 }
 633 
 634 #define S1_WAIT_FOR_COMPLETION(a)               \
 635     do {                                        \
 636         while ((a)) {                           \
 637             usleep(10);                         \
 638         }                                       \
 639     } while (0)
 640 
 641 struct fence_result {
 642     volatile int flag;
 643     int status;
 644 };
 645 
 646 static void fence_release(int status, void *cbdata)
 647 {
 648     struct fence_result *res = (struct fence_result*)cbdata;
 649     res->status = status;
 650     opal_atomic_wmb();
 651     res->flag = 0;
 652 }
 653 
 654 static int s1_fence(opal_list_t *procs, int collect_data)
 655 {
 656     struct fence_result result = { 1, OPAL_SUCCESS };
 657     s1_fencenb(procs, collect_data, fence_release, (void*)&result);
 658     S1_WAIT_FOR_COMPLETION(result.flag);
 659     return result.status;
 660 }
 661 
 662 
 663 static int s1_get(const opal_process_name_t *id,
 664                   const char *key, opal_list_t *info,
 665                   opal_value_t **kv)
 666 {
 667     int rc;
 668     opal_output_verbose(2, opal_pmix_base_framework.framework_output,
 669                         "%s pmix:s1 called get for key %s",
 670                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME), key);
 671 
 672     rc = opal_pmix_base_cache_keys_locally(id, key, kv, pmix_kvs_name, pmix_vallen_max, kvs_get);
 673     opal_output_verbose(2, opal_pmix_base_framework.framework_output,
 674                         "%s pmix:s1 got key %s",
 675                         OPAL_NAME_PRINT(OPAL_PROC_MY_NAME), key);
 676 
 677     return rc;
 678 }
 679 
 680 static int s1_publish(opal_list_t *info)
 681 {
 682     // SLURM PMIv1 doesn't implement this function
 683 
 684     return OPAL_ERR_NOT_SUPPORTED;
 685 }
 686 
 687 static int s1_lookup(opal_list_t *data, opal_list_t *info)
 688 {
 689     // Allocate mem for port here? Otherwise we won't get success!
 690     // SLURM PMIv1 doesn't implement this function
 691 
 692     return OPAL_ERR_NOT_SUPPORTED;
 693 }
 694 
 695 static int s1_unpublish(char **keys, opal_list_t *info)
 696 {
 697     // SLURM PMIv1 doesn't implement this function
 698 
 699     return OPAL_ERR_NOT_SUPPORTED;
 700 }
 701 
 702 static int s1_job_connect(opal_list_t *procs)
 703 {
 704     return OPAL_ERR_NOT_SUPPORTED;
 705 }
 706 
 707 static int s1_job_disconnect(opal_list_t *procs)
 708 {
 709     return OPAL_ERR_NOT_SUPPORTED;
 710 }
 711 
 712 static int s1_store_local(const opal_process_name_t *proc,
 713                           opal_value_t *val)
 714 {
 715     opal_pmix_base_store(proc, val);
 716 
 717     return OPAL_SUCCESS;
 718 }
 719 
 720 static const char *s1_get_nspace(opal_jobid_t jobid)
 721 {
 722     return "N/A";
 723 }
 724 static void s1_register_jobid(opal_jobid_t jobid, const char *nspace)
 725 {
 726     return;
 727 }
 728 
 729 static char* pmix_error(int pmix_err)
 730 {
 731     char * err_msg;
 732 
 733     switch(pmix_err) {
 734     case PMI_FAIL: err_msg = "Operation failed"; break;
 735     case PMI_ERR_INIT: err_msg = "PMI is not initialized"; break;
 736     case PMI_ERR_NOMEM: err_msg = "Input buffer not large enough"; break;
 737     case PMI_ERR_INVALID_ARG: err_msg = "Invalid argument"; break;
 738     case PMI_ERR_INVALID_KEY: err_msg = "Invalid key argument"; break;
 739     case PMI_ERR_INVALID_KEY_LENGTH: err_msg = "Invalid key length argument"; break;
 740     case PMI_ERR_INVALID_VAL: err_msg = "Invalid value argument"; break;
 741     case PMI_ERR_INVALID_VAL_LENGTH: err_msg = "Invalid value length argument"; break;
 742     case PMI_ERR_INVALID_LENGTH: err_msg = "Invalid length argument"; break;
 743     case PMI_ERR_INVALID_NUM_ARGS: err_msg = "Invalid number of arguments"; break;
 744     case PMI_ERR_INVALID_ARGS: err_msg = "Invalid args argument"; break;
 745     case PMI_ERR_INVALID_NUM_PARSED: err_msg = "Invalid num_parsed length argument"; break;
 746     case PMI_ERR_INVALID_KEYVALP: err_msg = "Invalid keyvalp argument"; break;
 747     case PMI_ERR_INVALID_SIZE: err_msg = "Invalid size argument"; break;
 748 #if defined(PMI_ERR_INVALID_KVS)
 749         /* pmix.h calls this a valid return code but mpich doesn't define it (slurm does). */
 750     case PMI_ERR_INVALID_KVS: err_msg = "Invalid kvs argument"; break;
 751 #endif
 752     case PMI_SUCCESS: err_msg = "Success"; break;
 753     default: err_msg = "Unkown error";
 754     }
 755     return err_msg;
 756 }
/* [<][>][^][v][top][bottom][index][help] */
root/opal/mca/pmix/s1/pmix_s1.c

DEFINITIONS
